tutorial

How to Monitor Kafka Streams Applications with Vigilmon

Kafka Streams apps can silently stall on rebalances, processing exceptions, or thread deaths without any HTTP errors. Learn how to expose a health endpoint, track processing lag, and use heartbeat monitoring to detect dead stream topologies with Vigilmon.

Kafka Streams is a lightweight Java library for building real-time stream processing applications that run inside your own JVM process — no separate cluster needed. But that simplicity comes with a monitoring blind spot: when a Kafka Streams thread dies, a topology enters an ERROR state, or consumer lag climbs undetected, your processing simply stops. There are no HTTP 500 errors and no alerts from Kafka itself. Events pile up in the input topic, and downstream consumers get nothing.

Vigilmon adds external observability to Kafka Streams applications through HTTP health endpoint monitoring and heartbeat monitoring for processing threads. This tutorial shows you how to expose stream health metrics, integrate with Vigilmon, and get alerted before silent stalls become data outages.

Why Kafka Streams Needs External Monitoring

Kafka Streams runs as a library embedded in your application process. Internal observability options include JMX metrics and the KafkaStreams.state() API — but these only help if you're actively polling them. External monitoring with Vigilmon adds:

  • Proactive alerting when the Kafka Streams instance leaves the RUNNING state
  • Thread-level health detection — a single dead stream thread can silently drop a partition's processing
  • Processing lag monitoring via consumer group offset tracking
  • Heartbeat liveness checks so you know immediately when a processor stops emitting output records, even if the thread is technically alive
  • Multi-region probing from outside your VPC or Kubernetes cluster

These layers are complementary: a streams app can appear healthy at the JVM level while silently sitting in a REBALANCING loop or dropping records due to a deserialization exception.

Step 1: Expose a Kafka Streams Health Endpoint

Kafka Streams does not expose an HTTP endpoint by default. You need to add one to your application. The KafkaStreams object provides a state() method that returns the current lifecycle state.

Spring Boot Application

If you are running Kafka Streams inside a Spring Boot application, add a health indicator:

// KafkaStreamsHealthIndicator.java
import org.apache.kafka.streams.KafkaStreams;
import org.springframework.boot.actuate.health.Health;
import org.springframework.boot.actuate.health.HealthIndicator;
import org.springframework.stereotype.Component;

@Component
public class KafkaStreamsHealthIndicator implements HealthIndicator {

    private final KafkaStreams kafkaStreams;

    public KafkaStreamsHealthIndicator(KafkaStreams kafkaStreams) {
        this.kafkaStreams = kafkaStreams;
    }

    @Override
    public Health health() {
        KafkaStreams.State state = kafkaStreams.state();
        if (state == KafkaStreams.State.RUNNING) {
            return Health.up()
                .withDetail("state", state.name())
                .withDetail("threads", kafkaStreams.metadataForLocalThreads().size())
                .build();
        }
        return Health.down()
            .withDetail("state", state.name())
            .build();
    }
}

With Spring Boot Actuator, this exposes /actuator/health with Kafka Streams state included. For a dedicated endpoint:

// StreamsHealthController.java
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.ThreadMetadata;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

@RestController
public class StreamsHealthController {

    private final KafkaStreams kafkaStreams;

    public StreamsHealthController(KafkaStreams kafkaStreams) {
        this.kafkaStreams = kafkaStreams;
    }

    @GetMapping("/health/streams")
    public ResponseEntity<Map<String, Object>> health() {
        KafkaStreams.State state = kafkaStreams.state();
        Set<ThreadMetadata> threads = kafkaStreams.metadataForLocalThreads();

        long activeThreads = threads.stream()
            .filter(t -> t.threadState().equals("RUNNING"))
            .count();

        Map<String, Object> body = Map.of(
            "state", state.name(),
            "totalThreads", threads.size(),
            "activeThreads", activeThreads
        );

        if (state == KafkaStreams.State.RUNNING && activeThreads == threads.size()) {
            return ResponseEntity.ok(body);
        }
        return ResponseEntity.status(503).body(body);
    }
}

Standalone Java Application (Javalin)

For a Kafka Streams app that is not Spring Boot, add a minimal HTTP server:

