Das nimmst du mit
- Transient und permanent trennen
- Backoff mit Jitter nutzen
- DLQ-Kontext erhalten
- Reprocessing absichern
Kapitelkompass
Fehlerpfade sind Teil des Designs
Retry, DLQ und Reprocessing brauchen Klassifikation, Begrenzung und eine nachvollziehbare Betriebsentscheidung.
Praxisfall
Eine Nachricht scheitert wegen Timeout; eine andere wegen ungültiger fachlicher Daten.
Entscheidung
Nur transiente Fehler automatisch erneut versuchen; permanente Fehler sichtbar isolieren.
Typisches Risiko
Unbegrenzte Retries blockieren Partitionen und verwandeln einen Einzelfehler in Consumer Lag.
09. Retry, DLQ, Reprocessing und Consumer Lag
Thema: Retry, DLQ, Reprocessing, Consumer-Lag und
Fehlerbetrieb im Messaging.
Position: Kapitel gesamt. Messaging-Vertiefung
Kapitel.
Ziel: Nicht nur wissen, dass es Retry und DLQ gibt,
sondern entscheiden koennen, wann Retry gefaehrlich ist, wann DLQ
sinnvoll ist und wie Reprocessing produktionsnah gesteuert wird.
1. Warum Messaging-Fehlerbetrieb ein eigenes Architekturthema ist
In vielen Enterprise-Systemen wird Messaging zu einfach erklaert: Ein Producer publiziert ein Event, ein Consumer liest es und fuehrt eine Aktion aus. Diese Darstellung ist fuer ein erstes Bild hilfreich, aber sie verschweigt den Teil, der im echten Betrieb am meisten Schmerzen verursacht: Fehler passieren nicht sauber, sondern verteilt, wiederholt und oft erst spaeter sichtbar.
Ein REST-Fehler ist fuer den Aufrufer meist unmittelbar sichtbar. Ein Messaging-Fehler kann dagegen im Hintergrund wachsen: Consumer-Lag steigt, ein Retry-Sturm ueberlastet die Datenbank, eine DLQ fuellt sich, ein einzelnes Poison Event blockiert eine Partition, ein Reprocessing erzeugt doppelte Nebenwirkungen oder ein Bug wird durch automatisches Wiederholen verschleiert. Genau deshalb ist Kapitel kein kleiner Zusatz, sondern ein Kernteil produktionsnaher Enterprise-Architektur.
Typisches falsches Verstaendnis
Eine naive Regel lautet oft: “Wenn Verarbeitung fehlschlaegt, versuche es einfach erneut.” Das klingt robust, ist aber gefaehrlich. Retry hilft nur, wenn die Fehlerursache durch Zeit verschwinden kann. Ein Datenbank-Timeout kann sich erholen. Ein ungueltiges Event-Schema wird durch Warten nicht besser. Ein unbekanntes Enum in einer alten Consumer-Version wird durch zehn Retries nicht gueltig. Ein fehlender Vertrag im Fachsystem muss vielleicht manuell korrigiert werden.
Besseres Mentalmodell
Ein Consumer ist nicht nur fachlicher Code. Er ist eine kleine Betriebsmaschine:
- Er erkennt Duplikate.
- Er klassifiziert Fehler.
- Er entscheidet zwischen Retry, DLQ, Parking Lot, Skip oder Stop.
- Er sammelt Diagnosekontext.
- Er schuetzt Downstream-Systeme vor Retry-Stuermen.
- Er ermoeglicht kontrolliertes Reprocessing.
2. Fehlerklassifikation statt blindem Retry
Nicht jeder Fehler gehoert in dieselbe Schublade. Eine gute Fehlerklassifikation trennt technische und fachliche Ursachen. Noch wichtiger: Sie trennt heilbare von nicht heilbaren Fehlern.
| Fehlerart | Beispiel | Retry sinnvoll? | Typische Aktion |
|---|---|---|---|
| Transient | Timeout, kurzzeitiger Broker-/DB-Fehler | ja | Retry mit Backoff |
| Fachlich temporaer | referenzierter Kunde noch nicht repliziert | ja, begrenzt | Delayed Retry |
| Fachlich final | Vertrag dauerhaft ungueltig | nein | DLQ/Parking Lot |
| Bug | NullPointer, Mapping-Fehler | nein | Alarm, Fix, danach Reprocessing |
| Poison Message | unlesbares Schema, falsche Version | nein | isolieren, nicht endlos wiederholen |
| Duplikat | Event erneut geliefert | nein | idempotent ueberspringen |
Schlechte Loesung
try {
consumer.handle(event);
}
catch (Exception ex) {
// Schlecht: jede Ursache wird gleich behandelt.
broker.publish("retry-topic", event);
}
Diese Loesung sieht simpel aus, fuehrt aber zu drei Problemen. Erstens wird ein Bug wie ein temporaerer Fehler behandelt. Zweitens kann ein Poison Event unendlich oft verarbeitet werden. Drittens fehlen Diagnoseinformationen. Im Betrieb sieht man nur, dass Events immer wieder auftauchen.
Bessere Loesung
package com.example.messaging.Kapitel;
public final class DefaultErrorClassifier implements ErrorClassifier {
@Override public ConsumerResult.Failure classify(RuntimeException exception, EventEnvelope event) {
String m = exception.getMessage() == null ? exception.getClass().getSimpleName() : exception.getMessage();
if (m.contains("timeout") || m.contains("temporarily")) return ConsumerResult.transientFailure(m);
if (m.contains("unknown customer")) return ConsumerResult.temporaryDomain(m);
if (m.contains("invalid schema")) return ConsumerResult.poison(m);
return ConsumerResult.bug(m);
}
}
Der DefaultErrorClassifier ist absichtlich klein, zeigt
aber die Architekturidee: Klassifikation ist eine eigene Entscheidung.
In echten Systemen wuerde diese Klasse nicht nur Text analysieren,
sondern Exception-Typen, Error Codes, Status Codes, Domain-Regeln und
Downstream-Antworten auswerten.
3. Retry ist ein Werkzeug, keine Garantie
Retry verbessert Robustheit nur, wenn drei Bedingungen erfuellt sind: Der Fehler ist heilbar, die Nebenwirkung ist idempotent, und der Retry belastet das System nicht staerker als der Originalfehler. Viele Systeme scheitern, weil sie nur die erste Bedingung betrachten.
Exponential Backoff
package com.example.messaging.Kapitel;
import java.time.Duration;
// Pattern: Strategy - kapselt Retry-Regeln, ohne Consumer-Code zu veraendern.
public final class ExponentialBackoffRetryPolicy implements RetryPolicy {
private final int maxAttempts;
private final Duration baseDelay;
public ExponentialBackoffRetryPolicy(int maxAttempts, Duration baseDelay) {
this.maxAttempts = maxAttempts;
this.baseDelay = baseDelay;
}
@Override public RetryDecision decide(EventEnvelope event, ConsumerResult.Failure failure) {
if (!failure.retryable()) return RetryDecision.stop("failure is not retryable: " + failure.kind());
if (event.attempt() >= maxAttempts) return RetryDecision.stop("max attempts reached: " + event.attempt());
long factor = 1L << Math.min(event.attempt(), 8);
return RetryDecision.retryAfter(baseDelay.multipliedBy(factor), "retryable " + failure.kind());
}
}
Die Policy begrenzt die Zahl der Versuche und vergroessert die Wartezeit. Dadurch wird aus “nochmal sofort” eine kontrollierte Entlastungsstrategie. Trotzdem loest Backoff nicht alle Probleme. Wenn 50.000 Events alle dieselbe defekte Payload haben, entsteht trotz Backoff ein Strom nutzloser Arbeit. Deshalb braucht Retry immer eine DLQ- oder Parking-Lot-Strategie.
Produktionshinweis
Retry sollte messbar sein. Wichtige Metriken sind Retry-Rate, Retry-Alter, Anzahl faelliger Retries, maximale Attempt-Zahl und Anteil erfolgreich nach Retry. Ohne diese Werte sieht ein System stabil aus, obwohl es im Hintergrund aufgestaute Arbeit erzeugt.
4. DLQ: Diagnose statt Muellhalde
Eine Dead Letter Queue ist nicht einfach ein Papierkorb. Sie ist ein Diagnosekanal. Ein DLQ-Eintrag muss so viel Kontext enthalten, dass man spaeter entscheiden kann, ob Reprocessing sicher ist.
package com.example.messaging.Kapitel;
import java.time.Instant;
import java.util.*;
// Pattern: Dead Letter Channel - isoliert nicht verarbeitbare Nachrichten mit Diagnosekontext.
public final class DeadLetterQueue {
private final List<DeadLetterRecord> records = new ArrayList<>();
public void add(EventEnvelope event, ConsumerResult.Failure failure, Instant now) {
records.add(new DeadLetterRecord(event, failure.kind(), failure.message(), now));
}
public List<DeadLetterRecord> records() {
return List.copyOf(records);
}
public DeadLetterRecord removeFirst() {
return records.removeFirst();
}
public int size() {
return records.size();
}
}
Mindestens gespeichert werden sollten:
- Event-ID und Event-Typ
- Aggregate-ID oder fachlicher Key
- Schema-Version
- Correlation-ID
- Fehlerklasse
- Fehlernachricht
- Attempt-Zahl
- Zeitpunkt
- Consumer-Version
Typischer Fehler
Viele Teams legen nur die Payload in eine DLQ. Beim Incident weiss dann niemand, welche Consumer-Version den Fehler erzeugt hat, ob es schon Retries gab oder ob dasselbe Event bereits teilweise Nebenwirkungen erzeugt hat. Eine solche DLQ ist fuer Betrieb fast wertlos.
5. Parking Lot: wenn automatische Verarbeitung gefaehrlich ist
Das Parking-Lot-Muster trennt nicht automatisch verarbeitbare Events von normalen DLQ-Faellen. Ein Event im Parking Lot braucht meist menschliche oder kontrollierte technische Entscheidung: Stammdaten reparieren, Consumer-Version fixen, Mapping anpassen oder Payload migrieren.
package com.example.messaging.Kapitel;
import java.util.*;
// Pattern: Quarantine/Parking Lot - trennt operative Korrektur von automatischen Retries.
public final class ParkingLot {
private final List<DeadLetterRecord> parked = new ArrayList<>();
public void park(DeadLetterRecord record) {
parked.add(record);
}
public List<DeadLetterRecord> parked() {
return List.copyOf(parked);
}
public int size() {
return parked.size();
}
}
Parking Lot ist besonders sinnvoll bei:
- Poison Messages
- unbekannten Schema-Versionen
- Mapping-Bugs
- fachlichen Datenfehlern, die manuell repariert werden muessen
- Events mit hohem finanziellen Risiko
6. Idempotenz: die wichtigste Consumer-Eigenschaft
Broker koennen Events erneut liefern. Consumer koennen nach einer Nebenwirkung abstuerzen, bevor der Offset gespeichert wurde. Publisher koennen nach dem Senden abstuerzen, bevor der Versandstatus aktualisiert wurde. Daher ist idempotente Verarbeitung keine Optimierung, sondern Pflicht.
package com.example.messaging.Kapitel;
import java.util.*;
// Pattern: Idempotent Receiver - verhindert doppelte Nebenwirkungen bei erneuter Zustellung.
public final class IdempotencyStore {
private final Set<String> processed = new HashSet<>();
public boolean alreadyProcessed(String eventId) {
return processed.contains(eventId);
}
public void markProcessed(String eventId) {
processed.add(eventId);
}
public int size() {
return processed.size();
}
}
Eine einfache Event-ID reicht nur, wenn jede fachliche Nebenwirkung
genau an diese ID gebunden ist. In echten Systemen ist oft ein
fachlicher Idempotency-Key besser, zum Beispiel
invoice:{orderId}:{eventVersion}. So verhindert man
doppelte Rechnungen, auch wenn dasselbe fachliche Ereignis mit neuer
technischer Event-ID erneut kommt.
7. Reliable Consumer Executor als Betriebs-Fassade
Der Consumer selbst sollte fachlich bleiben. Er sollte nicht voll mit
Retry, DLQ, Metrics und Parking-Lot-Logik sein. Deshalb kapselt der
ReliableConsumerExecutor die Betriebslogik.
package com.example.messaging.Kapitel;
import java.time.Instant;
// Pattern: Facade - buendelt Retry, DLQ, Idempotenz und Metrics um einen fachlichen Consumer.
public final class ReliableConsumerExecutor {
private final Consumer consumer;
private final IdempotencyStore idempotency;
private final RetryPolicy retryPolicy;
private final RetryScheduler retryScheduler;
private final DeadLetterQueue dlq;
private final ParkingLot parkingLot;
private final ErrorClassifier classifier;
private final ProcessingMetrics metrics;
public ReliableConsumerExecutor(Consumer consumer, IdempotencyStore idempotency, RetryPolicy retryPolicy,
RetryScheduler retryScheduler, DeadLetterQueue dlq, ParkingLot parkingLot,
ErrorClassifier classifier, ProcessingMetrics metrics) {
this.consumer = consumer;
this.idempotency = idempotency;
this.retryPolicy = retryPolicy;
this.retryScheduler = retryScheduler;
this.dlq = dlq;
this.parkingLot = parkingLot;
this.classifier = classifier;
this.metrics = metrics;
}
public ConsumerResult process(EventEnvelope event, Instant now) {
if (idempotency.alreadyProcessed(event.eventId())) {
metrics.duplicate();
return ConsumerResult.duplicate(event.eventId());
}
ConsumerResult result;
try {
result = consumer.handle(event);
}
catch (RuntimeException ex) {
result = classifier.classify(ex, event);
}
if (result instanceof ConsumerResult.Success) {
idempotency.markProcessed(event.eventId());
metrics.success();
return result;
}
if (result instanceof ConsumerResult.Failure failure) {
RetryDecision decision = retryPolicy.decide(event, failure);
if (decision.retry()) {
retryScheduler.schedule(event, decision.delay(), decision.reason(), now);
metrics.retryScheduled();
return failure;
}
dlq.add(event, failure, now);
metrics.dlq();
if (failure.kind() == FailureKind.POISON_MESSAGE || failure.kind() == FailureKind.BUG) {
parkingLot.park(dlq.records().getLast());
metrics.parked();
}
return failure;
}
return result;
}
}
Diese Fassade macht die wichtigsten Regeln sichtbar:
- Duplikate werden vor der Nebenwirkung erkannt.
- Erfolgreiche Verarbeitung markiert das Event als verarbeitet.
- Retryable Failures werden geplant, nicht sofort blind wiederholt.
- Nicht retrybare Fehler gehen in die DLQ.
- Bug und Poison Message landen zusaetzlich im Parking Lot.
- Metriken werden an einer Stelle gefuehrt.
8. Consumer-Lag: Betriebskennzahl mit Fachbezug
Consumer-Lag ist nicht nur eine technische Zahl. Lag bedeutet: Es gibt fachliche Arbeit, die noch nicht verarbeitet wurde. Wenn der Billing-Consumer 200.000 Events hinterherhinkt, fehlen vielleicht Rechnungen, Reporting-Zahlen sind veraltet oder Kunden bekommen keine Benachrichtigung.
package com.example.messaging.Kapitel;
import java.util.*;
public final class ConsumerLagMonitor {
public List<LagSnapshot> measure(InMemoryTopic topic, ConsumerGroupOffsets offsets) {
List<LagSnapshot> out = new ArrayList<>();
for (int p=0; p<topic.partitions(); p++) {
long end = topic.endOffset(p);
long committed = offsets.committedOffset(topic.name(), p);
out.add(new LagSnapshot(topic.name(), p, end, committed, Math.max(0, end - committed)));
}
return out;
}
public long totalLag(InMemoryTopic topic, ConsumerGroupOffsets offsets) {
return measure(topic, offsets).stream().mapToLong(LagSnapshot::lag).sum();
}
}
Lag richtig interpretieren
Ein hoher Lag kann verschiedene Ursachen haben:
- Consumer ist zu langsam.
- Downstream-Datenbank ist langsam.
- Partition-Key erzeugt Hotspot.
- Consumer-Gruppe rebalance-t staendig.
- Poison Message erzeugt Fehlerloop.
- Retry-Sturm blockiert normale Verarbeitung.
- Payload ist zu gross.
Ein Team sollte daher nicht nur “mehr Consumer starten”. Wenn ein einzelner Key eine Partition dominiert, hilft horizontales Skalieren nur begrenzt. Wenn die Datenbank der Flaschenhals ist, macht mehr Parallelitaet die Lage schlechter.
9. Reprocessing ist ein kontrollierter Betriebsprozess
Reprocessing ist gefaehrlich, wenn es als simpler Replay-Button verstanden wird. Ein Replay kann doppelte Rechnungen erzeugen, externe APIs erneut aufrufen oder alte Events gegen neue Regeln ausfuehren. Daher braucht Reprocessing Governance.
package com.example.messaging.Kapitel;
import java.time.Instant;
import java.util.*;
// Pattern: Process Manager - steuert manuelles Reprocessing mit Limit und Audit-Hinweis.
public final class ReprocessingService {
private final InMemoryTopic topic;
public ReprocessingService(InMemoryTopic topic) {
this.topic = topic;
}
public List<MessageRecord> replayFromDlq(DeadLetterQueue dlq, int maxRecords, String operator) {
List<MessageRecord> requeued = new ArrayList<>();
int count = Math.min(maxRecords, dlq.size());
for (int i=0; i<count; i++) {
DeadLetterRecord record = dlq.removeFirst();
EventEnvelope e = record.event();
EventEnvelope replay = new EventEnvelope(e.eventId(), e.eventType(), e.aggregateId(), e.schemaVersion(), e.occurredAt(),
new java.util.LinkedHashMap<>(e.headers()) {
{
put("reprocessedBy", operator); put("reprocessedAt", Instant.now().toString());
}
},
e.payload(), 0);
requeued.add(topic.append(replay.aggregateId(), replay));
}
return requeued;
}
}
Der Code zeigt bewusst ein Limit. Reprocessing sollte nicht standardmaessig “alles aus DLQ zurueck” bedeuten. Besser ist ein kleines Fenster, Beobachtung, dann weiteres Fenster.
Reprocessing-Checkliste
- Ist die Ursache behoben?
- Sind Consumer idempotent?
- Sind Nebenwirkungen sicher oder kompensierbar?
- Gibt es ein Limit pro Lauf?
- Wird der Operator protokolliert?
- Wird Lag waehrend Reprocessing beobachtet?
- Gibt es eine Abbruchregel?
10. Polling, Offset-Commit und Fehlerstrategie
package com.example.messaging.Kapitel;
import java.time.Instant;
import java.util.*;
public final class BatchPoller {
private final InMemoryTopic topic;
private final ConsumerGroupOffsets offsets;
private final ReliableConsumerExecutor executor;
public BatchPoller(InMemoryTopic topic, ConsumerGroupOffsets offsets, ReliableConsumerExecutor executor) {
this.topic = topic;
this.offsets = offsets;
this.executor = executor;
}
public int pollOnce(int maxPerPartition, Instant now) {
int processed = 0;
for (int p=0; p<topic.partitions(); p++) {
long start = offsets.committedOffset(topic.name(), p);
List<MessageRecord> records = topic.readFrom(p, start, maxPerPartition);
for (MessageRecord record : records) {
executor.process(record.event(), now);
offsets.commit(record);
// In echter Infrastruktur haengt Commit von Fehlerstrategie ab.
processed++;
}
}
return processed;
}
}
Der BatchPoller committet in diesem Lab nach der
Verarbeitung, um das Modell einfach zu halten. In echten Kafka-Systemen
ist diese Entscheidung sehr wichtig. Commit vor Nebenwirkung kann Daten
verlieren. Commit nach Nebenwirkung kann Duplikate erzeugen, wenn der
Prozess danach abstuerzt. Deshalb lautet die praktische Regel: Duplikate
akzeptieren und Nebenwirkungen idempotent machen ist meist robuster als
zu versuchen, Duplikate komplett zu vermeiden.
11. Testszenarien im Lab
package com.example.messaging.Kapitel;
import java.time.*;
public final class MessagingKapitel {
public static void main(String[] args) {
retriesTransientFailureAndThenSucceeds();
sendsFinalDomainFailureToDlq();
skipsDuplicateEvent();
parksPoisonMessage();
measuresConsumerLag();
reprocessesFromDlqWithLimit();
System.out.println("Kapitel");
}
static Env env() {
return new Env();
}
static void retriesTransientFailureAndThenSucceeds() {
Env e = env();
e.consumer.failTransientTimes("o-1", 1);
EventEnvelope event = EventEnvelope.newOrderEvent("o-1", "{}");
e.executor.process(event, Instant.EPOCH);
assertEquals(1, e.retry.size(), "retry scheduled");
EventEnvelope due = e.retry.due(Instant.EPOCH.plusSeconds(100)).getFirst();
e.executor.process(due, Instant.EPOCH.plusSeconds(100));
assertEquals(1, e.metrics.successCount(), "success after retry");
assertEquals(0, e.dlq.size(), "no dlq");
}
static void sendsFinalDomainFailureToDlq() {
Env e = env();
e.consumer.failFinalDomain("o-2");
e.executor.process(EventEnvelope.newOrderEvent("o-2", "{}"), Instant.EPOCH);
assertEquals(1, e.dlq.size(), "final failure dlq");
}
static void skipsDuplicateEvent() {
Env e = env();
EventEnvelope event = EventEnvelope.newOrderEvent("o-3", "{}");
e.executor.process(event, Instant.EPOCH);
e.executor.process(event, Instant.EPOCH);
assertEquals(1, e.metrics.successCount(), "one side effect");
assertEquals(1, e.metrics.duplicateCount(), "duplicate skipped");
}
static void parksPoisonMessage() {
Env e = env();
e.consumer.poison("o-4");
e.executor.process(EventEnvelope.newOrderEvent("o-4", "bad"), Instant.EPOCH);
assertEquals(1, e.dlq.size(), "poison dlq");
assertEquals(1, e.parking.size(), "poison parked");
}
static void measuresConsumerLag() {
InMemoryTopic topic = new InMemoryTopic("orders", 2);
ConsumerGroupOffsets offsets = new ConsumerGroupOffsets();
topic.append("a", EventEnvelope.newOrderEvent("a", "{}"));
topic.append("b", EventEnvelope.newOrderEvent("b", "{}"));
long lag = new ConsumerLagMonitor().totalLag(topic, offsets);
if (lag < 2) throw new AssertionError("lag must be at least 2 but was " + lag);
}
static void reprocessesFromDlqWithLimit() {
InMemoryTopic topic = new InMemoryTopic("orders", 2);
DeadLetterQueue dlq = new DeadLetterQueue();
dlq.add(EventEnvelope.newOrderEvent("x", "{}"), ConsumerResult.finalDomain("broken"), Instant.EPOCH);
dlq.add(EventEnvelope.newOrderEvent("y", "{}"), ConsumerResult.finalDomain("broken"), Instant.EPOCH);
ReprocessingService service = new ReprocessingService(topic);
assertEquals(1, service.replayFromDlq(dlq, 1, "ops").size(), "one replayed");
assertEquals(1, dlq.size(), "one remains");
}
static void assertEquals(Object expected, Object actual, String label) {
if (!expected.equals(actual)) throw new AssertionError(label + " expected=" + expected + " actual=" + actual);
}
static final class Env {
final FlakyBillingProjectionConsumer consumer = new FlakyBillingProjectionConsumer();
final RetryScheduler retry = new RetryScheduler();
final DeadLetterQueue dlq = new DeadLetterQueue();
final ParkingLot parking = new ParkingLot();
final ProcessingMetrics metrics = new ProcessingMetrics();
final ReliableConsumerExecutor executor = new ReliableConsumerExecutor(consumer, new IdempotencyStore(),
new ExponentialBackoffRetryPolicy(2, Duration.ofSeconds(5)), retry, dlq, parking,
new DefaultErrorClassifier(), metrics);
}
}
Die Tests pruefen nicht nur triviale Happy Paths. Sie pruefen die Fehlerwege, die im Betrieb relevant werden:
- transienter Fehler erzeugt Retry und wird spaeter erfolgreich verarbeitet
- finale fachliche Fehler gehen in DLQ
- Duplikate werden uebersprungen
- Poison Messages werden geparkt
- Consumer-Lag wird sichtbar
- DLQ-Reprocessing ist limitiert
12. Runbook fuer Incident-Situationen
Ein Messaging-Incident darf nicht mit blinden Aktionen beantwortet werden. Die falsche Reaktion auf hohen Lag ist oft “mehr Instanzen”. Die falsche Reaktion auf DLQ ist oft “alles erneut senden”. Beides kann den Schaden vergroessern.
package com.example.messaging.Kapitel;
public final class RunbookAdvisor {
public String advise(long lag, int dlqSize, int retrySize) {
if (dlqSize > 0 && retrySize > 0) return "DLQ und Retry gleichzeitig: Fehlerklassifikation pruefen, nicht blind skalieren.";
if (lag > 1000) return "Lag hoch: Downstream-Latenz, Partition-Hotspot und Consumer-Rebalancing pruefen.";
if (dlqSize > 0) return "DLQ vorhanden: Samples lesen, Fehlerklasse bestimmen, Reprocessing erst nach Fix.";
if (retrySize > 100) return "Retry-Sturm: Backoff vergroessern, Rate limitieren, Circuit Breaker pruefen.";
return "System stabil: SLO, Lag und DLQ weiter beobachten.";
}
}
Typische Produktionsfehler
| Symptom | Falsche Reaktion | Bessere Reaktion |
|---|---|---|
| Lag steigt | mehr Consumer starten | Ursache finden: Downstream, Hotspot, Fehlerloop |
| DLQ waechst | alles replayen | Samples klassifizieren, Ursache fixen |
| Retry-Rate steigt | Retry aggressiver machen | Backoff vergroessern, Downstream schuetzen |
| Poison Message | endlos retryen | isolieren, Schema/Mapping fixen |
| Reprocessing erzeugt Fehler | nochmal replayen | stoppen, Nebenwirkungen pruefen |
13. Code-Walkthrough: Demo
package com.example.messaging.Kapitel;
import java.time.*;
public final class MessagingKapitel {
public static void main(String[] args) {
InMemoryTopic topic = new InMemoryTopic("order-events", 3);
ConsumerGroupOffsets offsets = new ConsumerGroupOffsets();
FlakyBillingProjectionConsumer consumer = new FlakyBillingProjectionConsumer();
consumer.failTransientTimes("order-200", 1);
consumer.failFinalDomain("order-404");
consumer.poison("order-666");
RetryScheduler retry = new RetryScheduler();
DeadLetterQueue dlq = new DeadLetterQueue();
ParkingLot parking = new ParkingLot();
ProcessingMetrics metrics = new ProcessingMetrics();
ReliableConsumerExecutor executor = new ReliableConsumerExecutor(consumer, new IdempotencyStore(),
new ExponentialBackoffRetryPolicy(2, Duration.ofSeconds(10)), retry, dlq, parking,
new DefaultErrorClassifier(), metrics);
topic.append("order-100", EventEnvelope.newOrderEvent("order-100", "{total:42}"));
topic.append("order-200", EventEnvelope.newOrderEvent("order-200", "{total:99}"));
topic.append("order-404", EventEnvelope.newOrderEvent("order-404", "{total:11}"));
topic.append("order-666", EventEnvelope.newOrderEvent("order-666", "{total:13}"));
BatchPoller poller = new BatchPoller(topic, offsets, executor);
poller.pollOnce(10, Instant.parse("2026-01-01T10:00:00Z"));
for (EventEnvelope due : retry.due(Instant.parse("2026-01-01T10:01:00Z"))) executor.process(due, Instant.parse("2026-01-01T10:01:00Z"));
long lag = new ConsumerLagMonitor().totalLag(topic, offsets);
System.out.println("Kapitel metrics=" + metrics + " lag=" + lag + " dlq=" + dlq.size() + " parking=" + parking.size());
}
}
Die Demo erzeugt vier Events: ein erfolgreiches Event, ein transient
fehlschlagendes Event, ein final fachlich fehlschlagendes Event und ein
Poison Event. Dadurch sieht man, warum ein Consumer nicht nur
try/catch sein darf. Dasselbe technische Muster entscheidet
je nach Fehlerklasse anders.
14. Architekturentscheidungen
- Retry ist begrenzt. Unbegrenzter Retry versteckt Fehler und erzeugt Last.
- DLQ enthaelt Diagnosekontext. Payload allein reicht nicht.
- Parking Lot isoliert gefaehrliche Events. Nicht alles darf automatisch wieder in den normalen Fluss.
- Idempotenz ist Pflicht. Duplikate sind kein Sonderfall, sondern Normalfall verteilter Systeme.
- Reprocessing ist ein Betriebsprozess. Es braucht Limit, Operator, Audit und Beobachtung.
15. Deep-Dive Lernaufgaben
- Erweitere
DefaultErrorClassifier, sodass HTTP-Statuscodes getrennt klassifiziert werden. - Erweitere
RetryPolicy, sodass bestimmte Aggregate-IDs langsamer wiederholt werden. - Baue einen Report, der DLQ-Eintraege nach Fehlerart gruppiert.
- Erweitere
ReprocessingService, sodass nur Events eines Typs reprocessed werden. - Fuege eine Metrik fuer “oldest retry age” hinzu.
- Diskutiere, wann man bei Poison Messages den Consumer komplett stoppen sollte.
16. Zusammenfassung
Kapitel zeigt, dass Messaging-Betrieb eine Architekturdisziplin ist. Retry, DLQ und Reprocessing sind nicht nur technische Features eines Brokers, sondern Entscheidungen ueber Konsistenz, Risiko, Last und fachliche Korrektheit. Ein produktionsnahes Enterprise-System braucht deshalb nicht nur Topics und Consumer, sondern Fehlerklassifikation, Idempotenz, Metriken, Runbooks und kontrolliertes Reprocessing.