Spring Kafka: Monitoring & Observability

Production Kafka applications need visibility into consumer lag, throughput, and error rates. Spring Boot with Micrometer and Actuator exposes Kafka metrics out of the box: wire them to Prometheus and Grafana for dashboards and alerts without writing custom JMX code.

Why Monitor Kafka
Micrometer & Actuator Setup
Built-in Kafka Metrics
Consumer Lag Monitoring
Custom Metrics with MeterRegistry
Prometheus & Grafana
Health Indicators
Alerting Best Practices

Why Monitor Kafka

Kafka sits in the critical path of most event-driven architectures. Without monitoring you are flying blind: problems surface only when downstream systems break or users complain. Three categories of metrics matter most:

Consumer lag: the gap between the latest offset produced and the last offset committed by a consumer group. Growing lag means consumers cannot keep up.
Throughput: records produced and consumed per second. A sudden drop signals a failed producer or a stuck consumer.
Error rates: serialization failures, authorization errors, and network timeouts. Even a small spike can cascade.

What can go wrong without monitoring:

Consumer lag grows silently until the retention window expires and messages are lost forever.
A poison-pill record causes a consumer to enter an infinite retry loop while the rest of the partition stalls.
A rebalance storm (consumers repeatedly joining and leaving the group) kills throughput but produces no application-level exception.
Producer batches fail due to a broker leadership change: records pile up in the send buffer until memory runs out.

Kafka metrics are JMX-based under the hood. Spring Boot with Micrometer bridges these JMX MBeans into a unified metrics facade so you never have to deal with JMX directly.

Java

// A simple example: without monitoring, this silent failure goes unnoticed
@Service
public class OrderConsumer {

    @KafkaListener(topics = "orders", groupId = "order-service")
    public void consume(String payload) {
        try {
            orderService.process(payload);
        } catch (Exception e) {
            // Swallowed exception — no metric, no alert, no visibility
            log.error("Failed to process order", e);
        }
    }
}

By the end of this guide every failure like the one above will increment a counter, appear on a dashboard, and fire an alert.

Micrometer & Actuator Setup

Spring Boot Actuator exposes application metrics through the /actuator/metrics endpoint. Micrometer acts as the metrics facade: add the Prometheus registry to export metrics in a format Prometheus can scrape.

XML

<dependencies>
    <!-- Spring Kafka -->
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>

    <!-- Actuator for /actuator endpoints -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-actuator</artifactId>
    </dependency>

    <!-- Micrometer Prometheus registry -->
    <dependency>
        <groupId>io.micrometer</groupId>
        <artifactId>micrometer-registry-prometheus</artifactId>
    </dependency>
</dependencies>

Enable the Prometheus endpoint and Kafka metrics in application.properties:

Properties

# Expose health, metrics, and prometheus endpoints
management.endpoints.web.exposure.include=health,metrics,prometheus

# Enable Kafka consumer and producer metrics via Micrometer
spring.kafka.consumer.properties[metrics.recording.level]=INFO
spring.kafka.producer.properties[metrics.recording.level]=INFO

# Kafka connection
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=order-service
spring.kafka.consumer.auto-offset-reset=earliest

Setting metrics.recording.level to DEBUG captures per-broker and per-topic metrics. This generates a large number of time series: use INFO in production unless you need fine-grained breakdowns.

Once the application starts, verify that Kafka metrics are available:

Properties

# Check available Kafka metrics
curl http://localhost:8080/actuator/metrics | jq '.names[]' | grep kafka

# Fetch a specific metric
curl http://localhost:8080/actuator/metrics/kafka.consumer.fetch.manager.records.consumed.rate

# Prometheus format (for scraping)
curl http://localhost:8080/actuator/prometheus | grep kafka_consumer

Built-in Kafka Metrics

Spring Boot auto-registers Kafka client metrics with Micrometer when Actuator is on the classpath. Here are the key metrics grouped by role.

Consumer Metrics

Metric	Description
`kafka.consumer.fetch.manager.records.consumed.rate`	Records consumed per second across all topics
`kafka.consumer.fetch.manager.records.lag`	Number of records the consumer is behind the broker on a partition
`kafka.consumer.fetch.manager.records.lag.max`	Maximum lag across all assigned partitions
`kafka.consumer.fetch.manager.fetch.latency.avg`	Average time for a fetch request round-trip
`kafka.consumer.coordinator.commit.rate`	Offset commits per second
`kafka.consumer.coordinator.rebalance.rate.per.hour`	Consumer rebalances per hour: high values indicate instability

