From 949a4748a43a19c3e58517f9decb748500f70f79 Mon Sep 17 00:00:00 2001 From: Maurice Yap Date: Mon, 3 Aug 2026 15:24:12 +0100 Subject: [PATCH 1/2] Add jittered backoff and dead-letter fields to PulsarConfig This extends `PulsarConfig` with `MaxBackoffTime`, `BackoffRandomizationFactor`, `BackoffMultiplier`, `DeadLetterTopic`, and `DeadLetterMaxAttempts` (validated `gte=2`), plus a `Validate()` method checking `MaxBackoffTime >= BackoffTime`. Note this is a pure config struct change for now - there is no consumer of these fields yet. In a future PR, I will use `cenkalti/backoff/v4` as a durect dependency to implement exponential backoff (it is currently an indirect). Signed-off-by: Maurice Yap --- internal/common/config/pulsar.go | 27 ++++++++++++++++++++++++++- 1 file changed, 26 insertions(+), 1 deletion(-) diff --git a/internal/common/config/pulsar.go b/internal/common/config/pulsar.go index 9930af0377a..f85210e7b3a 100644 --- a/internal/common/config/pulsar.go +++ b/internal/common/config/pulsar.go @@ -1,6 +1,7 @@ package config import ( + "fmt" "time" "github.com/apache/pulsar-client-go/pulsar" @@ -44,10 +45,34 @@ type PulsarConfig struct { MaxAllowedMessageSize uint // Timeout when sending messages asynchronously SendTimeout time.Duration - // Backoff from polling when Pulsar returns an error + // Initial backoff in the exponential-with-jitter backoff sequence used when polling or + // retrying fails. BackoffTime time.Duration + // Upper bound on the backoff duration between retries. If unset or <= 0, defaults to + // BackoffTime (i.e. no growth). + MaxBackoffTime time.Duration + // Fraction by which each backoff is randomised: the actual wait is drawn uniformly from + // [interval * (1 - BackoffRandomizationFactor), interval * (1 + BackoffRandomizationFactor)]. + // If unset or < 0, defaults to backoff.DefaultRandomizationFactor (0.5). + BackoffRandomizationFactor float64 + // Factor by which the backoff interval grows after each retry. If unset or <= 0, defaults to + // backoff.DefaultMultiplier (1.5). + BackoffMultiplier float64 // Number of pulsar messages that will be queued by the pulsar consumer. ReceiverQueueSize int + // The pulsar topic that messages will be published to if a sink cannot store them after DeadLetterMaxAttempts attempts + DeadLetterTopic string + // Number of consecutive Sink.Store attempts before a message is published to DeadLetterTopic and acked. + // Must be at least 2: a value of 1 would dead-letter on the first failure with no retry at all. + DeadLetterMaxAttempts int `validate:"gte=2"` +} + +// Validate checks invariants that span multiple fields and so cannot be expressed via struct tags alone. +func (c PulsarConfig) Validate() error { + if c.MaxBackoffTime > 0 && c.MaxBackoffTime < c.BackoffTime { + return fmt.Errorf("pulsar.maxBackoffTime (%s) must be >= pulsar.backoffTime (%s) if set", c.MaxBackoffTime, c.BackoffTime) + } + return nil } type TopicDelayMonitor struct { From eb56749bb0fcf63147df2e8e01be6717d0849c1b Mon Sep 17 00:00:00 2001 From: Maurice Yap Date: Tue, 4 Aug 2026 11:16:01 +0100 Subject: [PATCH 2/2] comment Signed-off-by: Maurice Yap --- internal/common/config/pulsar.go | 4 +-- internal/common/config/pulsar_test.go | 49 +++++++++++++++++++++++++++ 2 files changed, 51 insertions(+), 2 deletions(-) create mode 100644 internal/common/config/pulsar_test.go diff --git a/internal/common/config/pulsar.go b/internal/common/config/pulsar.go index f85210e7b3a..be0989f75f7 100644 --- a/internal/common/config/pulsar.go +++ b/internal/common/config/pulsar.go @@ -63,8 +63,8 @@ type PulsarConfig struct { // The pulsar topic that messages will be published to if a sink cannot store them after DeadLetterMaxAttempts attempts DeadLetterTopic string // Number of consecutive Sink.Store attempts before a message is published to DeadLetterTopic and acked. - // Must be at least 2: a value of 1 would dead-letter on the first failure with no retry at all. - DeadLetterMaxAttempts int `validate:"gte=2"` + // If set, must be at least 2: a value of 1 would dead-letter on the first failure with no retry at all. + DeadLetterMaxAttempts int `validate:"omitempty,gte=2"` } // Validate checks invariants that span multiple fields and so cannot be expressed via struct tags alone. diff --git a/internal/common/config/pulsar_test.go b/internal/common/config/pulsar_test.go new file mode 100644 index 00000000000..eac5f2b1549 --- /dev/null +++ b/internal/common/config/pulsar_test.go @@ -0,0 +1,49 @@ +package config + +import ( + "testing" + "time" + + "github.com/go-playground/validator/v10" + "github.com/stretchr/testify/assert" +) + +func validatePulsarConfig(t *testing.T, c PulsarConfig) error { + t.Helper() + validate := validator.New() + return validate.Struct(c) +} + +func TestPulsarConfig_Validate(t *testing.T) { + t.Run("DeadLetterMaxAttempts unset is valid", func(t *testing.T) { + assert.NoError(t, validatePulsarConfig(t, PulsarConfig{})) + }) + + t.Run("DeadLetterMaxAttempts of 1 is invalid", func(t *testing.T) { + assert.Error(t, validatePulsarConfig(t, PulsarConfig{DeadLetterMaxAttempts: 1})) + }) + + t.Run("DeadLetterMaxAttempts of 2 is valid", func(t *testing.T) { + assert.NoError(t, validatePulsarConfig(t, PulsarConfig{DeadLetterMaxAttempts: 2})) + }) + + t.Run("MaxBackoffTime below BackoffTime is invalid", func(t *testing.T) { + assert.Error(t, PulsarConfig{ + BackoffTime: time.Second, + MaxBackoffTime: time.Millisecond, + }.Validate()) + }) + + t.Run("MaxBackoffTime at or above BackoffTime is valid", func(t *testing.T) { + assert.NoError(t, PulsarConfig{ + BackoffTime: time.Second, + MaxBackoffTime: time.Second, + }.Validate()) + }) + + t.Run("MaxBackoffTime unset is valid", func(t *testing.T) { + assert.NoError(t, PulsarConfig{ + BackoffTime: time.Second, + }.Validate()) + }) +}