DLQ

.env.example

@@ -4,6 +4,12 @@ REDPANDA_SASL_USER=superuser
 REDPANDA_SASL_PASSWORD=12345678
 # SCRAM-SHA-256 (default) or SCRAM-SHA-512
 REDPANDA_SASL_MECHANISM=SCRAM-SHA-256
+DLQ_TOPIC=dlq
+DLQ_PRODUCER_BATCH_MAX_BYTES=1048576
+DLQ_PRODUCER_LINGER=0ms
+DLQ_PRODUCER_MAX_BUFFERED_RECORDS=1000
+DLQ_PRODUCER_RECORD_RETRIES=5
+DLQ_PRODUCER_RETRY_TIMEOUT=1m
 CONSUMER_GROUP=event-processor
 CONSUMER_TOPIC=raw-events
 CONSUMER_RESET_OFFSET=end

cmd/processor/main.go

@@ -15,11 +15,14 @@ import (
 	"github.com/Tanilytics/processing/internal/observability"
 	"github.com/Tanilytics/processing/internal/pipeline"
 	"github.com/Tanilytics/processing/internal/processors"
+	"github.com/Tanilytics/processing/internal/producer"
 	"github.com/Tanilytics/processing/internal/server"
 	"github.com/Tanilytics/processing/internal/storage"
 	"github.com/rs/zerolog"
 )
 