Producer Metrics

Metric	Description
`kafka.producer.record.send.rate`	Records sent per second
`kafka.producer.record.error.rate`	Records that failed to send per second
`kafka.producer.request.latency.avg`	Average latency for produce requests
`kafka.producer.batch.size.avg`	Average number of bytes per batch
`kafka.producer.buffer.available.bytes`	Available send buffer: drops to zero under backpressure

Java

import io.micrometer.core.instrument.MeterRegistry;
import org.springframework.boot.actuate.autoconfigure.metrics.MeterRegistryCustomizer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class MetricsConfig {

    // Add common tags to every metric for easier filtering in Grafana
    @Bean
    public MeterRegistryCustomizer<MeterRegistry> commonTags() {
        return registry -> registry.config()
                .commonTags("application", "order-service")
                .commonTags("environment", "production");
    }
}

Common tags like application and environment are attached to every metric. This lets you filter dashboards by service or environment without changing individual metric registrations.

Consumer Lag Monitoring

Consumer lag is the single most important Kafka metric. It measures how far behind a consumer group is from the latest produced offset. A lag of zero means the consumer is fully caught up. Growing lag means the consumer cannot process records fast enough: eventually records may fall off the retention window and be lost.

The built-in metric kafka.consumer.fetch.manager.records.lag reports per-partition lag. Spring Boot tags it with client.id and topic so you can drill down.

Properties

# Query lag for a specific topic via Actuator
curl "http://localhost:8080/actuator/metrics/kafka.consumer.fetch.manager.records.lag?tag=topic:orders"

For more control, implement a ConsumerAwareRebalanceListener to track which partitions are assigned to each consumer instance:

Java

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.common.TopicPartition;
import org.springframework.kafka.listener.ConsumerAwareRebalanceListener;
import org.springframework.stereotype.Component;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Tags;

import java.util.Collection;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

@Component
public class LagTrackingRebalanceListener implements ConsumerAwareRebalanceListener {

    private final MeterRegistry meterRegistry;
    private final Set<TopicPartition> assignedPartitions = ConcurrentHashMap.newKeySet();

    public LagTrackingRebalanceListener(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
    }

    @Override
    public void onPartitionsAssigned(Consumer<?, ?> consumer,
                                     Collection<TopicPartition> partitions) {
        assignedPartitions.addAll(partitions);
        partitions.forEach(tp ->
            meterRegistry.gauge("kafka.consumer.partition.assigned",
                Tags.of("topic", tp.topic(), "partition", String.valueOf(tp.partition())),
                1)
        );
        log.info("Partitions assigned: {}", partitions);
    }

    @Override
    public void onPartitionsRevoked(Consumer<?, ?> consumer,
                                    Collection<TopicPartition> partitions) {
        assignedPartitions.removeAll(partitions);
        log.warn("Partitions revoked: {}", partitions);
    }

    public Set<TopicPartition> getAssignedPartitions() {
        return Set.copyOf(assignedPartitions);
    }
}

Java

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;

@Configuration
public class KafkaConsumerConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
            ConsumerFactory<String, String> consumerFactory,
            LagTrackingRebalanceListener rebalanceListener) {

        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.getContainerProperties().setConsumerRebalanceListener(rebalanceListener);
        return factory;
    }
}

Consumer lag reported by the Kafka client is based on the last fetch response. If a consumer is paused or stuck, the lag metric may not update. Use an external tool like Burrow or the Kafka Admin API for broker-side lag checks in critical systems.

Custom Metrics with MeterRegistry

Built-in metrics cover Kafka client internals. Application-level metrics (how many orders were processed, how many failed, how long processing took) require custom instrumentation. Inject MeterRegistry and create counters, timers, and gauges.

Java

import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Timer;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

@Service
public class OrderConsumer {

    private final OrderService orderService;
    private final Counter processedCounter;
    private final Counter failedCounter;
    private final Timer processingTimer;

    public OrderConsumer(OrderService orderService, MeterRegistry meterRegistry) {
        this.orderService = orderService;

        this.processedCounter = Counter.builder("kafka.orders.processed")
                .description("Total orders successfully processed")
                .tag("topic", "orders")
                .register(meterRegistry);

        this.failedCounter = Counter.builder("kafka.orders.failed")
                .description("Total orders that failed processing")
                .tag("topic", "orders")
                .register(meterRegistry);

        this.processingTimer = Timer.builder("kafka.orders.processing.duration")
                .description("Time taken to process each order")
                .tag("topic", "orders")
                .publishPercentileHistogram()
                .register(meterRegistry);
    }

