Enable dead-letter topics and jittered backoff for all three ingesters

_local/eventingester/config.yaml

@@ -7,6 +7,12 @@ pulsar:
   URL: "pulsar://localhost:6650"
   restURL: "http://localhost:8090"
   jobsetEventsTopic: "events"
+  backoffTime: 1s
+  maxBackoffTime: 5s
+  backoffRandomizationFactor: 0.5
+  backoffMultiplier: 1.5
+  deadLetterTopic: events-dead-letter
+  deadLetterMaxAttempts: 5
 subscriptionName: "events-ingester"
 metricsPort: 9004
 observability:

_local/lookoutingester/config.yaml

@@ -18,3 +18,9 @@ postgres:
 pulsar:
   URL: "pulsar://localhost:6650"
   restURL: "http://localhost:8090"
+  backoffTime: 1s
+  maxBackoffTime: 5s
+  backoffRandomizationFactor: 0.5
+  backoffMultiplier: 1.5
+  deadLetterTopic: events-dead-letter
+  deadLetterMaxAttempts: 5

_local/scheduleringester/config.yaml

@@ -18,3 +18,9 @@ postgres:
 pulsar:
   URL: "pulsar://localhost:6650"
   restURL: "http://localhost:8090"
+  backoffTime: 1s
+  maxBackoffTime: 5s
+  backoffRandomizationFactor: 0.5
+  backoffMultiplier: 1.5
+  deadLetterTopic: scheduler-events-dead-letter
+  deadLetterMaxAttempts: 5

config/eventingester/config.yaml

@@ -9,7 +9,12 @@ pulsar:
   restURL: "http://localhost:8090"
   jobsetEventsTopic: events
   backoffTime: 1s
+  maxBackoffTime: 30s
+  backoffRandomizationFactor: 0.5
+  backoffMultiplier: 1.5
   receiverQueueSize: 100
+  deadLetterTopic: events-dead-letter
+  deadLetterMaxAttempts: 5
   delayMonitor:
     enabled: false
     interval: 30s

config/lookoutingester/config.yaml

@@ -12,7 +12,12 @@ pulsar:
   restURL: "http://localhost:8090"
   jobsetEventsTopic: "events"
   backoffTime: 1s
+  maxBackoffTime: 30s
+  backoffRandomizationFactor: 0.5
+  backoffMultiplier: 1.5
   receiverQueueSize: 100
+  deadLetterTopic: events-dead-letter
+  deadLetterMaxAttempts: 5
   delayMonitor:
     enabled: false
     interval: 30s
@@ -21,4 +26,3 @@ batchSize: 10000
 batchDuration: 500ms
 minJobSpecCompressionSize: 1024
 userAnnotationPrefix: "armadaproject.io/"
-maxBackoff: 60

config/scheduleringester/config.yaml

@@ -14,7 +14,12 @@ pulsar:
   jobsetEventsTopic: "events"
   controlPlaneEventsTopic: "control-plane"
   backoffTime: 1s
+  maxBackoffTime: 30s
+  backoffRandomizationFactor: 0.5
+  backoffMultiplier: 1.5
   receiverQueueSize: 100
+  deadLetterTopic: scheduler-events-dead-letter
+  deadLetterMaxAttempts: 5
   delayMonitor:
     enabled: false
     interval: 30s

internal/eventingester/configuration/validation.go

@@ -1,14 +1,16 @@
 package configuration
 
 import (
+	"errors"
+
 	"github.com/go-playground/validator/v10"
 
 	commonconfig "github.com/armadaproject/armada/internal/common/config"
 )
 
 func (c EventIngesterConfiguration) Validate() error {
 	validate := validator.New()
-	return validate.Struct(c)
+	return errors.Join(validate.Struct(c), c.Pulsar.Validate())
 }
 
 func (c *EventIngesterConfiguration) Mutate() (commonconfig.Config, error) {

internal/lookoutingester/configuration/validation.go

@@ -1,14 +1,16 @@
 package configuration
 
 import (
+	"errors"
+
 	"github.com/go-playground/validator/v10"
 
 	commonconfig "github.com/armadaproject/armada/internal/common/config"
 )
 
 func (c LookoutIngesterConfiguration) Validate() error {
 	validate := validator.New()
-	return validate.Struct(c)
+	return errors.Join(validate.Struct(c), c.Pulsar.Validate())
 }
 
 func (c *LookoutIngesterConfiguration) Mutate() (commonconfig.Config, error) {

internal/scheduleringester/config.go

@@ -1,6 +1,7 @@
 package scheduleringester
 
 import (
+	"errors"
 	"time"
 
 	"github.com/go-playground/validator/v10"
@@ -37,5 +38,5 @@ func (c *Configuration) Mutate() (commonconfig.Config, error) {
 
 func (c Configuration) Validate() error {
 	validate := validator.New()
-	return validate.Struct(c)
+	return errors.Join(validate.Struct(c), c.Pulsar.Validate())
 }