// HealthServer.java
import io.javalin.Javalin;
import org.apache.kafka.streams.KafkaStreams;

import java.util.Map;

public class HealthServer {

    public static void start(KafkaStreams streams, int port) {
        Javalin app = Javalin.create().start(port);

        app.get("/health/streams", ctx -> {
            KafkaStreams.State state = streams.state();
            Map<String, Object> response = Map.of(
                "state", state.name(),
                "running", state == KafkaStreams.State.RUNNING
            );

            if (state == KafkaStreams.State.RUNNING) {
                ctx.status(200).json(response);
            } else {
                ctx.status(503).json(response);
            }
        });
    }
}

Then call HealthServer.start(streams, 8090) after streams.start().

Step 2: Add Processing Lag Detection

A Kafka Streams app in RUNNING state can still be far behind on its input topics. Add a lag-aware health check using the Kafka AdminClient:

// LagHealthEndpoint.java
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsResult;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.http.ResponseEntity;

import java.util.*;
import java.util.concurrent.ExecutionException;

@RestController
public class LagHealthEndpoint {

    private final AdminClient adminClient;
    private final String consumerGroupId;
    private final long lagThreshold;

    public LagHealthEndpoint(AdminClient adminClient) {
        this.adminClient = adminClient;
        this.consumerGroupId = System.getenv("STREAMS_APP_ID");
        this.lagThreshold = Long.parseLong(System.getenv().getOrDefault("LAG_THRESHOLD", "5000"));
    }

    @GetMapping("/health/streams/lag")
    public ResponseEntity<Map<String, Object>> lagHealth() throws ExecutionException, InterruptedException {
        ListConsumerGroupOffsetsResult offsetsResult = adminClient.listConsumerGroupOffsets(consumerGroupId);
        Map<TopicPartition, OffsetAndMetadata> committedOffsets = offsetsResult.partitionsToOffsetAndMetadata().get();

        Map<TopicPartition, Long> endOffsets = adminClient.listOffsets(
            committedOffsets.entrySet().stream()
                .collect(java.util.stream.Collectors.toMap(
                    Map.Entry::getKey,
                    e -> org.apache.kafka.clients.admin.OffsetSpec.latest()
                ))
        ).all().get().entrySet().stream()
            .collect(java.util.stream.Collectors.toMap(Map.Entry::getKey, e -> e.getValue().offset()));

        long maxLag = 0;
        for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : committedOffsets.entrySet()) {
            Long endOffset = endOffsets.get(entry.getKey());
            if (endOffset != null) {
                long lag = endOffset - entry.getValue().offset();
                maxLag = Math.max(maxLag, lag);
            }
        }

        Map<String, Object> body = Map.of("maxLag", maxLag, "threshold", lagThreshold);

        if (maxLag > lagThreshold) {
            return ResponseEntity.status(503).body(
                Map.of("status", "degraded", "reason", "consumer_lag_exceeded", "maxLag", maxLag)
            );
        }
        return ResponseEntity.ok(Map.of("status", "ok", "maxLag", maxLag));
    }
}

Step 3: Configure Vigilmon HTTP Monitors

  1. Log in to vigilmon.online and go to Monitors → New Monitor
  2. Choose HTTP / HTTPS
  3. Set the URL to your streams health endpoint: https://your-app.example.com/health/streams
  4. Set the check interval to 1 minute
  5. Under Expected response, configure:
    • Status code: 200
    • Response body contains: "state":"RUNNING"
    • Response time threshold: 3000ms
  6. Under Alert channels, assign your Slack or PagerDuty channel
  7. Save the monitor

Add a second monitor for lag:

  • URL: https://your-app.example.com/health/streams/lag
  • Expected: 200, body contains "status":"ok"
  • Interval: 1 minute
  • Alert channel: separate Slack channel (lag is typically P2, state errors are P1)

Vigilmon checks from multiple regions and only fires when the majority of probes agree — preventing false alarms from transient rebalances that resolve in seconds.

Step 4: Heartbeat Monitoring for Stream Processors

Thread state and consumer lag tell you the infrastructure is healthy. But a Kafka Streams topology can be in RUNNING state while a processor node silently discards records due to a deserialization exception, a null pointer in your ValueMapper, or a downstream emit failure.

