Messaging-Retry-und-DLQ-Lab

package com.example.messaging.run6b;

import java.time.*;

public final class MessagingRun6BDemo {
    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("RUN6B_DEMO_OK metrics=" + metrics + " lag=" + lag + " dlq=" + dlq.size() + " parking=" + parking.size());
    }
}

package com.example.messaging.run6b;

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;
    }
}

package com.example.messaging.run6b;

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);
    }
}

package com.example.messaging.run6b;

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());
    }
}

package com.example.messaging.run6b;

import java.time.*;
import java.util.*;

// Pattern: Scheduler - entkoppelt Retry-Zeitpunkt vom fachlichen Consumer.
public final class RetryScheduler {
    private final List<RetryEntry> entries = new ArrayList<>();
    public void schedule(EventEnvelope event, Duration delay, String reason, Instant now) {
        entries.add(new RetryEntry(event.nextAttempt(), now.plus(delay), reason));
    }
    public List<EventEnvelope> due(Instant now) {
        List<EventEnvelope> due = new ArrayList<>();
        Iterator<RetryEntry> it = entries.iterator();
        while (it.hasNext()) {
            RetryEntry entry = it.next();
            if (!entry.dueAt().isAfter(now)) { due.add(entry.event()); it.remove(); }
        }
        return due;
    }
    public int size() { return entries.size(); }
    public List<RetryEntry> entries() { return List.copyOf(entries); }
}

package com.example.messaging.run6b;

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(); }
}

package com.example.messaging.run6b;

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;
    }
}

package com.example.messaging.run6b;

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.";
    }
}