    @KafkaListener(topics = "orders", groupId = "order-service")
    public void consume(String payload) {
        processingTimer.record(() -> {
            try {
                orderService.process(payload);
                processedCounter.increment();
            } catch (Exception e) {
                failedCounter.increment();
                throw e; // Let the error handler deal with retries
            }
        });
    }
}

For a gauge that tracks a dynamic value such as an internal queue depth:

Java

import io.micrometer.core.instrument.MeterRegistry;
import org.springframework.stereotype.Component;
import jakarta.annotation.PostConstruct;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

@Component
public class OrderBuffer {

    private final BlockingQueue<String> queue = new LinkedBlockingQueue<>(10_000);
    private final MeterRegistry meterRegistry;

    public OrderBuffer(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
    }

    @PostConstruct
    public void registerMetrics() {
        // Gauge automatically reads queue.size() on each scrape
        meterRegistry.gauge("kafka.orders.buffer.size",
                queue, BlockingQueue::size);

        meterRegistry.gauge("kafka.orders.buffer.remaining.capacity",
                queue, BlockingQueue::remainingCapacity);
    }

    public void enqueue(String order) {
        queue.offer(order);
    }

    public String dequeue() throws InterruptedException {
        return queue.take();
    }
}

Use publishPercentileHistogram() on timers to get p50, p95, and p99 latency buckets in Prometheus. This adds histogram buckets to the metric but increases cardinality: use it selectively on high-value metrics.

Prometheus & Grafana

With Micrometer exporting metrics in Prometheus format at /actuator/prometheus, configure Prometheus to scrape your application.

YAML

# prometheus.yml — scrape config for Spring Boot app
scrape_configs:
  - job_name: 'order-service'
    metrics_path: '/actuator/prometheus'
    scrape_interval: 15s
    static_configs:
      - targets: ['order-service:8080']
        labels:
          environment: 'production'

Useful PromQL Queries

These queries power most Kafka dashboards. Copy them into Grafana panel queries.

Consumer lag per partition:

PromQL

kafka_consumer_fetch_manager_records_lag{topic="orders"}

Total consumer lag across all partitions (sum by consumer group):

PromQL

sum by (client_id) (kafka_consumer_fetch_manager_records_lag{topic="orders"})

Records consumed per second (rate over 5 minutes):

PromQL

rate(kafka_consumer_fetch_manager_records_consumed_total{topic="orders"}[5m])

Order processing failure rate:

PromQL

rate(kafka_orders_failed_total{topic="orders"}[5m])

P99 processing latency:

PromQL

histogram_quantile(0.99, rate(kafka_orders_processing_duration_seconds_bucket{topic="orders"}[5m]))

Producer error rate:

PromQL

rate(kafka_producer_record_error_total[5m])

Sample Grafana Panel: Consumer Lag

Import this JSON into a Grafana dashboard to create a consumer lag time-series panel:

JSON

{
  "type": "timeseries",
  "title": "Kafka Consumer Lag — orders",
  "datasource": "Prometheus",
  "targets": [
    {
      "expr": "sum by (client_id, partition) (kafka_consumer_fetch_manager_records_lag{topic=\"orders\"})",
      "legendFormat": "{{client_id}} — partition {{partition}}"
    }
  ],
  "fieldConfig": {
    "defaults": {
      "unit": "short",
      "thresholds": {
        "steps": [
          { "value": 0, "color": "green" },
          { "value": 1000, "color": "yellow" },
          { "value": 10000, "color": "red" }
        ]
      }
    }
  },
  "options": {
    "tooltip": { "mode": "multi" },
    "legend": { "displayMode": "table", "placement": "bottom" }
  }
}

Set threshold colors in Grafana to highlight lag severity: green below 1,000, yellow up to 10,000, red above. Adjust thresholds based on your throughput and SLA requirements.

Health Indicators

Spring Boot Actuator includes a health endpoint at /actuator/health. When spring-kafka is on the classpath, Spring Boot automatically registers a KafkaHealthIndicator that checks broker connectivity.

Properties

# Show full health details (UP/DOWN with component breakdown)
management.endpoint.health.show-details=always

# Include Kafka health indicator
management.health.kafka.enabled=true

The default health indicator verifies that the application can describe the Kafka cluster. For deeper checks, like verifying that a consumer group is active and partitions are assigned, create a custom health indicator:

