Add jittered backoff and dead-letter fields to PulsarConfig

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.
+	// 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.
+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 {

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())
+	})
+}