diff --git a/internal/common/pulsarutils/dlq.go b/internal/common/pulsarutils/dlq.go new file mode 100644 index 00000000000..b6514d879ce --- /dev/null +++ b/internal/common/pulsarutils/dlq.go @@ -0,0 +1,15 @@ +package pulsarutils + +import ( + "github.com/apache/pulsar-client-go/pulsar" +) + +// MessageIdsToStrings converts Pulsar message ids to their string representation, +// suitable for embedding in JSON payloads or message properties. +func MessageIdsToStrings(ids []pulsar.MessageID) []string { + out := make([]string, len(ids)) + for i, id := range ids { + out[i] = id.String() + } + return out +} diff --git a/internal/common/pulsarutils/dlq_publisher.go b/internal/common/pulsarutils/dlq_publisher.go new file mode 100644 index 00000000000..37407e7d385 --- /dev/null +++ b/internal/common/pulsarutils/dlq_publisher.go @@ -0,0 +1,88 @@ +package pulsarutils + +import ( + "strconv" + "strings" + "sync" + "time" + + "github.com/apache/pulsar-client-go/pulsar" + "github.com/pkg/errors" + + "github.com/armadaproject/armada/internal/common/armadacontext" +) + +// DeadLetterMetadata describes why a message ended up on the dead-letter topic. +// It is recorded as Pulsar message properties alongside the serialized payload. +type DeadLetterMetadata struct { + OriginalTopic string + Subscription string + Attempts int + LastError string + MessageIDs []string +} + +func (m DeadLetterMetadata) properties() map[string]string { + return map[string]string{ + "originalTopic": m.OriginalTopic, + "subscription": m.Subscription, + "attempts": strconv.Itoa(m.Attempts), + "lastError": m.LastError, + "messageIds": strings.Join(m.MessageIDs, ","), + } +} + +// DeadLetterPublisher publishes poisoned message payloads to a dead-letter Pulsar topic. +type DeadLetterPublisher struct { + producer pulsar.Producer + sendTimeout time.Duration +} + +func NewDeadLetterPublisher( + client pulsar.Client, + topic string, + compressionType pulsar.CompressionType, + compressionLevel pulsar.CompressionLevel, + sendTimeout time.Duration, +) (*DeadLetterPublisher, error) { + producer, err := client.CreateProducer(pulsar.ProducerOptions{ + Topic: topic, + CompressionType: compressionType, + CompressionLevel: compressionLevel, + BatcherBuilderType: pulsar.KeyBasedBatchBuilder, + DisableBatching: false, + }) + if err != nil { + return nil, errors.WithStack(err) + } + return &DeadLetterPublisher{ + producer: producer, + sendTimeout: sendTimeout, + }, nil +} + +// Publish sends payload to the dead-letter topic with meta recorded as message properties. +func (p *DeadLetterPublisher) Publish(ctx *armadacontext.Context, payload []byte, meta DeadLetterMetadata) error { + sendCtx, cancel := armadacontext.WithTimeout(ctx, p.sendTimeout) + defer cancel() + + wg := sync.WaitGroup{} + wg.Add(1) + var sendErr error + p.producer.SendAsync(sendCtx, &pulsar.ProducerMessage{ + Payload: payload, + Properties: meta.properties(), + }, func(_ pulsar.MessageID, _ *pulsar.ProducerMessage, err error) { + sendErr = err + wg.Done() + }) + wg.Wait() + if sendErr != nil { + return errors.WithMessage(sendErr, "error sending message to dead-letter topic") + } + return nil +} + +func (p *DeadLetterPublisher) Close() { + p.producer.Close() +} diff --git a/internal/common/pulsarutils/doc.go b/internal/common/pulsarutils/doc.go new file mode 100644 index 00000000000..22ac859ca29 --- /dev/null +++ b/internal/common/pulsarutils/doc.go @@ -0,0 +1,10 @@ +// Package pulsarutils provides shared helpers for constructing and using +// Pulsar clients, producers, and consumers. +// +// [NewPulsarClient] and [NewPulsarAdminClient] construct clients from +// [config.PulsarConfig]. [PulsarPublisher] proto-marshals and publishes +// [utils.ArmadaEvent] messages. [DeadLetterPublisher] publishes raw, +// sink-serialized payloads (with originating-topic/subscription/attempt +// metadata) to a dead-letter topic for messages an ingestion pipeline has +// given up retrying; see [internal/common/ingest]. +package pulsarutils