Vigilmon heartbeat monitors catch this: your processor sends a ping after each batch of successfully processed records. If pings stop, Vigilmon fires an alert.

Set Up the Heartbeat Monitor

  1. In Vigilmon, go to Monitors → New Monitor → Heartbeat
  2. Set the name: kafka-streams-order-processor
  3. Set the expected interval: 5 minutes
  4. Set the grace period: 10 minutes
  5. Save — copy the heartbeat URL: https://vigilmon.online/heartbeat/abc123xyz

Wire Heartbeat Into Your Topology

Add a foreach terminal node after your main processing to emit heartbeats:

// StreamTopology.java
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;

import java.net.HttpURLConnection;
import java.net.URL;
import java.util.concurrent.atomic.AtomicInteger;

public class StreamTopology {

    private static final String HEARTBEAT_URL = System.getenv("VIGILMON_HEARTBEAT_URL");
    private static final AtomicInteger recordCount = new AtomicInteger(0);
    private static final int HEARTBEAT_EVERY = 100;

    public static void build(StreamsBuilder builder) {
        KStream<String, Order> orders = builder.stream("orders");

        orders
            .filter((key, value) -> value != null)
            .mapValues(StreamTopology::enrichOrder)
            .foreach((key, value) -> {
                processOrder(key, value);

                // Ping Vigilmon every N records
                if (recordCount.incrementAndGet() % HEARTBEAT_EVERY == 0) {
                    pingVigilmon();
                }
            });
    }

    private static void pingVigilmon() {
        try {
            HttpURLConnection conn = (HttpURLConnection) new URL(HEARTBEAT_URL).openConnection();
            conn.setRequestMethod("GET");
            conn.setConnectTimeout(3000);
            conn.setReadTimeout(3000);
            conn.getResponseCode();
            conn.disconnect();
        } catch (Exception ignored) {
            // Heartbeat failure must never kill stream processing
        }
    }
}

For low-throughput topologies, send heartbeats on a scheduled timer instead:

// ScheduledHeartbeat.java
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.time.Instant;
import java.util.concurrent.atomic.AtomicReference;

public class ScheduledHeartbeat {

    private static final AtomicReference<Instant> lastProcessed = new AtomicReference<>(Instant.now());

    public static void start(String heartbeatUrl) {
        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
        scheduler.scheduleAtFixedRate(() -> {
            Instant last = lastProcessed.get();
            // Only ping if we processed something in the last 5 minutes
            if (last != null && Instant.now().minusSeconds(300).isBefore(last)) {
                pingVigilmon(heartbeatUrl);
            }
        }, 60, 60, TimeUnit.SECONDS);
    }

    public static void recordProcessed() {
        lastProcessed.set(Instant.now());
    }
}

Step 5: Alert Routing for Stream Processing Failures

Configure alert routing in Vigilmon based on failure severity:

| Monitor | Alert Channel | Priority | |---|---|---| | /health/streams (state not RUNNING) | Slack + PagerDuty | P1 | | /health/streams/lag (lag exceeded) | Slack | P2 | | Heartbeat: order-processor | Slack + email | P2 | | Heartbeat: analytics-processor | Email | P3 |

Set response time thresholds for early warning — a slow health endpoint response (>2000ms) can indicate GC pressure or thread contention before a full stall occurs.

For critical topologies, set up two heartbeat monitors with different intervals: a lenient one (grace period 10 minutes) for P2 paging, and a strict one (grace period 2 minutes) for immediate P1 escalation.

Summary

Kafka Streams applications fail silently. External monitoring with Vigilmon gives you the visibility layer that embedded JMX metrics alone cannot provide:

| Monitor Type | What It Covers | |---|---| | HTTP monitor on /health/streams | KafkaStreams lifecycle state, thread count | | HTTP monitor on /health/streams/lag | Consumer group lag threshold breaches | | Heartbeat monitor | Processor liveness, record throughput proof |

Get started free at vigilmon.online — your first Kafka Streams monitor is live in under two minutes.

Monitor your app with Vigilmon

Free plan — 5 monitors, no credit card required. Up and running in 60 seconds.

Start free →
← Back to all articles