Notification Pipeline on Kafka

You will build two services: an API that accepts notification requests and publishes them to Kafka, and a worker that consumes them and sends email or push messages through a flaky provider. The point is failure handling: transient errors are retried with exponential backoff without blocking the partition, poison messages land in a dead-letter topic, and redelivery never sends the same notification twice. Interviewers ask about Kafka delivery semantics constantly, and this project forces concrete answers.

Project layout

notification-api/     POST /notifications -> Kafka topic "notifications"
notification-worker/
  consumer/   NotificationListener, DlqListener
  service/    SenderService, ProcessedMessageRepository
  config/     KafkaConfig
docker-compose.yml    kafka (KRaft), postgres

Producer

Use an idempotent producer and key by recipient so one user's messages stay ordered within a partition.

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      acks: all
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      properties:
        enable.idempotence: true
        max.in.flight.requests.per.connection: 5

public CompletableFuture<SendResult<String, NotificationEvent>> publish(NotificationEvent e) {
    return kafka.send("notifications", e.userId(), e);
}

enable.idempotence stops broker-side duplicates caused by producer retries within one producer session. It does not deduplicate if your application sends the same event twice.

Retries with backoff and a DLQ

Blocking retries on the main topic stall every message behind the failing one in that partition. Spring Kafka's @RetryableTopic instead forwards failures to retry topics with a delay, then to a dead-letter topic.

@RetryableTopic(
    attempts = "4",
    backoff = @Backoff(delay = 2000, multiplier = 2.0, maxDelay = 30000),
    dltStrategy = DltStrategy.FAIL_ON_ERROR,
    exclude = { InvalidRecipientException.class })
@KafkaListener(topics = "notifications", groupId = "notification-worker")
public void onMessage(NotificationEvent event,
                      @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
    sender.send(event);
}

@DltHandler
public void onDead(NotificationEvent event,
                   @Header(KafkaHeaders.EXCEPTION_MESSAGE) String reason) {
    log.error("DLQ id={} reason={}", event.id(), reason);
    deadLetters.save(DeadLetter.of(event, reason));
}

This creates notifications-retry-2000, notifications-retry-4000, notifications-retry-8000 and notifications-dlt. Non-retryable errors such as InvalidRecipientException go straight to the DLT because of exclude. Retry topics preserve order only per partition and not across the retry hops, so do not rely on ordering after a failure.

Idempotent consumer

Kafka gives at-least-once delivery: a rebalance or a crash after sending but before the offset commit redelivers the record. Make the handler safe to repeat by recording the event id in the same transaction as the side effect that you control.

CREATE TABLE processed_message (
    event_id     VARCHAR(64) PRIMARY KEY,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

@Transactional
public void send(NotificationEvent e) {
    try {
        processed.saveAndFlush(new ProcessedMessage(e.id()));
    } catch (DataIntegrityViolationException dup) {
        log.info("duplicate {}, skipping", e.id());
        return;
    }
    provider.deliver(e);     // throws on transient failure -> transaction rolls back
}

If deliver throws, the transaction rolls back and the id row disappears, so the retry processes it again. The unavoidable gap: if the provider call succeeds and the commit then fails, you can send twice. Closing it requires an idempotency key at the provider; pass e.id() as the provider's request id when it supports one.

Testing

Use Testcontainers Kafka and wait for the outcome with Awaitility instead of sleeping.

@SpringBootTest
@Testcontainers
class PipelineIT {
    @Container @ServiceConnection
    static KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.6.0"));

    @Autowired KafkaTemplate<String, NotificationEvent> kafkaTemplate;
    @MockBean EmailProvider provider;

    @Test
    void duplicateEventSentOnce() {
        var e = new NotificationEvent("evt-1", "u1", "hello");
        kafkaTemplate.send("notifications", "u1", e);
        kafkaTemplate.send("notifications", "u1", e);
        await().atMost(Duration.ofSeconds(15))
               .untilAsserted(() -> verify(provider, times(1)).deliver(any()));
    }
}

Interview angle

• At-least-once, at-most-once or exactly-once? Default consumer with commit after processing is at-least-once. Exactly-once effects come from idempotent handlers, or Kafka transactions when both input and output are Kafka topics. Not from a config flag alone.
• What causes duplicates and how do you handle them? Producer retries without idempotence, consumer rebalances, and crashes between side effect and offset commit. Handle with a dedupe key stored atomically with the effect.
• Why retry topics instead of Thread.sleep in the listener? Sleeping blocks the partition and can exceed max.poll.interval.ms, triggering a rebalance. Retry topics move waiting out of the hot path.
• How do you pick the partition count? Parallelism ceiling equals partitions per consumer group; size for peak throughput divided by per-consumer rate, and remember increasing it later breaks key-to-partition mapping.