[Ret2Hell]

Summary by CodeRabbit

• New Features

  • Added dead-letter queue (DLQ) support to capture malformed/poison records, with configurable topic and producer tuning parameters.

• Refactor

  • Record processing now routes invalid records to the DLQ, adds richer metadata headers, and improves error handling and shutdown sequencing.

• Tests

  • New unit coverage for DLQ routing and producer behavior.

• Chores

  • Updated example environment file with DLQ-related variables and defaults.

[coderabbitai]

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Free

Run ID: 13d48cad-a039-46a9-af75-9996d5b3f77b

📥 Commits

Reviewing files that changed from the base of the PR and between fa43f6c and a6ff308.

📒 Files selected for processing (2)

• cmd/processor/main.go
• internal/consumer/consumer_test.go

🚧 Files skipped from review as they are similar to previous changes (1)

• internal/consumer/consumer_test.go

---

📝 Walkthrough

Walkthrough

A Dead-Letter Queue (DLQ) feature was added: new DLQ environment variables, a new internal/producer Redpanda producer, consumer wiring to route malformed records to the DLQ, and main process startup/shutdown changes to create and close the DLQ producer. Runtime parsing helpers were refactored.

Changes

Cohort / File(s)	Summary
Environment & Config \.env.example, internal/config/config.go	Added DLQ env vars (DLQ_TOPIC, DLQ_PRODUCER_*), extended ProcessorConfig with DLQ fields, and moved numeric/duration parsing into helper functions (parseInt, parseInt32, parseDuration).
Producer Implementation & Tests internal/producer/producer.go, internal/producer/producer_test.go	New producer package with RedpandaProducer (kgo-backed) and options including batching, linger, buffering, retries, SASL. Implements ProduceSync and Close. Tests for construction across SASL options and header cloning.
Consumer Logic & Tests internal/consumer/consumer.go, internal/consumer/consumer_test.go	Added DLQProducer interface and injected DLQ producer into NewEventConsumer. Introduced handleRecord and routeToDLQ to send malformed records to DLQ (with headers); collectPending now propagates errors if DLQ routing fails. Tests include fake DLQ producer and assertions for DLQ behavior.
Application Wiring cmd/processor/main.go	Create and validate DLQ producer at startup, pass DLQ producer into event consumer, and ensure DLQ producer is closed during shutdown; added error-path cleanup for ClickHouse writer and DLQ producer.

Sequence Diagram

sequenceDiagram
    participant EventConsumer
    participant RecordHandler
    participant DLQProducer
    participant Redpanda

    EventConsumer->>RecordHandler: process incoming record
    RecordHandler->>RecordHandler: attempt unmarshal
    alt Unmarshal succeeds
        RecordHandler->>EventConsumer: append parsed event to batch
    else Unmarshal fails
        RecordHandler->>DLQProducer: ProduceSync(ctx, key, value, headers)
        DLQProducer->>DLQProducer: clone headers and build DLQ record
        DLQProducer->>Redpanda: sync produce DLQ record
        Redpanda-->>DLQProducer: ack / error
        alt Produce success
            DLQProducer-->>RecordHandler: success
            RecordHandler->>EventConsumer: continue (skip record)
        else Produce failure
            DLQProducer-->>RecordHandler: error
            RecordHandler-->>EventConsumer: return error (stop batch)
        end
    end

Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~50 minutes

Poem

🐰 I nibble at code with a curious peek,

Malformed bytes now find a safe creek.
Headers kept tidy, the DLQ hums low,
Poison records land where they ought to go.
Hooray — the pipeline hops on, steady and meek.

---

Note

🎁 Summarized by CodeRabbit Free

Your organization is on the Free plan. CodeRabbit will generate a high-level summary and a walkthrough for each pull request. For a comprehensive line-by-line review, please upgrade your subscription to CodeRabbit Pro by visiting https://app.coderabbit.ai/login.

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[sonarqubecloud]

Quality Gate passed

Issues
0 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud