From 0e0aa646b4e4fc4ece25da0814878fc59914310c Mon Sep 17 00:00:00 2001 From: Ret2Hell Date: Thu, 26 Mar 2026 00:43:02 +0100 Subject: [PATCH 1/4] added a sync producer for dlq --- .env.example | 6 ++ internal/config/config.go | 49 +++++++--- internal/producer/producer.go | 142 +++++++++++++++++++++++++++++ internal/producer/producer_test.go | 99 ++++++++++++++++++++ 4 files changed, 282 insertions(+), 14 deletions(-) create mode 100644 internal/producer/producer.go create mode 100644 internal/producer/producer_test.go diff --git a/.env.example b/.env.example index e150de6..2a7922b 100644 --- a/.env.example +++ b/.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=10ms +DLQ_PRODUCER_MAX_BUFFERED_RECORDS=50000 +DLQ_PRODUCER_RECORD_RETRIES=3 +DLQ_PRODUCER_RETRY_TIMEOUT=30s CONSUMER_GROUP=event-processor CONSUMER_TOPIC=raw-events CONSUMER_RESET_OFFSET=end diff --git a/internal/config/config.go b/internal/config/config.go index 05ee298..6a3c93d 100644 --- a/internal/config/config.go +++ b/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) diff --git a/internal/producer/producer.go b/internal/producer/producer.go new file mode 100644 index 0000000..58d5dc4 --- /dev/null +++ b/internal/producer/producer.go @@ -0,0 +1,142 @@ +package producer + +import ( + "context" + "fmt" + "os" + "strings" + "time" + + "github.com/rs/zerolog" + "github.com/twmb/franz-go/pkg/kgo" + "github.com/twmb/franz-go/pkg/sasl/scram" +) + +type Options struct { + Topic string + BatchMaxBytes int32 + Linger time.Duration + MaxBufferedRecords int + RecordRetries int + RetryTimeout time.Duration +} + +type ConnectionOptions struct { + SASL SASLOptions +} + +type SASLOptions struct { + User string + Password string + Mechanism string +} + +type RedpandaProducer struct { + client *kgo.Client + logger *zerolog.Logger + topic string +} + +func NewRedpandaProducer( + brokers []string, + connection ConnectionOptions, + options Options, + logger *zerolog.Logger, +) (*RedpandaProducer, error) { + topic := strings.TrimSpace(options.Topic) + if topic == "" { + return nil, fmt.Errorf("topic must not be empty") + } + if options.BatchMaxBytes <= 0 { + return nil, fmt.Errorf("batch max bytes must be > 0") + } + if options.Linger < 0 { + return nil, fmt.Errorf("linger must be >= 0") + } + if options.MaxBufferedRecords <= 0 { + return nil, fmt.Errorf("max buffered records must be > 0") + } + if options.RecordRetries < 0 { + return nil, fmt.Errorf("record retries must be >= 0") + } + if options.RetryTimeout <= 0 { + return nil, fmt.Errorf("retry timeout must be > 0") + } + + opts := []kgo.Opt{ + kgo.SeedBrokers(brokers...), + kgo.ProducerBatchCompression(kgo.Lz4Compression()), + kgo.ProducerBatchMaxBytes(options.BatchMaxBytes), + kgo.ProducerLinger(options.Linger), + kgo.RequiredAcks(kgo.AllISRAcks()), + kgo.RecordPartitioner(kgo.StickyKeyPartitioner(nil)), + kgo.MaxBufferedRecords(options.MaxBufferedRecords), + kgo.RecordRetries(options.RecordRetries), + kgo.RetryTimeout(options.RetryTimeout), + kgo.WithLogger(kgo.BasicLogger(os.Stderr, kgo.LogLevelWarn, nil)), + } + + if connection.SASL.User != "" { + auth := scram.Auth{ + User: connection.SASL.User, + Pass: connection.SASL.Password, + } + switch strings.TrimSpace(connection.SASL.Mechanism) { + case "SCRAM-SHA-512": + opts = append(opts, kgo.SASL(auth.AsSha512Mechanism())) + default: + opts = append(opts, kgo.SASL(auth.AsSha256Mechanism())) + } + } + + client, err := kgo.NewClient(opts...) + if err != nil { + return nil, fmt.Errorf("create kafka client: %w", err) + } + + return &RedpandaProducer{client: client, logger: logger, topic: topic}, nil +} + +func (p *RedpandaProducer) ProduceSync( + ctx context.Context, + key string, + value []byte, + headers []kgo.RecordHeader, +) error { + record := &kgo.Record{ + Topic: p.topic, + Key: []byte(key), + Value: append([]byte(nil), value...), + Headers: cloneHeaders(headers), + } + + if err := p.client.ProduceSync(ctx, record).FirstErr(); err != nil { + if p.logger != nil { + p.logger.Error().Err(err).Str("topic", p.topic).Msg("sync produce failed") + } + return fmt.Errorf("produce to topic %q: %w", p.topic, err) + } + + return nil +} + +func cloneHeaders(headers []kgo.RecordHeader) []kgo.RecordHeader { + if len(headers) == 0 { + return nil + } + + cloned := make([]kgo.RecordHeader, 0, len(headers)) + for _, header := range headers { + cloned = append(cloned, kgo.RecordHeader{ + Key: header.Key, + Value: append([]byte(nil), header.Value...), + }) + } + + return cloned +} + +func (p *RedpandaProducer) Close() { + p.client.Flush(context.Background()) // nolint: errcheck + p.client.Close() +} diff --git a/internal/producer/producer_test.go b/internal/producer/producer_test.go new file mode 100644 index 0000000..1b0c7a8 --- /dev/null +++ b/internal/producer/producer_test.go @@ -0,0 +1,99 @@ +package producer + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" + "github.com/twmb/franz-go/pkg/kgo" +) + +const testBrokerAddr = "localhost:69" + +func testProducerOptions() Options { + return Options{ + Topic: "dlq", + BatchMaxBytes: 1024 * 1024, + Linger: 10 * time.Millisecond, + MaxBufferedRecords: 50000, + RecordRetries: 3, + RetryTimeout: 30 * time.Second, + } +} + +func TestNewRedpandaProducerNoSASL(t *testing.T) { + p, err := NewRedpandaProducer( + []string{testBrokerAddr}, + ConnectionOptions{}, + testProducerOptions(), + nil, + ) + require.NoError(t, err) + require.NotNil(t, p) + defer p.Close() +} + +func TestNewRedpandaProducerWithSASLSHA256(t *testing.T) { + p, err := NewRedpandaProducer( + []string{testBrokerAddr}, + ConnectionOptions{ + SASL: SASLOptions{ + User: "user", + Password: "pass", + Mechanism: "SCRAM-SHA-256", + }, + }, + testProducerOptions(), + nil, + ) + require.NoError(t, err) + require.NotNil(t, p) + defer p.Close() +} + +func TestNewRedpandaProducerWithSASLSHA512(t *testing.T) { + p, err := NewRedpandaProducer( + []string{testBrokerAddr}, + ConnectionOptions{ + SASL: SASLOptions{ + User: "user", + Password: "pass", + Mechanism: "SCRAM-SHA-512", + }, + }, + testProducerOptions(), + nil, + ) + require.NoError(t, err) + require.NotNil(t, p) + defer p.Close() +} + +func TestNewRedpandaProducerUnknownMechanismFallsBackToSHA256(t *testing.T) { + p, err := NewRedpandaProducer( + []string{testBrokerAddr}, + ConnectionOptions{ + SASL: SASLOptions{ + User: "user", + Password: "pass", + Mechanism: "UNKNOWN", + }, + }, + testProducerOptions(), + nil, + ) + require.NoError(t, err) + require.NotNil(t, p) + defer p.Close() +} + +func TestCloneHeadersClonesHeaderValues(t *testing.T) { + value := []byte("bad payload") + headers := []kgo.RecordHeader{{Key: "error_reason", Value: value}} + + cloned := cloneHeaders(headers) + value[0] = 'g' + + require.Equal(t, []byte("bad payload"), cloned[0].Value) + require.NotEqual(t, &headers[0].Value[0], &cloned[0].Value[0]) +} From 6a9baf69430e4b3ff1aee6718da97e47b24c8541 Mon Sep 17 00:00:00 2001 From: Ret2Hell Date: Thu, 26 Mar 2026 01:01:38 +0100 Subject: [PATCH 2/4] route to dlq for malformed events --- cmd/processor/main.go | 33 ++++++++ internal/consumer/consumer.go | 92 ++++++++++++++++---- internal/consumer/consumer_test.go | 129 +++++++++++++++++++++++++++-- 3 files changed, 231 insertions(+), 23 deletions(-) diff --git a/cmd/processor/main.go b/cmd/processor/main.go index e4361a3..28577f1 100644 --- a/cmd/processor/main.go +++ b/cmd/processor/main.go @@ -15,6 +15,7 @@ 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" @@ -58,6 +59,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("clickhouse writer shutdown error") + } + return + } eventConsumer, err := consumer.NewEventConsumer( cfg.RedpandaBrokers, @@ -81,11 +108,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("clickhouse writer shutdown error") + } return } @@ -124,6 +156,7 @@ 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") } diff --git a/internal/consumer/consumer.go b/internal/consumer/consumer.go index fa8e158..c965283 100644 --- a/internal/consumer/consumer.go +++ b/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) { diff --git a/internal/consumer/consumer_test.go b/internal/consumer/consumer_test.go index c589a92..c4bc246 100644 --- a/internal/consumer/consumer_test.go +++ b/internal/consumer/consumer_test.go @@ -3,6 +3,7 @@ package consumer import ( "context" "encoding/json" + "errors" "testing" "time" @@ -21,11 +22,40 @@ const ( testGroupID = "test-group" ) +type fakeDLQProducer struct { + produced []producedRecord + err error +} + +type producedRecord struct { + key string + value []byte + headers []kgo.RecordHeader +} + +func (p *fakeDLQProducer) ProduceSync( + _ context.Context, + key string, + value []byte, + headers []kgo.RecordHeader, +) error { + if p.err != nil { + return p.err + } + + p.produced = append(p.produced, producedRecord{ + key: key, + value: append([]byte(nil), value...), + headers: append([]kgo.RecordHeader(nil), headers...), + }) + return nil +} + func TestNewEventConsumerWithoutSASL(t *testing.T) { logger := zerolog.Nop() brokers := []string{testBroker} - consumer, err := NewEventConsumer(brokers, ConnectionOptions{GroupID: testGroupID}, testOptions(), testPipeline(), &logger) + consumer, err := NewEventConsumer(brokers, ConnectionOptions{GroupID: testGroupID}, testOptions(), &fakeDLQProducer{}, testPipeline(), &logger) require.NoError(t, err) assert.NotNil(t, consumer) @@ -49,6 +79,7 @@ func TestNewEventConsumerWithSASLSHA256(t *testing.T) { }, }, testOptions(), + &fakeDLQProducer{}, testPipeline(), &logger, ) @@ -73,6 +104,7 @@ func TestNewEventConsumerWithSASLSHA512(t *testing.T) { }, }, testOptions(), + &fakeDLQProducer{}, testPipeline(), &logger, ) @@ -96,6 +128,7 @@ func TestNewEventConsumerWithEmptyMechanismFallsBackToSHA256(t *testing.T) { }, }, testOptions(), + &fakeDLQProducer{}, testPipeline(), &logger, ) @@ -119,6 +152,7 @@ func TestNewEventConsumerWithUnknownMechanismFallsBackToSHA256(t *testing.T) { }, }, testOptions(), + &fakeDLQProducer{}, testPipeline(), &logger, ) @@ -131,7 +165,7 @@ func TestNewEventConsumerWithMultipleBrokers(t *testing.T) { logger := zerolog.Nop() brokers := []string{"broker1:9092", "broker2:9092", "broker3:9092"} - consumer, err := NewEventConsumer(brokers, ConnectionOptions{GroupID: testGroupID}, testOptions(), testPipeline(), &logger) + consumer, err := NewEventConsumer(brokers, ConnectionOptions{GroupID: testGroupID}, testOptions(), &fakeDLQProducer{}, testPipeline(), &logger) require.NoError(t, err) assert.NotNil(t, consumer) @@ -141,7 +175,7 @@ func TestEventConsumerClose(t *testing.T) { logger := zerolog.Nop() brokers := []string{testBroker} - consumer, err := NewEventConsumer(brokers, ConnectionOptions{GroupID: testGroupID}, testOptions(), testPipeline(), &logger) + consumer, err := NewEventConsumer(brokers, ConnectionOptions{GroupID: testGroupID}, testOptions(), &fakeDLQProducer{}, testPipeline(), &logger) require.NoError(t, err) assert.NotPanics(t, func() { @@ -247,7 +281,7 @@ func TestRunContextCancellation(t *testing.T) { logger := zerolog.Nop() brokers := []string{testBroker} - consumer, err := NewEventConsumer(brokers, ConnectionOptions{GroupID: testGroupID}, testOptions(), testPipeline(), &logger) + consumer, err := NewEventConsumer(brokers, ConnectionOptions{GroupID: testGroupID}, testOptions(), &fakeDLQProducer{}, testPipeline(), &logger) require.NoError(t, err) defer consumer.Close() @@ -273,7 +307,7 @@ func TestNewEventConsumerAppliesCustomOptions(t *testing.T) { BatchTimeout: 2 * time.Second, } - consumer, err := NewEventConsumer([]string{testBroker}, ConnectionOptions{GroupID: testGroupID}, opts, testPipeline(), &logger) + consumer, err := NewEventConsumer([]string{testBroker}, ConnectionOptions{GroupID: testGroupID}, opts, &fakeDLQProducer{}, testPipeline(), &logger) require.NoError(t, err) defer consumer.Close() @@ -323,6 +357,7 @@ func TestNewEventConsumerRejectsInvalidResetOffset(t *testing.T) { []string{testBroker}, ConnectionOptions{GroupID: testGroupID}, invalidResetOptions(), + &fakeDLQProducer{}, testPipeline(), &logger, ) @@ -337,7 +372,7 @@ func TestNewEventConsumerRejectsNonPositiveFetchMaxBytes(t *testing.T) { opts := testOptions() opts.FetchMaxBytes = 0 - consumer, err := NewEventConsumer([]string{testBroker}, ConnectionOptions{GroupID: testGroupID}, opts, testPipeline(), &logger) + consumer, err := NewEventConsumer([]string{testBroker}, ConnectionOptions{GroupID: testGroupID}, opts, &fakeDLQProducer{}, testPipeline(), &logger) require.Error(t, err) assert.Nil(t, consumer) @@ -349,7 +384,7 @@ func TestNewEventConsumerRejectsNonPositiveBatchSize(t *testing.T) { opts := testOptions() opts.BatchSize = 0 - consumer, err := NewEventConsumer([]string{testBroker}, ConnectionOptions{GroupID: testGroupID}, opts, testPipeline(), &logger) + consumer, err := NewEventConsumer([]string{testBroker}, ConnectionOptions{GroupID: testGroupID}, opts, &fakeDLQProducer{}, testPipeline(), &logger) require.Error(t, err) assert.Nil(t, consumer) @@ -361,7 +396,7 @@ func TestNewEventConsumerRejectsNonPositiveBatchTimeout(t *testing.T) { opts := testOptions() opts.BatchTimeout = 0 - consumer, err := NewEventConsumer([]string{testBroker}, ConnectionOptions{GroupID: testGroupID}, opts, testPipeline(), &logger) + consumer, err := NewEventConsumer([]string{testBroker}, ConnectionOptions{GroupID: testGroupID}, opts, &fakeDLQProducer{}, testPipeline(), &logger) require.Error(t, err) assert.Nil(t, consumer) @@ -377,6 +412,84 @@ func TestShouldFlushBatch(t *testing.T) { assert.True(t, shouldFlushBatch(10, now, 100, time.Second, now.Add(time.Second))) } +func TestHandleRecordRoutesMalformedJSONToDLQ(t *testing.T) { + logger := zerolog.Nop() + dlqProducer := &fakeDLQProducer{} + consumer := &EventConsumer{dlqProducer: dlqProducer, logger: &logger} + batch := &consumerBatch{} + record := &kgo.Record{ + Topic: "raw-events", + Partition: 7, + Offset: 11, + Value: []byte("{invalid json}"), + } + + err := consumer.handleRecord(context.Background(), batch, record) + + require.NoError(t, err) + require.Empty(t, batch.pending) + require.Len(t, dlqProducer.produced, 1) + assert.Equal(t, "raw-events:7:11", dlqProducer.produced[0].key) + assert.Equal(t, []byte("{invalid json}"), dlqProducer.produced[0].value) + assertHeaderValue(t, dlqProducer.produced[0].headers, "source_topic", "raw-events") + assertHeaderValue(t, dlqProducer.produced[0].headers, "source_partition", "7") + assertHeaderValue(t, dlqProducer.produced[0].headers, "source_offset", "11") + assertHeaderValueContains(t, dlqProducer.produced[0].headers, "error_reason", "unmarshal event") + assertHeaderValueContains(t, dlqProducer.produced[0].headers, "failed_at", "T") +} + +func TestHandleRecordReturnsDLQError(t *testing.T) { + logger := zerolog.Nop() + dlqProducer := &fakeDLQProducer{err: errors.New("boom")} + consumer := &EventConsumer{dlqProducer: dlqProducer, logger: &logger} + batch := &consumerBatch{} + record := &kgo.Record{Topic: "raw-events", Value: []byte("{invalid json}")} + + err := consumer.handleRecord(context.Background(), batch, record) + + require.Error(t, err) + assert.ErrorContains(t, err, "route poison record to dlq") + require.Empty(t, batch.pending) + require.Empty(t, dlqProducer.produced) +} + +func TestHandleRecordAppendsValidEvent(t *testing.T) { + logger := zerolog.Nop() + dlqProducer := &fakeDLQProducer{} + consumer := &EventConsumer{dlqProducer: dlqProducer, logger: &logger} + batch := &consumerBatch{} + record := &kgo.Record{Value: []byte(`{"event_id":"evt-1","site_id":"site-1","visitor_id":"vis-1","event_type":"page_view","timestamp":1700000000000,"url":"https://example.com"}`)} + + err := consumer.handleRecord(context.Background(), batch, record) + + require.NoError(t, err) + require.Len(t, batch.pending, 1) + assert.Equal(t, "evt-1", batch.pending[0].EventID) + assert.Empty(t, dlqProducer.produced) +} + +func assertHeaderValue(t *testing.T, headers []kgo.RecordHeader, key, want string) { + t.Helper() + assert.Equal(t, want, headerValue(t, headers, key)) +} + +func assertHeaderValueContains(t *testing.T, headers []kgo.RecordHeader, key, want string) { + t.Helper() + assert.Contains(t, headerValue(t, headers, key), want) +} + +func headerValue(t *testing.T, headers []kgo.RecordHeader, key string) string { + t.Helper() + for _, header := range headers { + if header.Key == key { + return string(header.Value) + } + } + + t.Fatalf("missing header %q", key) + return "" +} + func testOptions() Options { return Options{ Topic: "raw-events", From fa43f6c6bf0fc2d4e4ecf25e8d46bd252e9e09e2 Mon Sep 17 00:00:00 2001 From: Ret2Hell Date: Thu, 26 Mar 2026 01:10:41 +0100 Subject: [PATCH 3/4] update defaults for dlq --- .env.example | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/.env.example b/.env.example index 2a7922b..53e873d 100644 --- a/.env.example +++ b/.env.example @@ -6,10 +6,10 @@ REDPANDA_SASL_PASSWORD=12345678 REDPANDA_SASL_MECHANISM=SCRAM-SHA-256 DLQ_TOPIC=dlq DLQ_PRODUCER_BATCH_MAX_BYTES=1048576 -DLQ_PRODUCER_LINGER=10ms -DLQ_PRODUCER_MAX_BUFFERED_RECORDS=50000 -DLQ_PRODUCER_RECORD_RETRIES=3 -DLQ_PRODUCER_RETRY_TIMEOUT=30s +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 From a6ff30845f2d966540bc7ff003f2742be7234e08 Mon Sep 17 00:00:00 2001 From: Ret2Hell Date: Thu, 26 Mar 2026 01:16:01 +0100 Subject: [PATCH 4/4] fix issues --- cmd/processor/main.go | 8 +++++--- internal/consumer/consumer_test.go | 22 ++++++++++++---------- 2 files changed, 17 insertions(+), 13 deletions(-) diff --git a/cmd/processor/main.go b/cmd/processor/main.go index 28577f1..9f8b52f 100644 --- a/cmd/processor/main.go +++ b/cmd/processor/main.go @@ -21,6 +21,8 @@ import ( "github.com/rs/zerolog" ) +const clickhouseWriterShutdownErrorMsg = "clickhouse writer shutdown error" + func main() { // 1. Load configuration cfg := config.LoadProcessorConfig() @@ -81,7 +83,7 @@ func main() { if err != nil { logger.Error().Err(err).Msg("failed to initialize dlq producer") if closeErr := chWriter.Close(); closeErr != nil { - logger.Error().Err(closeErr).Msg("clickhouse writer shutdown error") + logger.Error().Err(closeErr).Msg(clickhouseWriterShutdownErrorMsg) } return } @@ -116,7 +118,7 @@ func main() { logger.Error().Err(err).Msg("failed to initialize consumer") dlqProducer.Close() if closeErr := chWriter.Close(); closeErr != nil { - logger.Error().Err(closeErr).Msg("clickhouse writer shutdown error") + logger.Error().Err(closeErr).Msg(clickhouseWriterShutdownErrorMsg) } return } @@ -158,7 +160,7 @@ func main() { 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") diff --git a/internal/consumer/consumer_test.go b/internal/consumer/consumer_test.go index c4bc246..dcde5d9 100644 --- a/internal/consumer/consumer_test.go +++ b/internal/consumer/consumer_test.go @@ -18,8 +18,10 @@ import ( ) const ( - testBroker = "localhost:9092" - testGroupID = "test-group" + testBroker = "localhost:9092" + testGroupID = "test-group" + testRawEventsTopic = "raw-events" + testInvalidJSONBody = "{invalid json}" ) type fakeDLQProducer struct { @@ -252,7 +254,7 @@ func TestInternalEventUnmarshalMinimalEvent(t *testing.T) { } func TestInternalEventUnmarshalInvalidJSON(t *testing.T) { - invalidJSON := `{invalid json}` + invalidJSON := testInvalidJSONBody var event models.InternalEvent err := json.Unmarshal([]byte(invalidJSON), &event) @@ -418,10 +420,10 @@ func TestHandleRecordRoutesMalformedJSONToDLQ(t *testing.T) { consumer := &EventConsumer{dlqProducer: dlqProducer, logger: &logger} batch := &consumerBatch{} record := &kgo.Record{ - Topic: "raw-events", + Topic: testRawEventsTopic, Partition: 7, Offset: 11, - Value: []byte("{invalid json}"), + Value: []byte(testInvalidJSONBody), } err := consumer.handleRecord(context.Background(), batch, record) @@ -429,9 +431,9 @@ func TestHandleRecordRoutesMalformedJSONToDLQ(t *testing.T) { require.NoError(t, err) require.Empty(t, batch.pending) require.Len(t, dlqProducer.produced, 1) - assert.Equal(t, "raw-events:7:11", dlqProducer.produced[0].key) - assert.Equal(t, []byte("{invalid json}"), dlqProducer.produced[0].value) - assertHeaderValue(t, dlqProducer.produced[0].headers, "source_topic", "raw-events") + assert.Equal(t, testRawEventsTopic+":7:11", dlqProducer.produced[0].key) + assert.Equal(t, []byte(testInvalidJSONBody), dlqProducer.produced[0].value) + assertHeaderValue(t, dlqProducer.produced[0].headers, "source_topic", testRawEventsTopic) assertHeaderValue(t, dlqProducer.produced[0].headers, "source_partition", "7") assertHeaderValue(t, dlqProducer.produced[0].headers, "source_offset", "11") assertHeaderValueContains(t, dlqProducer.produced[0].headers, "error_reason", "unmarshal event") @@ -443,7 +445,7 @@ func TestHandleRecordReturnsDLQError(t *testing.T) { dlqProducer := &fakeDLQProducer{err: errors.New("boom")} consumer := &EventConsumer{dlqProducer: dlqProducer, logger: &logger} batch := &consumerBatch{} - record := &kgo.Record{Topic: "raw-events", Value: []byte("{invalid json}")} + record := &kgo.Record{Topic: testRawEventsTopic, Value: []byte(testInvalidJSONBody)} err := consumer.handleRecord(context.Background(), batch, record) @@ -492,7 +494,7 @@ func headerValue(t *testing.T, headers []kgo.RecordHeader, key string) string { func testOptions() Options { return Options{ - Topic: "raw-events", + Topic: testRawEventsTopic, ResetOffset: "end", FetchMinBytes: 64 * 1024, FetchMaxBytes: 50 * 1024 * 1024,