Kafka-Überwachung: Consumer Lag und wichtige Metriken

1. Überblick

Ein Kafka-System kann »gesund« aussehen: Broker laufen, Producer sendet Nachrichten. Aber der Consumer könnte Millionen von Nachrichten hinter sich herlaufen, ohne dass jemand es bemerkt. Die Metrik, die diesen Zustand widerspiegelt, ist Consumer Lag.

In diesem Artikel lernst du, wie Consumer Lag berechnet wird und wie man es auf drei Wegen misst: über CLI, AdminClient in Java und Micrometer in Spring Boot. Danach schauen wir uns die wichtigsten Broker-, Producer- und Consumer-Metriken an und wie man sinnvolle Alerts für sie einrichtet.

2. Was ist Consumer Lag?

Jede Partition in Kafka ist ein reines Append-Log. Es gibt zwei Offsets, die du beachten musst:

• Log-end offset (LEO): der Offset der nächsten Nachricht, die in die Partition geschrieben wird.
• Committed offset: der Offset, den die Consumer Group committed hat, also die Position, von der die Gruppe nach einem Neustart weiterlesen würde.

Der Lag einer Partition wird wie folgt berechnet:

lag = log-end offset - committed offset

Der Lag der gesamten Consumer Group ist die Summe der Lags über alle Partitionen, die die Group liest.

Ein Lag von 0 bedeutet, dass der Consumer aufgeholt hat. Leichte Schwankungen des Lag sind normal, da Nachrichten ständig ankommen. Ein stetig wachsender Lag ist das Zeichen für ein ernstes Problem. Das bedeutet, der Consumer verarbeitet langsamer als der Producer schreibt.

Hinweis: Bei Topics mit Transaktionen oder Compaction können Offsets »Lücken« aufweisen (Control Records, gelöschte Nachrichten). Lag wird nach Offsets berechnet und kann daher etwas größer sein als die Anzahl der Nachrichten, die tatsächlich noch zu lesen sind. Das ist kein Fehler.

3. Lag über CLI prüfen

Der schnellste Weg ist das Skript, das mit Kafka mitgeliefert wird:

kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group order-service

Die Ausgabe sieht so aus:

GROUP          TOPIC   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG   CONSUMER-ID       HOST
order-service  orders  0          15230           15232           2     consumer-1-a1b2   /10.0.0.5
order-service  orders  1          14980           18410           3430  consumer-1-c3d4   /10.0.0.6
order-service  orders  2          15102           15102           0     consumer-1-e5f6   /10.0.0.7

Partition 1 hängt 3430 Nachrichten hinterher, während die anderen zwei Partitionen auf dem Laufenden sind. Ein solches Ungleichgewicht hat normalerweise zwei Ursachen: Entweder gibt es einen »heißen« Key, der viele Nachrichten in eine Partition lenkt, oder der Consumer auf Host 10.0.0.6 hat ein Problem.

Die CLI ist sehr nützlich zur Fehlerbehebung. Allerdings kannst du diesen Befehl nicht jede Minute manuell ausführen. Der nächste Abschnitt zeigt, wie man die Messung automatisiert.

4. Lag mit AdminClient berechnen

Füge die Abhängigkeit kafka-clients hinzu (Spring Boot verwaltet die Version):

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
</dependency>

Zunächst isolieren wir die Berechnung in eine reine Funktion, um sie leicht testen zu können:

public final class LagCalculator {

    private LagCalculator() {}

    public static Map<TopicPartition, Long> compute(Map<TopicPartition, Long> committed,
                                                    Map<TopicPartition, Long> endOffsets) {
        Map<TopicPartition, Long> lag = new HashMap<>();
        committed.forEach((tp, committedOffset) -> {
            Long end = endOffsets.get(tp);
            if (end != null) {
                lag.put(tp, Math.max(0L, end - committedOffset));
            }
        });
        return lag;
    }
}

Die Funktion berechnet Lag nur für Partitionen, die sowohl einen Committed Offset als auch einen End Offset haben. Math.max(0L, …) hat seinen Grund. Committed Offset und End Offset werden durch zwei separate Anfragen ermittelt. Daher kann der Committed Offset manchmal »neuer« sein als der gerade geholte End Offset. Das Ergebnis wäre negativ und ist sinnlos, also setzen wir es auf 0.

