Implementing Dead Letter Queues

orders -> process (fail) -> orders.dlq (with error metadata)

kafka-topics.sh --bootstrap-server localhost:9092 --create \
  --topic orders.dlq --partitions 3 --replication-factor 3 \
  --config retention.ms=1209600000

try { process(r); }
catch (Exception e) {
  dlq.send(new ProducerRecord<>("orders.dlq", r.key(), r.value()));
}

stream.split()
  .branch((k,v) -> isValid(v), Branched.withConsumer(s -> s.to("orders.clean")))
  .defaultBranch(Branched.withConsumer(s -> s.to("orders.dlq")));

{"errors.tolerance":"all",
 "errors.deadletterqueue.topic.name":"orders.dlq",
 "errors.deadletterqueue.context.headers.enable":"true"}

ProducerRecord<String,String> d = new ProducerRecord<>("orders.dlq", r.key(), r.value());
d.headers().add("error.message", e.getMessage().getBytes());
d.headers().add("error.class", e.getClass().getName().getBytes());
dlq.send(d);

kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group dlq-monitor

for (var r : dlqConsumer.poll(d)) {
  String fixed = repair(r.value());
  producer.send(new ProducerRecord<>("orders", r.key(), fixed));
}

kafka-configs.sh --bootstrap-server localhost:9092 --alter \
  --entity-type topics --entity-name orders.dlq \
  --add-config retention.ms=1209600000

if (breaker.isOpen()) {
  consumer.pause(consumer.assignment());
} else {
  consumer.resume(consumer.assignment());
}