+const clickhouseWriterShutdownErrorMsg = "clickhouse writer shutdown error"
+
 func main() {
 	// 1. Load configuration
 	cfg := config.LoadProcessorConfig()
@@ -58,6 +61,32 @@ func main() {
 	}
 
 	processorPipeline := pipeline.NewPipeline(anonymizer, uaParser, chWriter, logger)
+	dlqProducer, err := producer.NewRedpandaProducer(
+		cfg.RedpandaBrokers,
+		producer.ConnectionOptions{
+			SASL: producer.SASLOptions{
+				User:      cfg.RedpandaSASLUser,
+				Password:  cfg.RedpandaSASLPassword,
+				Mechanism: cfg.RedpandaSASLMechanism,
+			},
+		},
+		producer.Options{
+			Topic:              cfg.DLQTopic,
+			BatchMaxBytes:      cfg.DLQProducerBatchMaxBytes,
+			Linger:             cfg.DLQProducerLinger,
+			MaxBufferedRecords: cfg.DLQProducerMaxBufferedRecords,
+			RecordRetries:      cfg.DLQProducerRecordRetries,
+			RetryTimeout:       cfg.DLQProducerRetryTimeout,
+		},
+		logger,
+	)
+	if err != nil {
+		logger.Error().Err(err).Msg("failed to initialize dlq producer")
+		if closeErr := chWriter.Close(); closeErr != nil {
+			logger.Error().Err(closeErr).Msg(clickhouseWriterShutdownErrorMsg)
+		}
+		return
+	}
 
 	eventConsumer, err := consumer.NewEventConsumer(
 		cfg.RedpandaBrokers,
@@ -81,11 +110,16 @@ func main() {
 			BatchSize:              cfg.ClickhouseBatchSize,
 			BatchTimeout:           cfg.ClickhouseBatchTimeout,
 		},
+		dlqProducer,
 		processorPipeline,
 		logger,
 	)
 	if err != nil {
 		logger.Error().Err(err).Msg("failed to initialize consumer")
+		dlqProducer.Close()
+		if closeErr := chWriter.Close(); closeErr != nil {
+			logger.Error().Err(closeErr).Msg(clickhouseWriterShutdownErrorMsg)
+		}
 		return
 	}
 
@@ -124,8 +158,9 @@ func main() {
 	gracefulShutdown(app, logger)
 	consumerWG.Wait()
 	eventConsumer.Close()
+	dlqProducer.Close()
 	if err := chWriter.Close(); err != nil {
-		logger.Error().Err(err).Msg("clickhouse writer shutdown error")
+		logger.Error().Err(err).Msg(clickhouseWriterShutdownErrorMsg)
 	}
 
 	logger.Info().Msg("shutdown complete")

internal/config/config.go

@@ -21,6 +21,12 @@ type ProcessorConfig struct {
 	// Supported values: "SCRAM-SHA-256" (default), "SCRAM-SHA-512".
 	// Leave empty to disable SASL.
 	RedpandaSASLMechanism          string
+	DLQTopic                       string
+	DLQProducerBatchMaxBytes       int32
+	DLQProducerLinger              time.Duration
+	DLQProducerMaxBufferedRecords  int
+	DLQProducerRecordRetries       int
+	DLQProducerRetryTimeout        time.Duration
 	ConsumerGroup                  string
 	ConsumerTopic                  string
 	ConsumerResetOffset            string
@@ -47,17 +53,23 @@ type ProcessorConfig struct {
 
 func LoadProcessorConfig() ProcessorConfig {
 	return ProcessorConfig{
-		Port:                  getEnv("PROCESSING_PORT"),
-		RedpandaBrokers:       strings.Split(getEnv("REDPANDA_BROKERS"), ","),
-		RedpandaSASLUser:      getEnv("REDPANDA_SASL_USER"),
-		RedpandaSASLPassword:  getEnv("REDPANDA_SASL_PASSWORD"),
-		RedpandaSASLMechanism: getEnv("REDPANDA_SASL_MECHANISM"),
-		ConsumerGroup:         getEnv("CONSUMER_GROUP"),
-		ConsumerTopic:         getEnv("CONSUMER_TOPIC"),
-		ConsumerResetOffset:   getEnv("CONSUMER_RESET_OFFSET"),
-		ConsumerFetchMinBytes: getEnvInt32("CONSUMER_FETCH_MIN_BYTES"),
-		ConsumerFetchMaxBytes: getEnvInt32("CONSUMER_FETCH_MAX_BYTES"),
-		ConsumerFetchMaxWait:  getEnvDuration("CONSUMER_FETCH_MAX_WAIT"),
+		Port:                          getEnv("PROCESSING_PORT"),
+		RedpandaBrokers:               strings.Split(getEnv("REDPANDA_BROKERS"), ","),
+		RedpandaSASLUser:              getEnv("REDPANDA_SASL_USER"),
+		RedpandaSASLPassword:          getEnv("REDPANDA_SASL_PASSWORD"),
+		RedpandaSASLMechanism:         getEnv("REDPANDA_SASL_MECHANISM"),
+		DLQTopic:                      getEnv("DLQ_TOPIC"),
+		DLQProducerBatchMaxBytes:      getEnvInt32("DLQ_PRODUCER_BATCH_MAX_BYTES"),
+		DLQProducerLinger:             getEnvDuration("DLQ_PRODUCER_LINGER"),
+		DLQProducerMaxBufferedRecords: getEnvInt("DLQ_PRODUCER_MAX_BUFFERED_RECORDS"),
+		DLQProducerRecordRetries:      getEnvInt("DLQ_PRODUCER_RECORD_RETRIES"),
+		DLQProducerRetryTimeout:       getEnvDuration("DLQ_PRODUCER_RETRY_TIMEOUT"),
+		ConsumerGroup:                 getEnv("CONSUMER_GROUP"),
+		ConsumerTopic:                 getEnv("CONSUMER_TOPIC"),
+		ConsumerResetOffset:           getEnv("CONSUMER_RESET_OFFSET"),
+		ConsumerFetchMinBytes:         getEnvInt32("CONSUMER_FETCH_MIN_BYTES"),
+		ConsumerFetchMaxBytes:         getEnvInt32("CONSUMER_FETCH_MAX_BYTES"),
+		ConsumerFetchMaxWait:          getEnvDuration("CONSUMER_FETCH_MAX_WAIT"),
 		ConsumerFetchMaxPartitionBytes: getEnvInt32(
 			"CONSUMER_FETCH_MAX_PARTITION_BYTES",
 		),
@@ -93,8 +105,11 @@ func getEnv(key string) string {
 
 func getEnvInt(key string) int {
 	v := getEnv(key)
+	return parseInt(key, v)
+}
 
-	i, err := strconv.Atoi(strings.TrimSpace(v))
+func parseInt(key, value string) int {
+	i, err := strconv.Atoi(strings.TrimSpace(value))
 	if err != nil {
 		log.Printf(invalidEnvVarLogFormat, key, err)
 		os.Exit(1)
@@ -105,8 +120,11 @@ func getEnvInt(key string) int {
 
 func getEnvInt32(key string) int32 {
 	v := getEnv(key)
+	return parseInt32(key, v)
+}
 
-	i, err := strconv.ParseInt(strings.TrimSpace(v), 10, 32)
+func parseInt32(key, value string) int32 {
+	i, err := strconv.ParseInt(strings.TrimSpace(value), 10, 32)
 	if err != nil {
 		log.Printf(invalidEnvVarLogFormat, key, err)
 		os.Exit(1)
@@ -117,8 +135,11 @@ func getEnvInt32(key string) int32 {
 
 func getEnvDuration(key string) time.Duration {
 	v := getEnv(key)
+	return parseDuration(key, v)
+}
 
-	d, err := time.ParseDuration(strings.TrimSpace(v))
+func parseDuration(key, value string) time.Duration {
+	d, err := time.ParseDuration(strings.TrimSpace(value))
 	if err != nil {
 		log.Printf(invalidEnvVarLogFormat, key, err)
 		os.Exit(1)

internal/consumer/consumer.go

@@ -15,6 +15,10 @@ import (
 	"github.com/twmb/franz-go/pkg/sasl/scram"
 )
 
+type DLQProducer interface {
+	ProduceSync(ctx context.Context, key string, value []byte, headers []kgo.RecordHeader) error
+}
+
 type Options struct {
 	Topic                  string
 	ResetOffset            string
@@ -41,6 +45,7 @@ type SASLOptions struct {
 
 type EventConsumer struct {
 	client               *kgo.Client
+	dlqProducer          DLQProducer
 	pipeline             *pipeline.Pipeline
 	logger               *zerolog.Logger
 	batchSize            int
@@ -53,7 +58,14 @@ type consumerBatch struct {
 	startedAt time.Time
 }
 
-func NewEventConsumer(brokers []string, connection ConnectionOptions, options Options, p *pipeline.Pipeline, logger *zerolog.Logger) (*EventConsumer, error) {
+func NewEventConsumer(
+	brokers []string,
+	connection ConnectionOptions,
+	options Options,
+	dlqProducer DLQProducer,
+	p *pipeline.Pipeline,
+	logger *zerolog.Logger,
+) (*EventConsumer, error) {
 	if strings.TrimSpace(options.Topic) == "" {
 		return nil, fmt.Errorf("topic must not be empty")
 	}
@@ -121,6 +133,7 @@ func NewEventConsumer(brokers []string, connection ConnectionOptions, options Op
 	}
 	return &EventConsumer{
 		client:               client,
+		dlqProducer:          dlqProducer,
 		pipeline:             p,
 		logger:               logger,
 		batchSize:            options.BatchSize,
@@ -179,7 +192,10 @@ func (c *EventConsumer) runBlockedOnce(ctx context.Context) error {
 	}
 
 	batch := consumerBatch{}
-	recordsFetched := c.collectPending(fetches, &batch)
+	recordsFetched, err := c.collectPending(ctx, fetches, &batch)
+	if err != nil {
+		return err
+	}
 	if len(batch.pending) == 0 {
 		c.commitFetchedOffsets(ctx, recordsFetched)
 		return nil
@@ -206,7 +222,10 @@ func (c *EventConsumer) runOnce(ctx context.Context, batch *consumerBatch) error
 		return nil
 	}
 
-	recordsFetched := c.collectPending(fetches, batch)
+	recordsFetched, err := c.collectPending(ctx, fetches, batch)
+	if err != nil {
+		return err
+	}
 	if len(batch.pending) == 0 {
 		c.commitFetchedOffsets(ctx, recordsFetched)
 		return nil
@@ -273,32 +292,75 @@ func (c *EventConsumer) pollFetches(ctx context.Context, batch *consumerBatch) (
 	return fetches, pollErr, true
 }
 
-func (c *EventConsumer) collectPending(fetches kgo.Fetches, batch *consumerBatch) int {
+func (c *EventConsumer) collectPending(ctx context.Context, fetches kgo.Fetches, batch *consumerBatch) (int, error) {
 	c.logFetchErrors(fetches)
 
 	batchWasEmpty := len(batch.pending) == 0
 	recordsFetched := 0
+	var collectErr error
 	fetches.EachRecord(func(record *kgo.Record) {
+		if collectErr != nil {
+			return
+		}
+
 		recordsFetched++
 
-		var event models.InternalEvent
-		if err := json.Unmarshal(record.Value, &event); err != nil {
-			c.logger.Error().
-				Err(err).
-				Int32("partition", record.Partition).
-				Int64("offset", record.Offset).
-				Msg("unmarshal event")
-			return // skip malformed records (will be DLQ'd when I feel like doing it)
+		if err := c.handleRecord(ctx, batch, record); err != nil {
+			collectErr = err
 		}
-
-		batch.pending = append(batch.pending, &event)
 	})
 
 	if batchWasEmpty && len(batch.pending) > 0 {
 		batch.startedAt = time.Now()
 	}
 
-	return recordsFetched
+	return recordsFetched, collectErr
+}
+
+func (c *EventConsumer) handleRecord(ctx context.Context, batch *consumerBatch, record *kgo.Record) error {
+	var event models.InternalEvent
+	if err := json.Unmarshal(record.Value, &event); err != nil {
+		reason := fmt.Sprintf("unmarshal event: %v", err)
+		c.logger.Error().
+			Err(err).
+			Str("topic", record.Topic).
+			Int32("partition", record.Partition).
+			Int64("offset", record.Offset).
+			Msg("unmarshal event")
+
+		if err := c.routeToDLQ(ctx, record, reason); err != nil {
+			return fmt.Errorf("route poison record to dlq: %w", err)
+		}
+
+		return nil
+	}
+
+	batch.pending = append(batch.pending, &event)
+	return nil
+}
+
+func (c *EventConsumer) routeToDLQ(ctx context.Context, record *kgo.Record, reason string) error {
+	key := fmt.Sprintf("%s:%d:%d", record.Topic, record.Partition, record.Offset)
+	headers := []kgo.RecordHeader{
+		{Key: "error_reason", Value: []byte(reason)},
+		{Key: "source_topic", Value: []byte(record.Topic)},
+		{Key: "source_partition", Value: fmt.Appendf(nil, "%d", record.Partition)},
+		{Key: "source_offset", Value: fmt.Appendf(nil, "%d", record.Offset)},
+		{Key: "failed_at", Value: []byte(time.Now().UTC().Format(time.RFC3339))},
+	}
+
+	if err := c.dlqProducer.ProduceSync(ctx, key, record.Value, headers); err != nil {
+		return err
+	}
+
+	c.logger.Warn().
+		Str("topic", record.Topic).
+		Int32("partition", record.Partition).
+		Int64("offset", record.Offset).
+		Str("reason", reason).
+		Msg("routed poison record to dlq")
+
+	return nil
 }
 
 func (c *EventConsumer) logFetchErrors(fetches kgo.Fetches) {