Java

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ConsumerGroupDescription;
import org.apache.kafka.clients.admin.DescribeConsumerGroupsResult;
import org.apache.kafka.common.ConsumerGroupState;
import org.springframework.boot.actuate.health.Health;
import org.springframework.boot.actuate.health.HealthIndicator;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.stereotype.Component;

import java.util.Collections;
import java.util.Map;
import java.util.concurrent.TimeUnit;

@Component("kafkaConsumerGroupHealth")
public class KafkaConsumerGroupHealthIndicator implements HealthIndicator {

    private final KafkaAdmin kafkaAdmin;
    private final String consumerGroup = "order-service";

    public KafkaConsumerGroupHealthIndicator(KafkaAdmin kafkaAdmin) {
        this.kafkaAdmin = kafkaAdmin;
    }

    @Override
    public Health health() {
        try (AdminClient client = AdminClient.create(kafkaAdmin.getConfigurationProperties())) {
            DescribeConsumerGroupsResult result =
                    client.describeConsumerGroups(Collections.singletonList(consumerGroup));

            Map<String, ConsumerGroupDescription> groups =
                    result.all().get(5, TimeUnit.SECONDS);

            ConsumerGroupDescription group = groups.get(consumerGroup);

            if (group == null) {
                return Health.down()
                        .withDetail("consumerGroup", consumerGroup)
                        .withDetail("reason", "Consumer group not found")
                        .build();
            }

            ConsumerGroupState state = group.state();
            int memberCount = group.members().size();

            if (state == ConsumerGroupState.STABLE && memberCount > 0) {
                return Health.up()
                        .withDetail("consumerGroup", consumerGroup)
                        .withDetail("state", state.toString())
                        .withDetail("members", memberCount)
                        .build();
            }

            return Health.down()
                    .withDetail("consumerGroup", consumerGroup)
                    .withDetail("state", state.toString())
                    .withDetail("members", memberCount)
                    .build();

        } catch (Exception e) {
            return Health.down(e)
                    .withDetail("consumerGroup", consumerGroup)
                    .build();
        }
    }
}

A custom health check for broker connectivity that validates topic metadata exists:

Java

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.DescribeClusterResult;
import org.apache.kafka.common.Node;
import org.springframework.boot.actuate.health.Health;
import org.springframework.boot.actuate.health.HealthIndicator;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.stereotype.Component;

import java.util.Collection;
import java.util.concurrent.TimeUnit;

@Component("kafkaBrokerHealth")
public class KafkaBrokerHealthIndicator implements HealthIndicator {

    private final KafkaAdmin kafkaAdmin;
    private final int expectedBrokers;

    public KafkaBrokerHealthIndicator(KafkaAdmin kafkaAdmin) {
        this.kafkaAdmin = kafkaAdmin;
        this.expectedBrokers = 3; // Adjust for your cluster
    }

    @Override
    public Health health() {
        try (AdminClient client = AdminClient.create(kafkaAdmin.getConfigurationProperties())) {
            DescribeClusterResult cluster = client.describeCluster();
            Collection<Node> nodes = cluster.nodes().get(5, TimeUnit.SECONDS);
            String clusterId = cluster.clusterId().get(5, TimeUnit.SECONDS);

            Health.Builder builder = (nodes.size() >= expectedBrokers)
                    ? Health.up()
                    : Health.down().withDetail("reason", "Expected " + expectedBrokers
                            + " brokers but found " + nodes.size());

            return builder
                    .withDetail("clusterId", clusterId)
                    .withDetail("brokerCount", nodes.size())
                    .withDetail("brokers", nodes.stream()
                            .map(n -> n.host() + ":" + n.port())
                            .toList())
                    .build();

        } catch (Exception e) {
            return Health.down(e).build();
        }
    }
}

Custom health indicators that open an AdminClient on every health check can be expensive. Use the management.endpoint.health.cache.time-to-live property (e.g., 10s) to cache the health result and avoid hammering the Kafka cluster.

Properties

# Cache health check results for 10 seconds
management.endpoint.health.cache.time-to-live=10s

Alerting Best Practices

Dashboards show what is happening; alerts tell you when something needs attention. Focus alerts on actionable conditions and avoid noise.

What to Alert On

Consumer lag threshold: lag above a sustained threshold means consumers are falling behind. Alert when total lag exceeds a value for more than 5 minutes.
Error rate spike: a sudden increase in kafka.orders.failed or kafka.producer.record.error.rate indicates a systemic problem.
Consumer rebalance storms: frequent rebalances (more than a few per hour) cause processing pauses and duplicate delivery.
Zero throughput: if records.consumed.rate drops to zero for a topic that normally has traffic, the consumer is likely dead or stuck.
Producer buffer exhaustion: buffer.available.bytes near zero means the producer is blocked waiting for space.