Als nächstes die Klasse, die AdminClient aufruft:

public class ConsumerLagChecker {

    private final Admin admin;

    public ConsumerLagChecker(Admin admin) {
        this.admin = admin;
    }

    public Map<TopicPartition, Long> lagOf(String groupId) throws Exception {
        Map<TopicPartition, Long> committed = admin.listConsumerGroupOffsets(groupId)
                .partitionsToOffsetAndMetadata()
                .get(10, TimeUnit.SECONDS)
                .entrySet().stream()
                .filter(e -> e.getValue() != null)
                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().offset()));

        Map<TopicPartition, OffsetSpec> request = committed.keySet().stream()
                .collect(Collectors.toMap(tp -> tp, tp -> OffsetSpec.latest()));

        Map<TopicPartition, Long> endOffsets = admin.listOffsets(request)
                .all()
                .get(10, TimeUnit.SECONDS)
                .entrySet().stream()
                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().offset()));

        return LagCalculator.compute(committed, endOffsets);
    }
}

Die Schritte sind:

• listConsumerGroupOffsets gibt die Committed Offsets der Group zurück. Der Wert kann null sein, wenn die Partition noch nie committed wurde, deshalb filtern wir diese heraus.
• listOffsets mit OffsetSpec.latest() holt den Log-End Offset jeder Partition.
• Jeder Aufruf hat ein Timeout von 10 Sekunden. So wird der Thread nicht unendlich lang blockiert, wenn der Cluster Probleme hat.

Verwendung:

