Kafka consumer group JVM heap exhaustion during peak tensor feature streaming

Problem Description

During peak AI inference traffic, Kafka consumer pods in a Kubernetes cluster begin to terminate with the following errors:

java.lang.OutOfMemoryError: Java heap space
    at org.apache.kafka.clients.consumer.internals.Fetcher.fetchRecords(Fetcher.java:267)
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1152)
    ...
java.lang.OutOfMemoryError: Direct buffer memory
    at java.nio.Bits.reserveMemory(Bits.java:698)
    at java.nio.DirectByteBuffer.(DirectByteBuffer.java:112)
    at org.apache.kafka.common.record.MemoryRecordsBuilder.(MemoryRecordsBuilder.java:124)
    ...
[Consumer clientId=consumer-1, groupId=feature-stream] Failed to allocate memory for record batch

Operational impact includes:

• Consumer group stalls, causing inference latency spikes.
• Frequent full GC cycles (“Full GC (Allocation Failure) … total heap size 8G, used 7.9G”).
• Horizontal Pod Autoscaler (HPA) fails to scale because pods are repeatedly OOM‑killed.
• Rebalance storms triggered by session timeout expirations.

Root Cause Analysis

The exhaustion originates from two coupled mechanisms:

1. Unbounded fetch buffer growth. The default fetch.max.bytes (50 MiB) and max.partition.fetch.bytes (1 MiB) allow the consumer to request large batches per partition. When tensor feature messages reach tens of megabytes (common for image or video embeddings), a single poll can pull multiple such records, inflating the internal FetchResponse buffer. The consumer stores these batches in a per‑partition cache until they are deserialized and handed to the application. If the application processing time exceeds the poll interval, the cache grows unchecked, consuming heap and direct memory.
2. Direct memory allocation via Netty/Kafka client. The Kafka client uses ByteBuffer.allocateDirect for zero‑copy network I/O. Large record batches trigger allocations that exceed the JVM’s -XX:MaxDirectMemorySize (default is same as -Xmx). Once the limit is hit, the client throws java.lang.OutOfMemoryError: Direct buffer memory, as observed in the GitHub issue kafka-incubator/kafka-streams-examples #78.

Both effects are amplified during inference spikes because:

• Tensor payloads are larger than typical log or telemetry messages.
• Back‑pressure is absent; the consumer continues to poll at the configured max.poll.interval.ms, leading to “fetch‑fetch‑fetch” loops.
• Session timeout settings (e.g., session.timeout.ms) are too short relative to processing latency, causing frequent rebalances that reset the fetch buffers, as seen in the health‑tech startup incident.

Investigation and Debugging

Step‑by‑step diagnostics that proved effective in the fintech AI platform incident:

1. Log inspection. Grep consumer logs for OOM patterns and fetch failures:

kubectl logs -l app=feature-consumer -c consumer | grep -E "OutOfMemoryError|Failed to allocate memory"

2. Heap dump analysis. Capture a heap dump on OOM and examine the org.apache.kafka.clients.consumer.internals.PartitionRecords instances:

jmap -dump:live,format=b,file=heap.hprof $(pgrep -f KafkaConsumer)

Heap histogram shows PartitionRecords occupying >2 GB, confirming unbounded cache growth.

3. Direct memory monitoring. Enable -XX:+PrintDirectMemoryUsage and review GC logs:

2026-07-31T12:45:23.123+0000: [Full GC (Allocation Failure) 2026-07-31T12:45:23.123+0000: [Direct Memory: 7.8G used, 8.0G max]

4. Network capture. Use tcpdump on the consumer pod to verify batch sizes:

kubectl exec -it consumer-0 -- tcpdump -i any -s 0 -w /tmp/capture.pcap port 9092

Wireshark analysis shows FetchResponse frames of 45 MiB, far exceeding the default limits.

5. Configuration audit. Dump effective consumer configs:

kubectl exec consumer-0 -- java -cp consumer.jar com.example.ConfigDump
# Output excerpt
max.partition.fetch.bytes = 1048576
fetch.max.bytes = 52428800
max.poll.records = 500
max.poll.interval.ms = 300000
session.timeout.ms = 10000

6. Resource limits verification. Review pod spec:

apiVersion: v1
kind: Pod
metadata:
  name: consumer-0
spec:
  containers:
  - name: consumer
    image: consumer:latest
    resources:
      requests:
        memory: "4Gi"
      limits:
        memory: "8Gi"

The container limit (8 Gi) matches -Xmx8g, but direct memory also competes for this quota, leaving insufficient headroom.

Resolution

The fix combines consumer configuration tuning, JVM flag adjustments, and Kubernetes resource updates.

Before

# application.yml (original)
spring.kafka.consumer.max-poll-records: 500
spring.kafka.consumer.max-poll-interval-ms: 300000
spring.kafka.consumer.max-partition-fetch-bytes: 1048576
spring.kafka.consumer.fetch-max-bytes: 52428800
# JVM flags
JAVA_OPTS="-Xms8g -Xmx8g"

After

# application.yml (tuned)
spring.kafka.consumer.max-poll-records: 100          # smaller batches
spring.kafka.consumer.max-poll-interval-ms: 600000   # give processing more time
spring.kafka.consumer.max-partition-fetch-bytes: 5242880   # 5 MiB per partition
spring.kafka.consumer.fetch-max-bytes: 20971520          # 20 MiB overall
spring.kafka.consumer.session.timeout.ms: 30000         # align with processing latency
spring.kafka.consumer.enable-auto-commit: false
spring.kafka.consumer.max-poll-records: 100

# Enable explicit back‑pressure
spring.kafka.listener.concurrency: 4
spring.kafka.listener.ack-mode: manual_immediate

# JVM flags (increase direct memory, reduce heap pressure)
JAVA_OPTS="-Xms4g -Xmx4g -XX:MaxDirectMemorySize=6g -XX:+UseG1GC -XX:InitiatingHeapOccupancyPercent=35"

# Kubernetes pod spec (updated resources)
apiVersion: v1
kind: Pod
metadata:
  name: consumer-0
spec:
  containers:
  - name: consumer
    image: consumer:latest
    resources:
      requests:
        memory: "6Gi"
        cpu: "2000m"
      limits:
        memory: "8Gi"
        cpu: "4000m"

Key changes explained:

• Reduced fetch sizes. Limiting max.partition.fetch.bytes to 5 MiB prevents a single partition from monopolizing the fetch buffer.
• Smaller poll batch. max.poll.records set to 100 keeps the in‑memory record list manageable.
• Increased direct memory. -XX:MaxDirectMemorySize=6g separates off‑heap allocation from the heap, avoiding “Direct buffer memory” OOM while preserving heap for object deserialization.
• Adjusted heap. Reducing -Xmx to 4 GiB gives the GC more headroom to operate efficiently and aligns with the new requests.memory value.
• Back‑pressure via manual commits. The consumer now pauses after processing a batch and commits offsets only after successful inference, preventing uncontrolled fetch loops.
• Session timeout alignment. Raising session.timeout.ms to 30 s avoids premature rebalances during long inference calls.

Validation

After applying the changes, the following checks confirmed resolution:

1. Pod stability. kubectl get pods -l app=feature-consumer shows all pods Running with no restarts over a 2‑hour peak window.
2. Heap and direct memory usage. JMX metrics (via Prometheus) show:
   

   Metric	Before	After
   Heap Used (GiB)	7.9	2.3
   Direct Memory Used (GiB)	6.5	1.1
   Full GC Count (per hour)	12	1

3. Consumer lag. kafka-consumer-groups.sh --describe --group feature-stream reports lag < 100 records per partition, well within SLA.
4. Inference latency. End‑to‑end latency dropped from 850 ms (peak) to 210 ms, matching expected performance.
5. HPA scaling. Autoscaler successfully added two extra consumer pods during the next traffic spike, confirming that OOM no longer blocks scaling.

Prevention and Best Practices

• Set max.partition.fetch.bytes to a value comfortably larger than the typical tensor payload but far smaller than the maximum message size.
• Keep fetch.max.bytes no more than max.partition.fetch.bytes * number_of_partitions_per_consumer.
• Use manual offset commits and pause()/resume() to apply back‑pressure when processing time exceeds max.poll.interval.ms.
• Allocate direct memory explicitly with -XX:MaxDirectMemorySize and monitor DirectMemoryUsed via JMX.
• Configure Kubernetes resource requests to reflect both heap and off‑heap needs; avoid setting container memory limits equal to -Xmx alone.
• Enable G1GC or ZGC for low‑pause collections on large heaps; tune InitiatingHeapOccupancyPercent to trigger concurrent cycles earlier.
• Instrument consumer lag and memory metrics; trigger alerts when heap usage exceeds 75 % or direct memory exceeds 80 % of its limit.
• Regularly test with synthetic payloads matching the largest expected tensor size to validate fetch configuration.

Related Topic Hub: Data Infrastructure Troubleshooting Hub

FAQ

1. Why does increasing max.poll.records worsen OOM? Larger max.poll.records means more records are retained in memory per poll. With multi‑megabyte tensors, each record can consume tens of megabytes of heap and direct memory, quickly exhausting the JVM.
2. Can I keep fetch.max.bytes high and still avoid OOM? Only if you also limit max.poll.records and implement back‑pressure. The fetch size controls network transfer; the poll batch size controls in‑memory retention.
3. Is increasing -Xmx a safe fix? Not in Kubernetes where container memory limits are enforced. Raising -Xmx without adjusting the pod’s resources.limits.memory leads to OOM kills by the kubelet, not the JVM.
4. How do I monitor direct buffer usage? Enable -XX:+PrintDirectMemoryUsage and expose the sun.misc.VM.directMemory MBean. Prometheus JMX exporter can scrape java.lang:type=MemoryPool,name=DirectMemoryPool.
5. What role does session.timeout.ms play in this scenario? If the consumer cannot finish processing before the session timeout, the broker marks it dead, triggering a rebalance. Rebalances discard fetch buffers, causing additional fetch cycles and memory churn.