Prometheus Alert Rules

YAML

# prometheus-alerts.yml
groups:
  - name: kafka-alerts
    rules:
      # Consumer lag too high for more than 5 minutes
      - alert: KafkaConsumerLagHigh
        expr: sum by (client_id) (kafka_consumer_fetch_manager_records_lag{topic="orders"}) > 10000
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Kafka consumer lag is high"
          description: "Consumer {{ $labels.client_id }} has lag {{ $value }} on topic orders for more than 5 minutes."

      # Error rate exceeds 1 per second
      - alert: KafkaProcessingErrorRateHigh
        expr: rate(kafka_orders_failed_total[5m]) > 1
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "Kafka order processing error rate is high"
          description: "Error rate is {{ $value }}/s — check application logs for details."

      # Consumer rebalance storm
      - alert: KafkaRebalanceStorm
        expr: kafka_consumer_coordinator_rebalance_rate_per_hour > 10
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "Kafka consumer group is rebalancing excessively"
          description: "Consumer {{ $labels.client_id }} is rebalancing {{ $value }} times/hour."

      # Zero consumer throughput on a normally active topic
      - alert: KafkaConsumerThroughputZero
        expr: rate(kafka_consumer_fetch_manager_records_consumed_total{topic="orders"}[10m]) == 0
        for: 10m
        labels:
          severity: critical
        annotations:
          summary: "Kafka consumer has stopped consuming"
          description: "No records consumed from topic orders for 10 minutes."

Log Correlation with MDC and TraceId

When an alert fires, you need to trace the problem back to specific records. Use MDC (Mapped Diagnostic Context) to inject Kafka metadata into every log line produced during message processing:

Java

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import java.util.UUID;

@Service
public class OrderConsumer {

    private static final Logger log = LoggerFactory.getLogger(OrderConsumer.class);
    private final OrderService orderService;

    public OrderConsumer(OrderService orderService) {
        this.orderService = orderService;
    }

    @KafkaListener(topics = "orders", groupId = "order-service")
    public void consume(ConsumerRecord<String, String> record) {
        // Set MDC fields for log correlation
        MDC.put("traceId", UUID.randomUUID().toString());
        MDC.put("kafkaTopic", record.topic());
        MDC.put("kafkaPartition", String.valueOf(record.partition()));
        MDC.put("kafkaOffset", String.valueOf(record.offset()));
        MDC.put("kafkaKey", record.key());

        try {
            log.info("Processing order");
            orderService.process(record.value());
            log.info("Order processed successfully");
        } catch (Exception e) {
            log.error("Failed to process order", e);
            throw e;
        } finally {
            MDC.clear();
        }
    }
}

Configure the log pattern to include MDC fields:

Properties

# application.properties — include Kafka context in log lines
logging.pattern.console=%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} [traceId=%X{traceId}, topic=%X{kafkaTopic}, partition=%X{kafkaPartition}, offset=%X{kafkaOffset}] - %msg%n

With this configuration every log line produced during Kafka message processing includes the topic, partition, offset, and a unique trace ID. When an alert fires for a high error rate, search logs by traceId to see the full processing context for any failed record.

If you use Spring Cloud Sleuth or Micrometer Tracing, the traceId is set automatically. You can skip the manual UUID.randomUUID() and let the tracing library propagate trace context from Kafka headers using a ProducerInterceptor and ConsumerInterceptor.

Putting It All Together

A production monitoring setup for Spring Kafka follows this pipeline:

Spring Boot exposes Kafka and custom metrics via /actuator/prometheus.
Prometheus scrapes the endpoint every 15 seconds.
Grafana dashboards display consumer lag, throughput, error rates, and processing latency.
Prometheus alert rules fire on sustained lag, error spikes, rebalance storms, and zero throughput.
Alertmanager routes alerts to Slack, PagerDuty, or email based on severity labels.
MDC-enriched logs let you correlate any alert back to specific Kafka records.

Contents

Why Monitor Kafka

Micrometer & Actuator Setup

Built-in Kafka Metrics

Consumer Metrics

Producer Metrics

Consumer Lag Monitoring

Custom Metrics with MeterRegistry

Prometheus & Grafana

Useful PromQL Queries

Sample Grafana Panel: Consumer Lag

Health Indicators

Alerting Best Practices

What to Alert On

Prometheus Alert Rules

Log Correlation with MDC and TraceId

Putting It All Together

Related articles