You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "geniusjoe (via GitHub)" <gi...@apache.org> on 2026/01/20 10:53:46 UTC
[PR] [Issue 1128][Producer] Regard ProducerBlockedQuotaExceededException as a retryable exception for reconnecting [pulsar-client-go]
geniusjoe opened a new pull request, #1457:
URL: https://github.com/apache/pulsar-client-go/pull/1457
Fixes https://github.com/apache/pulsar-client-go/pull/1134
Master Issue: https://github.com/apache/pulsar-client-go/pull/1128
### Motivation
Currently, the Go client handles connections in two ways:
1. If the current connection is an initialization connection, meaning when the producer is initialized for the first time, the `grabCnx()`method will not trigger retry logic. If an error occurs, it will be directly returned to the caller.
2. If the current connection is an already active connection, the `runEventsLoop()` method will detect the disconnection and call the `reconnectToBroker()` method to re-establish the connection. During reconnection, the method identifies whether the current exception is a retryable error. If it is a retryable error, the `internal.Retry()` inside the `reconnectToBroker()` method will retry indefinitely based on the backoff strategy. If it is a non-retryable error, such as the current `errMsgProducerBlockedQuotaExceededException`, the partition producer will be closed, causing any following messages sent to that partition to result in a `SendTimeoutError`.
Since the `backlogQuotaExceedException` can be resolved by modifying the TTL or increasing consumption speed, it is likely to recover after some retry attempts. I believe the exception handling for `errMsgProducerBlockedQuotaExceededException` in the `reconnectToBroker()` method should align as closely as possible with the Java implementation. Java’s current retry logic for connections is as follows:
1. If an error is received after sending a message or a heartbeat times out, the callback function `handleSendError()` is triggered to close the connection.
2. When the connection is closed, the `connectionClosed` method is called, which in turn invokes the `grabCnx()` method, and subsequently `grabCnx()` calls the `connectionOpened()` method to reconnect:
```Java
// Schedule a reconnection task
state.client.timer().newTimeout(timeout -> {
log.info("[{}] [{}] Reconnecting after timeout", state.topic, state.getHandlerName());
grabCnx(); // Re-establish the connection
}, delayMs, TimeUnit.MILLISECONDS);
```
3. If the connection still fails in the `connectionOpened()` method, the client will execute different logic based on the current exception. For the `ProducerBlockedQuotaExceededException` exception, the Java client does not treat it as an `isUnrecoverableError`. Instead, it first cleans up the pending messages and then attempts to reconnect.
### Modifications
To maintain consistency with the Java client, I suggest that when `errMsgProducerBlockedQuotaExceededException` is encountered in `reconnectToBroker()`, only `failPendingMessages()` should be executed, but the connection should not be closed. Specifically, the code should be modified as follows:
```Go
if strings.Contains(errMsg, errMsgProducerBlockedQuotaExceededException) {
p.log.Warn("Producer was blocked by quota exceed exception, failing pending messages, stop reconnecting")
p.failPendingMessages(errors.Join(ErrProducerBlockedQuotaExceeded, err))
// Do not close the connection here
// return struct{}{}, err
}
```
### Verifying this change
- [x] Make sure that the change passes the CI checks.
This change added tests and can be verified as follows:
`TestProducerReconnectWhenBacklogQuotaExceed`
### Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API: (no)
- The schema: (no)
- The default values of configurations: (no)
- The wire protocol: (no)
### Documentation
- Does this pull request introduce a new feature? (no)
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Issue 1128][Producer] Regard ProducerBlockedQuotaExceededException as a retryable exception for reconnecting [pulsar-client-go]
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie commented on code in PR #1457:
URL: https://github.com/apache/pulsar-client-go/pull/1457#discussion_r2745104813
##########
pulsar/producer_test.go:
##########
@@ -2952,3 +2953,135 @@ 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(512*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
+ _consumer, err := client.Subscribe(ConsumerOptions{
+ Topic: topicName,
+ SubscriptionName: "my-sub",
+ Type: Exclusive,
+ ReceiverQueueSize: 1,
+ SubscriptionInitialPosition: SubscriptionPositionEarliest,
+ })
+ assert.Nil(t, err)
+ defer _consumer.Close()
+
+ // Step 3: Create producer with custom backoff policy to reduce retry interval
+ bo := newTestReconnectBackoffPolicy(100*time.Millisecond, 1*time.Second)
+ _producer, err := client.CreateProducer(ProducerOptions{
+ Topic: topicName,
+ DisableBatching: true,
+ SendTimeout: 5 * time.Minute,
+ BackOffPolicyFunc: func() backoff.Policy {
+ return bo
+ },
+ })
+ assert.NoError(t, err)
+ defer _producer.Close()
+
+ // Step 4: Start goroutine to continuously send 512KB messages and monitor statistics
+ go func() {
+ for {
+ // Producer sends 512KB message every 10 seconds
+ _producer.SendAsync(context.Background(), &ProducerMessage{
+ Payload: make([]byte, 512*1024),
+ }, func(msgId MessageID, _ *ProducerMessage, err error) {
+ if err != nil {
+ logger.Error("sendAsync fail", "time", time.Now().String(), "err", err.Error())
+ return
+ }
+ logger.Info("sendAsync success", "msgId", msgId.String(), "time", time.Now().String())
+ })
+
+ // Get topic statistics for debugging
+ stats, err := admin.Topics().GetPartitionedStats(*tn, false)
+ assert.NoError(t, err)
+ logger.Info("current stats", "stats", stats)
+ time.Sleep(10 * time.Second)
+ }
+ }()
Review Comment:
I think we can simply send some messages to exceed the backlog, without using a dedicated goroutine.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Issue 1128][Producer] Regard ProducerBlockedQuotaExceededException as a retryable exception for reconnecting [pulsar-client-go]
Posted by "geniusjoe (via GitHub)" <gi...@apache.org>.
geniusjoe commented on code in PR #1457:
URL: https://github.com/apache/pulsar-client-go/pull/1457#discussion_r2746015675
##########
pulsar/producer_test.go:
##########
@@ -2952,3 +2953,135 @@ 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
Review Comment:
The current test scenario does not use this function, so the implementation of this member function is skipped.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Issue 1128][Producer] Regard ProducerBlockedQuotaExceededException as a retryable exception for reconnecting [pulsar-client-go]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
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/golangci-lint@v1.61.0
+ GOBIN=$(shell pwd)/bin go install github.com/golangci/golangci-lint/cmd/golangci-lint@v1.64.2
# 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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Issue 1128][Producer] Regard ProducerBlockedQuotaExceededException as a retryable exception for reconnecting [pulsar-client-go]
Posted by "geniusjoe (via GitHub)" <gi...@apache.org>.
geniusjoe commented on code in PR #1457:
URL: https://github.com/apache/pulsar-client-go/pull/1457#discussion_r2773351698
##########
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/golangci-lint@v1.61.0
+ GOBIN=$(shell pwd)/bin go install github.com/golangci/golangci-lint/cmd/golangci-lint@v1.64.2
# 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:
Added this to the PR description and explained why it's included.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Issue 1128][Producer] Regard ProducerBlockedQuotaExceededException as a retryable exception for reconnecting [pulsar-client-go]
Posted by "geniusjoe (via GitHub)" <gi...@apache.org>.
geniusjoe commented on code in PR #1457:
URL: https://github.com/apache/pulsar-client-go/pull/1457#discussion_r2773359677
##########
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:
Fixed
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Issue 1128][Producer] Regard ProducerBlockedQuotaExceededException as a retryable exception for reconnecting [pulsar-client-go]
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie commented on code in PR #1457:
URL: https://github.com/apache/pulsar-client-go/pull/1457#discussion_r2773218084
##########
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
+ _consumer, err := client.Subscribe(ConsumerOptions{
+ Topic: topicName,
+ SubscriptionName: "my-sub",
+ Type: Exclusive,
+ ReceiverQueueSize: 1,
+ SubscriptionInitialPosition: SubscriptionPositionEarliest,
+ })
+ assert.Nil(t, err)
+ defer _consumer.Close()
+
+ // Step 3: Create producer with custom backoff policy to reduce retry interval
+ bo := newTestReconnectBackoffPolicy(100*time.Millisecond, 1*time.Second)
+ _producer, err := client.CreateProducer(ProducerOptions{
+ Topic: topicName,
+ DisableBatching: true,
+ SendTimeout: 5 * time.Minute,
+ BackOffPolicyFunc: func() backoff.Policy {
+ return bo
+ },
+ })
+ assert.NoError(t, err)
+ defer _producer.Close()
+
+ // Step 4: Send 512KB messages and monitor statistics
+ // Limit to 10 iterations to avoid infinite loop in test
+ isReachMaxBackoff := false
+ for i := 0; i < 10; i++ {
Review Comment:
Why do we need 10 iterations? This seems to be consistently reproducible. Perhaps we don’t need the for loop and `time.Sleep`.
##########
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:
```suggestion
// Step 1: Create namespace and configure 10KB backlog quota with producer_exception policy
```
It should be 10KB according to the code? Or maybe need to change the line 3018.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Issue 1128][Producer] Regard ProducerBlockedQuotaExceededException as a retryable exception for reconnecting [pulsar-client-go]
Posted by "geniusjoe (via GitHub)" <gi...@apache.org>.
geniusjoe commented on code in PR #1457:
URL: https://github.com/apache/pulsar-client-go/pull/1457#discussion_r2746182284
##########
pulsar/producer_test.go:
##########
@@ -2952,3 +2953,135 @@ 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(512*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
+ _consumer, err := client.Subscribe(ConsumerOptions{
+ Topic: topicName,
+ SubscriptionName: "my-sub",
+ Type: Exclusive,
+ ReceiverQueueSize: 1,
+ SubscriptionInitialPosition: SubscriptionPositionEarliest,
+ })
+ assert.Nil(t, err)
+ defer _consumer.Close()
+
+ // Step 3: Create producer with custom backoff policy to reduce retry interval
+ bo := newTestReconnectBackoffPolicy(100*time.Millisecond, 1*time.Second)
+ _producer, err := client.CreateProducer(ProducerOptions{
+ Topic: topicName,
+ DisableBatching: true,
+ SendTimeout: 5 * time.Minute,
+ BackOffPolicyFunc: func() backoff.Policy {
+ return bo
+ },
+ })
+ assert.NoError(t, err)
+ defer _producer.Close()
+
+ // Step 4: Start goroutine to continuously send 512KB messages and monitor statistics
+ go func() {
+ for {
+ // Producer sends 512KB message every 10 seconds
+ _producer.SendAsync(context.Background(), &ProducerMessage{
+ Payload: make([]byte, 512*1024),
+ }, func(msgId MessageID, _ *ProducerMessage, err error) {
+ if err != nil {
+ logger.Error("sendAsync fail", "time", time.Now().String(), "err", err.Error())
+ return
+ }
+ logger.Info("sendAsync success", "msgId", msgId.String(), "time", time.Now().String())
+ })
+
+ // Get topic statistics for debugging
+ stats, err := admin.Topics().GetPartitionedStats(*tn, false)
+ assert.NoError(t, err)
Review Comment:
The goroutine has been removed, and the test scenario is completed using the same main thread.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Issue 1128][Producer] Regard ProducerBlockedQuotaExceededException as a retryable exception for reconnecting [pulsar-client-go]
Posted by "geniusjoe (via GitHub)" <gi...@apache.org>.
geniusjoe commented on code in PR #1457:
URL: https://github.com/apache/pulsar-client-go/pull/1457#discussion_r2746046445
##########
pulsar/producer_test.go:
##########
@@ -2952,3 +2953,135 @@ 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() {
Review Comment:
The current test scenario does not use this function, so the implementation of this member function is skipped.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Issue 1128][Producer] Regard ProducerBlockedQuotaExceededException as a retryable exception for reconnecting [pulsar-client-go]
Posted by "geniusjoe (via GitHub)" <gi...@apache.org>.
geniusjoe commented on code in PR #1457:
URL: https://github.com/apache/pulsar-client-go/pull/1457#discussion_r2746025118
##########
pulsar/producer_partition.go:
##########
@@ -523,9 +523,10 @@ func (p *partitionProducer) reconnectToBroker(connectionClosed *connectionClosed
}
if strings.Contains(errMsg, errMsgProducerBlockedQuotaExceededException) {
+ // ProducerBlockedQuotaExceededException is a retryable exception,
+ // we only fail pending messages but continue trying to reconnect
p.log.Warn("Producer was blocked by quota exceed exception, failing pending messages, stop reconnecting")
Review Comment:
Fixed
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Issue 1128][Producer] Regard ProducerBlockedQuotaExceededException as a retryable exception for reconnecting [pulsar-client-go]
Posted by "geniusjoe (via GitHub)" <gi...@apache.org>.
geniusjoe commented on code in PR #1457:
URL: https://github.com/apache/pulsar-client-go/pull/1457#discussion_r2773361271
##########
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:
Change comment to 10KB
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Issue 1128][Producer] Regard ProducerBlockedQuotaExceededException as a retryable exception for reconnecting [pulsar-client-go]
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie merged PR #1457:
URL: https://github.com/apache/pulsar-client-go/pull/1457
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Issue 1128][Producer] Regard ProducerBlockedQuotaExceededException as a retryable exception for reconnecting [pulsar-client-go]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #1457:
URL: https://github.com/apache/pulsar-client-go/pull/1457#discussion_r2745067307
##########
pulsar/producer_partition.go:
##########
@@ -523,9 +523,10 @@ func (p *partitionProducer) reconnectToBroker(connectionClosed *connectionClosed
}
if strings.Contains(errMsg, errMsgProducerBlockedQuotaExceededException) {
+ // ProducerBlockedQuotaExceededException is a retryable exception,
+ // we only fail pending messages but continue trying to reconnect
p.log.Warn("Producer was blocked by quota exceed exception, failing pending messages, stop reconnecting")
Review Comment:
The log message states "stop reconnecting" but the code change actually allows reconnection to continue (which is the intent of this PR). The log message should be updated to accurately reflect that reconnection will continue, for example: "Producer was blocked by quota exceed exception, failing pending messages, will retry reconnecting"
##########
pulsar/producer_test.go:
##########
@@ -2952,3 +2953,135 @@ 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
Review Comment:
The IsExpectedIntervalFrom method signature is incorrect. It should accept a time.Time parameter like the testBackoffPolicy in reader_test.go (line 862). The current implementation returns a constant true value and doesn't validate timing at all, making it useless for testing.
```suggestion
func (b *testReconnectBackoffPolicy) IsExpectedIntervalFrom(start time.Time) bool {
// Verify that the elapsed time since 'start' is close to the current backoff interval.
b.lock.Lock()
expected := b.curBackoff
b.lock.Unlock()
// If no backoff is expected yet, do not enforce any interval.
if expected <= 0 {
return true
}
elapsed := time.Since(start)
// Allow some tolerance around the expected backoff interval (50%–150%).
lowerBound := expected / 2
upperBound := expected + expected/2
return elapsed >= lowerBound && elapsed <= upperBound
```
##########
pulsar/producer_test.go:
##########
@@ -2952,3 +2953,135 @@ 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() {
Review Comment:
The Reset method is empty but should reset curBackoff to 0 to match the interface contract and the behavior of DefaultBackoff. While this may not affect the test since Reset isn't called in the test scenario, it makes the implementation incomplete and could lead to unexpected behavior if the backoff policy is reused.
```suggestion
func (b *testReconnectBackoffPolicy) Reset() {
b.lock.Lock()
defer b.lock.Unlock()
b.curBackoff = 0
b.retryTime = 0
```
##########
pulsar/producer_test.go:
##########
@@ -2952,3 +2953,135 @@ 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(512*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
+ _consumer, err := client.Subscribe(ConsumerOptions{
+ Topic: topicName,
+ SubscriptionName: "my-sub",
+ Type: Exclusive,
+ ReceiverQueueSize: 1,
+ SubscriptionInitialPosition: SubscriptionPositionEarliest,
+ })
+ assert.Nil(t, err)
+ defer _consumer.Close()
+
+ // Step 3: Create producer with custom backoff policy to reduce retry interval
+ bo := newTestReconnectBackoffPolicy(100*time.Millisecond, 1*time.Second)
+ _producer, err := client.CreateProducer(ProducerOptions{
+ Topic: topicName,
+ DisableBatching: true,
+ SendTimeout: 5 * time.Minute,
+ BackOffPolicyFunc: func() backoff.Policy {
+ return bo
+ },
+ })
+ assert.NoError(t, err)
+ defer _producer.Close()
+
+ // Step 4: Start goroutine to continuously send 512KB messages and monitor statistics
+ go func() {
+ for {
+ // Producer sends 512KB message every 10 seconds
+ _producer.SendAsync(context.Background(), &ProducerMessage{
+ Payload: make([]byte, 512*1024),
+ }, func(msgId MessageID, _ *ProducerMessage, err error) {
+ if err != nil {
+ logger.Error("sendAsync fail", "time", time.Now().String(), "err", err.Error())
+ return
+ }
+ logger.Info("sendAsync success", "msgId", msgId.String(), "time", time.Now().String())
+ })
+
+ // Get topic statistics for debugging
+ stats, err := admin.Topics().GetPartitionedStats(*tn, false)
+ assert.NoError(t, err)
Review Comment:
The goroutine calls assert.NoError which doesn't properly fail the test when called from a goroutine. The testing.T methods are not safe to call from goroutines other than the one running the test. Use require.NoError with proper error handling or send errors through a channel to be checked in the main test goroutine.
##########
pulsar/producer_test.go:
##########
@@ -2952,3 +2953,135 @@ 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(512*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
+ _consumer, err := client.Subscribe(ConsumerOptions{
+ Topic: topicName,
+ SubscriptionName: "my-sub",
+ Type: Exclusive,
+ ReceiverQueueSize: 1,
+ SubscriptionInitialPosition: SubscriptionPositionEarliest,
+ })
+ assert.Nil(t, err)
+ defer _consumer.Close()
+
+ // Step 3: Create producer with custom backoff policy to reduce retry interval
+ bo := newTestReconnectBackoffPolicy(100*time.Millisecond, 1*time.Second)
+ _producer, err := client.CreateProducer(ProducerOptions{
+ Topic: topicName,
+ DisableBatching: true,
+ SendTimeout: 5 * time.Minute,
+ BackOffPolicyFunc: func() backoff.Policy {
+ return bo
+ },
+ })
+ assert.NoError(t, err)
+ defer _producer.Close()
+
+ // Step 4: Start goroutine to continuously send 512KB messages and monitor statistics
+ go func() {
+ for {
+ // Producer sends 512KB message every 10 seconds
+ _producer.SendAsync(context.Background(), &ProducerMessage{
+ Payload: make([]byte, 512*1024),
+ }, func(msgId MessageID, _ *ProducerMessage, err error) {
+ if err != nil {
+ logger.Error("sendAsync fail", "time", time.Now().String(), "err", err.Error())
+ return
+ }
+ logger.Info("sendAsync success", "msgId", msgId.String(), "time", time.Now().String())
+ })
+
+ // Get topic statistics for debugging
+ stats, err := admin.Topics().GetPartitionedStats(*tn, false)
+ assert.NoError(t, err)
+ logger.Info("current stats", "stats", stats)
+ time.Sleep(10 * time.Second)
+ }
+ }()
Review Comment:
The goroutine started at line 3061 runs an infinite loop and is never stopped. This creates a goroutine leak that will continue running even after the test completes. The goroutine should be controlled with a context, channel, or some other mechanism to ensure it stops when the test ends. Additionally, the infinite loop can continue sending messages after the test assertion passes, potentially interfering with other tests.
##########
pulsar/producer_test.go:
##########
@@ -2952,3 +2953,135 @@ 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
Review Comment:
The backoff calculation logic is incorrect. When curBackoff is 0, adding it to itself (curBackoff += curBackoff) keeps it at 0, so it never increases. This means the backoff will always stay at minBackoff and never reach maxBackoff. The correct implementation should use multiplication (e.g., curBackoff *= 2) or match the DefaultBackoff implementation from backoff.go which uses curBackoff += curBackoff but initializes to a non-zero value or handles the zero case differently.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Issue 1128][Producer] Regard ProducerBlockedQuotaExceededException as a retryable exception for reconnecting [pulsar-client-go]
Posted by "geniusjoe (via GitHub)" <gi...@apache.org>.
geniusjoe commented on code in PR #1457:
URL: https://github.com/apache/pulsar-client-go/pull/1457#discussion_r2773372852
##########
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
+ _consumer, err := client.Subscribe(ConsumerOptions{
+ Topic: topicName,
+ SubscriptionName: "my-sub",
+ Type: Exclusive,
+ ReceiverQueueSize: 1,
+ SubscriptionInitialPosition: SubscriptionPositionEarliest,
+ })
+ assert.Nil(t, err)
+ defer _consumer.Close()
+
+ // Step 3: Create producer with custom backoff policy to reduce retry interval
+ bo := newTestReconnectBackoffPolicy(100*time.Millisecond, 1*time.Second)
+ _producer, err := client.CreateProducer(ProducerOptions{
+ Topic: topicName,
+ DisableBatching: true,
+ SendTimeout: 5 * time.Minute,
+ BackOffPolicyFunc: func() backoff.Policy {
+ return bo
+ },
+ })
+ assert.NoError(t, err)
+ defer _producer.Close()
+
+ // Step 4: Send 512KB messages and monitor statistics
+ // Limit to 10 iterations to avoid infinite loop in test
+ isReachMaxBackoff := false
+ for i := 0; i < 10; i++ {
Review Comment:
@RobertIndie
The broker updates the backlog quota based on metrics calculations. Since metrics are collected approximately every 30 seconds, sending a single message larger than 10KB will not immediately trigger backlog quota throttling. Therefore, a for loop is needed to send multiple messages repeatedly, waiting for the backlog to reach the quota limit.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Issue 1128][Producer] Regard ProducerBlockedQuotaExceededException as a retryable exception for reconnecting [pulsar-client-go]
Posted by "geniusjoe (via GitHub)" <gi...@apache.org>.
geniusjoe commented on code in PR #1457:
URL: https://github.com/apache/pulsar-client-go/pull/1457#discussion_r2773337207
##########
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:
Fixed
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Issue 1128][Producer] Regard ProducerBlockedQuotaExceededException as a retryable exception for reconnecting [pulsar-client-go]
Posted by "geniusjoe (via GitHub)" <gi...@apache.org>.
geniusjoe commented on code in PR #1457:
URL: https://github.com/apache/pulsar-client-go/pull/1457#discussion_r2746042373
##########
pulsar/producer_test.go:
##########
@@ -2952,3 +2953,135 @@ 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
Review Comment:
```
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++
```
There is no issue with the code here. When retrying for the first time, `b.curBackoff` is assigned the value of `b.minBackoff`. And `b.minBackoff` should never be 0.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Issue 1128][Producer] Regard ProducerBlockedQuotaExceededException as a retryable exception for reconnecting [pulsar-client-go]
Posted by "geniusjoe (via GitHub)" <gi...@apache.org>.
geniusjoe commented on code in PR #1457:
URL: https://github.com/apache/pulsar-client-go/pull/1457#discussion_r2746183063
##########
pulsar/producer_test.go:
##########
@@ -2952,3 +2953,135 @@ 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(512*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
+ _consumer, err := client.Subscribe(ConsumerOptions{
+ Topic: topicName,
+ SubscriptionName: "my-sub",
+ Type: Exclusive,
+ ReceiverQueueSize: 1,
+ SubscriptionInitialPosition: SubscriptionPositionEarliest,
+ })
+ assert.Nil(t, err)
+ defer _consumer.Close()
+
+ // Step 3: Create producer with custom backoff policy to reduce retry interval
+ bo := newTestReconnectBackoffPolicy(100*time.Millisecond, 1*time.Second)
+ _producer, err := client.CreateProducer(ProducerOptions{
+ Topic: topicName,
+ DisableBatching: true,
+ SendTimeout: 5 * time.Minute,
+ BackOffPolicyFunc: func() backoff.Policy {
+ return bo
+ },
+ })
+ assert.NoError(t, err)
+ defer _producer.Close()
+
+ // Step 4: Start goroutine to continuously send 512KB messages and monitor statistics
+ go func() {
+ for {
+ // Producer sends 512KB message every 10 seconds
+ _producer.SendAsync(context.Background(), &ProducerMessage{
+ Payload: make([]byte, 512*1024),
+ }, func(msgId MessageID, _ *ProducerMessage, err error) {
+ if err != nil {
+ logger.Error("sendAsync fail", "time", time.Now().String(), "err", err.Error())
+ return
+ }
+ logger.Info("sendAsync success", "msgId", msgId.String(), "time", time.Now().String())
+ })
+
+ // Get topic statistics for debugging
+ stats, err := admin.Topics().GetPartitionedStats(*tn, false)
+ assert.NoError(t, err)
+ logger.Info("current stats", "stats", stats)
+ time.Sleep(10 * time.Second)
+ }
+ }()
Review Comment:
The goroutine has been removed, and the test scenario is completed using the same main thread.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Issue 1128][Producer] Regard ProducerBlockedQuotaExceededException as a retryable exception for reconnecting [pulsar-client-go]
Posted by "geniusjoe (via GitHub)" <gi...@apache.org>.
geniusjoe commented on code in PR #1457:
URL: https://github.com/apache/pulsar-client-go/pull/1457#discussion_r2746183929
##########
pulsar/producer_test.go:
##########
@@ -2952,3 +2953,135 @@ 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(512*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
+ _consumer, err := client.Subscribe(ConsumerOptions{
+ Topic: topicName,
+ SubscriptionName: "my-sub",
+ Type: Exclusive,
+ ReceiverQueueSize: 1,
+ SubscriptionInitialPosition: SubscriptionPositionEarliest,
+ })
+ assert.Nil(t, err)
+ defer _consumer.Close()
+
+ // Step 3: Create producer with custom backoff policy to reduce retry interval
+ bo := newTestReconnectBackoffPolicy(100*time.Millisecond, 1*time.Second)
+ _producer, err := client.CreateProducer(ProducerOptions{
+ Topic: topicName,
+ DisableBatching: true,
+ SendTimeout: 5 * time.Minute,
+ BackOffPolicyFunc: func() backoff.Policy {
+ return bo
+ },
+ })
+ assert.NoError(t, err)
+ defer _producer.Close()
+
+ // Step 4: Start goroutine to continuously send 512KB messages and monitor statistics
+ go func() {
+ for {
+ // Producer sends 512KB message every 10 seconds
+ _producer.SendAsync(context.Background(), &ProducerMessage{
+ Payload: make([]byte, 512*1024),
+ }, func(msgId MessageID, _ *ProducerMessage, err error) {
+ if err != nil {
+ logger.Error("sendAsync fail", "time", time.Now().String(), "err", err.Error())
+ return
+ }
+ logger.Info("sendAsync success", "msgId", msgId.String(), "time", time.Now().String())
+ })
+
+ // Get topic statistics for debugging
+ stats, err := admin.Topics().GetPartitionedStats(*tn, false)
+ assert.NoError(t, err)
+ logger.Info("current stats", "stats", stats)
+ time.Sleep(10 * time.Second)
+ }
+ }()
Review Comment:
Fixed. The goroutine has been removed, and the test scenario is completed using the same main thread.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org