try (Admin admin = Admin.create(Map.of(
        AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"))) {
    new ConsumerLagChecker(admin).lagOf("order-service")
            .forEach((tp, lag) -> System.out.println(tp + " -> " + lag));
}

orders-0 -> 2
orders-1 -> 3430
orders-2 -> 0

4.1. Test für LagCalculator

Mit JUnit 5 und AssertJ (bereits in spring-boot-starter-test enthalten):

class LagCalculatorTest {

    private final TopicPartition p0 = new TopicPartition("orders", 0);
    private final TopicPartition p1 = new TopicPartition("orders", 1);

    @Test
    void shouldComputeLagPerPartition() {
        Map<TopicPartition, Long> lag = LagCalculator.compute(
                Map.of(p0, 100L, p1, 50L),
                Map.of(p0, 120L, p1, 50L));

        assertThat(lag).containsEntry(p0, 20L).containsEntry(p1, 0L);
    }

    @Test
    void shouldClampNegativeLagToZero() {
        Map<TopicPartition, Long> lag = LagCalculator.compute(
                Map.of(p0, 130L),
                Map.of(p0, 120L));

        assertThat(lag).containsEntry(p0, 0L);
    }

    @Test
    void shouldSkipPartitionWithoutEndOffset() {
        Map<TopicPartition, Long> lag = LagCalculator.compute(
                Map.of(p0, 10L), Map.of());

        assertThat(lag).isEmpty();
    }
}

5. Lag über Micrometer in Spring Boot exportieren

Füge Actuator und die Prometheus Registry hinzu:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
    <groupId>io.micrometer</groupId>
    <artifactId>micrometer-registry-prometheus</artifactId>
</dependency>

management.endpoints.web.exposure.include=health,prometheus

5.1. Verfügbare Client-seitige Metriken

Wenn MeterRegistry vorhanden ist, bindet Spring Boot automatisch MicrometerConsumerListener in die ConsumerFactory ein. Alle internen Kafka-Consumer-Metriken erscheinen dann unter /actuator/prometheus, darunter:

kafka_consumer_fetch_manager_records_lag_max{client_id="consumer-order-service-1",...} 3430.0
kafka_consumer_fetch_manager_records_lag{topic="orders",partition="1",...} 3430.0

records-lag-max ist der größte Lag, den dieser spezifische Consumer beobachtet hat. Kafka berechnet ihn aus der Fetch-Position, nicht aus dem Committed Offset.

5.2. Gauge für extern gemessenen Lag

Die client-seitige Metrik hat einen großen Schwachpunkt: Wenn der Consumer abstürzt, verschwindet auch die Metrik. Genau dann, wenn der Lag am schnellsten wächst, sind deine Daten weg. Daher solltest du Lag zusätzlich von einem unabhängigen Prozess messen, indem du AdminClient verwendest:

@Component
public class ConsumerLagMetrics {

    private final ConsumerLagChecker checker;
    private final MeterRegistry registry;
    private final Map<TopicPartition, AtomicLong> gauges = new ConcurrentHashMap<>();

    public ConsumerLagMetrics(KafkaAdmin kafkaAdmin, MeterRegistry registry) {
        this.checker = new ConsumerLagChecker(Admin.create(kafkaAdmin.getConfigurationProperties()));
        this.registry = registry;
    }

    @Scheduled(fixedDelay = 30_000)
    public void refresh() throws Exception {
        checker.lagOf("order-service").forEach((tp, lag) ->
                gauges.computeIfAbsent(tp, this::register).set(lag));
    }

    private AtomicLong register(TopicPartition tp) {
        AtomicLong holder = new AtomicLong();
        Gauge.builder("kafka.consumer.group.lag", holder, AtomicLong::get)
                .tag("group", "order-service")
                .tag("topic", tp.topic())
                .tag("partition", String.valueOf(tp.partition()))
                .register(registry);
        return holder;
    }
}

Denk daran, Scheduling mit @EnableScheduling auf einer @Configuration-Klasse zu aktivieren. Alle 30 Sekunden wird das Gauge kafka_consumer_group_lag aktualisiert. Dieser Wert basiert auf Committed Offsets und bleibt korrekt, selbst wenn alle Consumer der Group gestoppt sind.

In Production verwenden viele Teams spezialisierte Tools wie kafka-exporter oder Burrow statt des Code-Snippets oben. Das Prinzip dahinter ist identisch: Committed Offset mit Log-End Offset vergleichen.

Hinweis: Platziere dieses Gauge nicht im Consumer-Service selbst, wenn dieser mehrere Instanzen hat. Jede Instanz würde denselben Lag-Wert exportieren, und Summen im Dashboard wären falsch. Führe es stattdessen in einem separaten Monitoring-Service aus oder nur auf einer einzigen Instanz.

6. Zu überwachende Metriken

Lag sagt dir, dass es ein Problem gibt. Diese Metriken helfen dir herauszufinden, wo das Problem liegt.

6.1. Broker

Metrik (JMX)	Bedeutung	Gewünschter Wert
UnderReplicatedPartitions	Partitionen mit nicht synchronisierten Replicas	0
OfflinePartitionsCount	Partitionen ohne Leader	0
ActiveControllerCount	Anzahl aktiver Controller (gesamt Cluster)	genau 1
IsrShrinksPerSec / IsrExpandsPerSec	ISR wächst oder schrumpft	nahe 0, keine kontinuierlichen Schwankungen
RequestHandlerAvgIdlePercent	Auslastung der Request-Handler-Threads	> 0.3
NetworkProcessorAvgIdlePercent	Auslastung der Netzwerk-Threads	> 0.3

OfflinePartitionsCount > 0 ist ein kritischer Fehler: diese Partitionen können nicht gelesen oder geschrieben werden. Anhaltend hohe UnderReplicatedPartitions deuten meist auf überladene Broker, langsame Festplatten oder Netzwerkprobleme hin.

6.2. Producer

• record-error-rate: Quote der Sendefehler nach Retry-Ablauf. Jeder Wert > 0 verdient Aufmerksamkeit.
• request-latency-avg: Latenz der Anfragen zum Broker. Wenn das steigt, sinkt der Durchsatz.
• record-retry-rate: Viele Retries bedeuten Broker- oder Netzwerkprobleme.
• buffer-available-bytes: Wenn der Buffer aufgebraucht ist, blockiert send().

6.3. Consumer

• records-lag-max: bereits oben erwähnt.
• records-consumed-rate: Verarbeitungsgeschwindigkeit. Vergleiche mit der Schreibgeschwindigkeit des Producers zum Topic.
• rebalance-rate-per-hour und last-rebalance-seconds-ago: Häufiges Rebalancing führt zu Pausen. Das ist eine häufige Ursache für steigende Lags.
• commit-latency-avg: Langsames Committing reduziert Durchsatz, besonders bei synchronem Commit.
• last-poll-seconds-ago: Wenn dies max.poll.interval.ms überschreitet, wird der Consumer aus der Gruppe geworfen.

7. Alerts sinnvoll konfigurieren

Der häufigste Fehler ist ein absoluter Schwellwert wie »Lag > 1000 warnen«. Bei einem Topic mit 50.000 Nachrichten/Sekunde ist 1000 bedeutungslos. Bei einem Topic mit wenigen Nachrichten pro Minute könnte 1000 bereits Stunden Verzögerung bedeuten.

Zwei bessere Strategien gibt es. Die erste ist ein Alert, wenn Lag kontinuierlich steigt:

groups:
  - name: kafka-consumer
    rules:
      - alert: KafkaConsumerLagGrowing
        expr: sum by (group, topic) (deriv(kafka_consumer_group_lag[10m])) > 0
          and sum by (group, topic) (kafka_consumer_group_lag) > 500
        for: 15m
        labels:
          severity: warning
        annotations:
          summary: "Lag von {{ $labels.group }} auf {{ $labels.topic }} wächst 15 Minuten kontinuierlich"

Diese Regel feuert nur ab, wenn beide Bedingungen erfüllt sind: Lag hat einen wachsenden Trend (deriv > 0) UND hat bereits ein Mindestmaß überschritten. Die Bedingung muss 15 Minuten anhalten. So vermeidest du Fehlalarme nach Deployments durch kurzzeitige Spitzen.

Die zweite Strategie ist, Lag in Zeit umzurechnen. Teile Lag durch die Verarbeitungsgeschwindigkeit, um zu schätzen, wie viele Sekunden der Consumer hinterherhinkt. Diese Zahl ist direkt an SLAs gekoppelt. Zum Beispiel: »Bestellungen müssen innerhalb von 2 Minuten bearbeitet werden« ist verständlicher als »Lag unter 5000 halten«.

Zusätzlich solltest du einen Alert für fehlende Metriken haben, etwa mit absent(kafka_consumer_group_lag). Keine Daten heißt nicht automatisch, dass alles okay ist.

8. Häufige Fehler

• Nur auf records-lag-max verlassen. Wenn der Consumer stirbt, verschwindet diese Metrik. Nutze immer eine externe Quelle zur Lag-Messung.
• Lag einer inaktiven Consumer Group lesen. Die CLI zeigt noch den Committed Offset, aber die Spalte CONSUMER-ID ist leer. Wachsender Lag bedeutet hier, dass niemand verarbeitet.
• Consumer-Instanzen über die Partition-Anzahl hinzufügen. Überschüssige Consumer sitzen herum, Lag sinkt nicht. Um Parallelität zu erhöhen, musst du Partitionen hinzufügen.
• Rebalancing ignorieren. Wenn die Verarbeitung eines Batches länger dauert als max.poll.interval.ms, wird der Consumer rausgeworfen. Die Gruppe rebalanciert, und das Spiel beginnt von vorne. Der Lag wächst sägezahnförmig. Reduziere max.poll.records oder lagere schwere Verarbeitung auf andere Threads aus.
• AdminClient zu häufig aufrufen. Das Abfragen von Offsets jede Sekunde für hunderte Groups erzeugt unnötige Last auf dem Broker. Ein Zyklus alle 15–60 Sekunden reicht für die meisten Systeme.

9. Fazit

Consumer Lag ist die direkteste Metrik dafür, ob dein Kafka-System mit den Daten Schritt hält. Lag wird als Log-End Offset minus Committed Offset berechnet. Du kannst es über kafka-consumer-groups.sh sehen, mit AdminClient berechnen oder über Micrometer zu Prometheus exportieren.

Für effektive Überwachung kombiniere client-seitige Metriken mit einer unabhängigen Messung. Neben Lag solltest du auch Broker-Metriken wie UnderReplicatedPartitions und OfflinePartitionsCount im Auge behalten sowie die Rebalancing-Häufigkeit. Zuletzt: Alerts basierend auf Trends und Verzögerungszeit setzen, nicht auf einem fixen Lag-Wert. So vermeidest du Fehlalarme, ohne echte Probleme zu übersehen.