Copilot commented on code in PR #1457:
URL: https://github.com/apache/pulsar-client-go/pull/1457#discussion_r2773213823
##########
pulsar/producer_test.go:
##########
@@ -2952,3 +2953,136 @@ func TestPartitionUpdateFailed(t *testing.T) {
time.Sleep(time.Second * 1)
}
}
+
+type testReconnectBackoffPolicy struct {
+ curBackoff, minBackoff, maxBackoff time.Duration
+ retryTime int
+ lock sync.Mutex
+}
+
+func newTestReconnectBackoffPolicy(minBackoff, maxBackoff time.Duration)
*testReconnectBackoffPolicy {
+ return &testReconnectBackoffPolicy{
+ curBackoff: 0,
+ minBackoff: minBackoff,
+ maxBackoff: maxBackoff,
+ }
+}
+
+func (b *testReconnectBackoffPolicy) Next() time.Duration {
+ b.lock.Lock()
+ defer b.lock.Unlock()
+
+ // Double the delay each time
+ b.curBackoff += b.curBackoff
+ if b.curBackoff.Nanoseconds() < b.minBackoff.Nanoseconds() {
+ b.curBackoff = b.minBackoff
+ } else if b.curBackoff.Nanoseconds() > b.maxBackoff.Nanoseconds() {
+ b.curBackoff = b.maxBackoff
+ }
+ b.retryTime++
+ return b.curBackoff
+}
+func (b *testReconnectBackoffPolicy) IsMaxBackoffReached() bool {
+ b.lock.Lock()
+ defer b.lock.Unlock()
+ return b.curBackoff >= b.maxBackoff
+}
+
+func (b *testReconnectBackoffPolicy) Reset() {
+}
+
+func (b *testReconnectBackoffPolicy) IsExpectedIntervalFrom() bool {
+ return true
+}
+
+func TestProducerReconnectWhenBacklogQuotaExceed(t *testing.T) {
+ logger := slog.New(slog.NewJSONHandler(os.Stdout,
&slog.HandlerOptions{Level: slog.LevelInfo}))
+ slog.SetDefault(logger)
+ client, err := NewClient(ClientOptions{
+ URL: serviceURL,
+ Logger: plog.NewLoggerWithSlog(logger),
+ })
+ defer client.Close()
+ namespace := "public/" + generateRandomName()
+ assert.NoError(t, err)
+ admin, err := pulsaradmin.NewClient(&config.Config{
+ WebServiceURL: adminURL,
+ })
+ assert.NoError(t, err)
+ // Step 1: Create namespace and configure 512KB backlog quota with
producer_exception policy
+ // When subscription backlog stats refresh and reach the limit,
producer will encounter BlockQuotaExceed exception
+ err = admin.Namespaces().CreateNamespace(namespace)
+ assert.NoError(t, err)
+ err = admin.Namespaces().SetBacklogQuota(
+ namespace,
+ utils.NewBacklogQuota(10*1024, -1, utils.ProducerException),
+ utils.DestinationStorage,
+ )
+ assert.NoError(t, err)
+
+ // Verify backlog quota configuration
+ quotaMap, err := admin.Namespaces().GetBacklogQuotaMap(namespace)
+ assert.NoError(t, err)
+ logger.Info(fmt.Sprintf("quotaMap = %v", quotaMap))
+
+ // Create test topic
+ topicName := namespace + "/test-topic"
+ tn, err := utils.GetTopicName(topicName)
+ assert.NoError(t, err)
+ err = admin.Topics().Create(*tn, 1)
+ assert.NoError(t, err)
+
+ // Step 2: Create consumer with small receiver queue size and earliest
subscription position
+ // This ensures that once 512KB message is sent, producer will quickly
reach backlog quota limit
Review Comment:
The comment says "once 512KB message is sent" but the actual message size
sent in line 3065 is 512KB, while the backlog quota is only 10KB. This comment
should clarify that the message size (512KB) greatly exceeds the quota (10KB)
to ensure the quota is quickly reached.
```suggestion
// This ensures that by sending a 512KB message (much larger than the
10KB backlog quota), the producer will quickly reach the backlog quota limit
```
##########
Makefile:
##########
@@ -44,13 +44,13 @@ lint: bin/golangci-lint
bin/golangci-lint run
bin/golangci-lint:
- GOBIN=$(shell pwd)/bin go install
github.com/golangci/golangci-lint/cmd/[email protected]
+ GOBIN=$(shell pwd)/bin go install
github.com/golangci/golangci-lint/cmd/[email protected]
# an alternative to above `make lint` command
# use golangCi-lint docker to avoid local golang env issues
# https://golangci-lint.run/welcome/install/
lint-docker:
- docker run --rm -v $(shell pwd):/app -w /app
golangci/golangci-lint:v1.61.0 golangci-lint run -v
+ docker run --rm -v $(shell pwd):/app -w /app
golangci/golangci-lint:v1.64.2 golangci-lint run -v
Review Comment:
The golangci-lint version update from v1.61.0 to v1.64.2 appears unrelated
to the main purpose of this PR (handling ProducerBlockedQuotaExceededException
as retryable). This change is not mentioned in the PR description. Consider
either removing this change from this PR and creating a separate PR for the
dependency update, or adding it to the PR description to explain why it's
included.
##########
pulsar/producer_test.go:
##########
@@ -2952,3 +2953,136 @@ func TestPartitionUpdateFailed(t *testing.T) {
time.Sleep(time.Second * 1)
}
}
+
+type testReconnectBackoffPolicy struct {
+ curBackoff, minBackoff, maxBackoff time.Duration
+ retryTime int
+ lock sync.Mutex
+}
+
+func newTestReconnectBackoffPolicy(minBackoff, maxBackoff time.Duration)
*testReconnectBackoffPolicy {
+ return &testReconnectBackoffPolicy{
+ curBackoff: 0,
+ minBackoff: minBackoff,
+ maxBackoff: maxBackoff,
+ }
+}
+
+func (b *testReconnectBackoffPolicy) Next() time.Duration {
+ b.lock.Lock()
+ defer b.lock.Unlock()
+
+ // Double the delay each time
+ b.curBackoff += b.curBackoff
+ if b.curBackoff.Nanoseconds() < b.minBackoff.Nanoseconds() {
+ b.curBackoff = b.minBackoff
+ } else if b.curBackoff.Nanoseconds() > b.maxBackoff.Nanoseconds() {
+ b.curBackoff = b.maxBackoff
+ }
+ b.retryTime++
+ return b.curBackoff
+}
+func (b *testReconnectBackoffPolicy) IsMaxBackoffReached() bool {
+ b.lock.Lock()
+ defer b.lock.Unlock()
+ return b.curBackoff >= b.maxBackoff
+}
+
+func (b *testReconnectBackoffPolicy) Reset() {
+}
+
+func (b *testReconnectBackoffPolicy) IsExpectedIntervalFrom() bool {
+ return true
+}
+
+func TestProducerReconnectWhenBacklogQuotaExceed(t *testing.T) {
+ logger := slog.New(slog.NewJSONHandler(os.Stdout,
&slog.HandlerOptions{Level: slog.LevelInfo}))
+ slog.SetDefault(logger)
+ client, err := NewClient(ClientOptions{
+ URL: serviceURL,
+ Logger: plog.NewLoggerWithSlog(logger),
+ })
+ defer client.Close()
+ namespace := "public/" + generateRandomName()
+ assert.NoError(t, err)
+ admin, err := pulsaradmin.NewClient(&config.Config{
+ WebServiceURL: adminURL,
+ })
+ assert.NoError(t, err)
+ // Step 1: Create namespace and configure 512KB backlog quota with
producer_exception policy
Review Comment:
The comment says "512KB backlog quota" but the actual quota set in line 3018
is 10KB (10*1024). The comment should be updated to match the actual quota
value of 10KB, or the quota value should be changed to match the comment.
```suggestion
// Step 1: Create namespace and configure 10KB backlog quota with
producer_exception policy
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
