Apache Kafka is one of the most widely deployed pieces of data infrastructure in the world, and it is a demanding target for static analysis. Its codebase spans a client library, a broker, a storage engine, and the Kafka Streams processing framework, and untrusted data moves between them through binary protocols, compressed record batches, and consumer group coordination.
When I ran Endor Labs' AI SAST engine against Kafka, it surfaced the same dangerous pattern in three separate places. In each one, code reads a size or count field from untrusted input and passes it directly to a memory allocation, with no check against the bytes present in the input. Someone with basic permission to send data into Kafka could send a message just a few dozen bytes long that crashes the applications reading it, and in some common setups the Kafka servers themselves. The problem doesn’t go away on its own: affected systems crash again each time they restart, and one variant quietly stops Kafka’s routine cleanup of its own internal data while the servers keep looking healthy.
This is a good example of AI SAST performing at scale. The engine surfaced the same flaw in three places in one scan by following untrusted data across module boundaries, through compressed record streams, a public deserializer interface, and the group coordination protocol, rather than pattern-matching any single dangerous call.
The Apache Kafka project fixed all three and published two CVEs, both rated important:
Affected versions and fixes
How AI SAST identified them
All three findings came from a single scan on May 1, 2026. For each one, the engine produced a complete source-to-sink data flow, graded the finding against a six-point verification scorecard covering sink, source, flow completeness, sanitization, production reachability, and code exclusions, and assigned a severity. All three passed every criterion and were classified as true positives.
Finding 1: the compressed record decoder
AI SAST output (abridged):
{
"ruleId": "AI SAST: Potential Denial of Service via Excessive Memory Allocation due to Maliciously Sized Record Body",
"message": "This method reads a record body size from an InputStream using ByteUtils.readVarint and
passes the resulting value directly to ByteBuffer.allocate without any upper-bound validation.
An attacker can specify an extremely large size, causing the JVM to attempt a huge heap
allocation. This can trigger an OutOfMemoryError, resulting in denial of service for the
process (Kafka broker or client) handling the record.",
"locations": [{ "physicalLocation": {
"artifactLocation": { "uri": "clients/src/main/java/org/apache/kafka/common/record/internal/DefaultRecord.java" },
"region": { "startLine": 286 }
}}]
}
The engine pinpointed the exact vulnerable line, rated the finding High, and noted a telling detail: a sibling overload of the same method, which decodes from a ByteBuffer, performs validation that this one lacks. That contrast is the bug in one sentence.
Finding 2: ListDeserializer
AI SAST output (abridged):
{
"ruleId": "AI SAST: Potential Denial of Service via excessively large list in deserializeNullIndexList",
"message": "The deserializeNullIndexList function in ListDeserializer is vulnerable to a Denial of Service
attack due to unbounded allocation. An attacker can send a Kafka message with a crafted
nullIndexListSize value, leading to excessive memory allocation and potential OutOfMemoryError.",
"locations": [{ "physicalLocation": {
"artifactLocation": { "uri": "clients/src/main/java/org/apache/kafka/common/serialization/ListDeserializer.java" },
"region": { "startLine": 151 }
}}]
}
The scorecard confirmed reachability through Kafka's public Deserializer interface, which means any consumer configured with ListDeserializer or ListSerde processes attacker-supplied bytes on this path.
Finding 3: Kafka Streams assignment decoding
AI SAST output (abridged):
{
"ruleId": "AI SAST: Potential Denial of Service due to unbounded integer read for named topology characters",
"message": "Because this integer originates from network assignment metadata, a malicious or corrupted
assignment could specify an extremely large value, leading to excessive memory consumption,
resulting in a denial-of-service condition for the Kafka Streams client. The path is executed
during normal group rebalance handling, so it is reachable in production.",
"locations": [{ "physicalLocation": {
"artifactLocation": { "uri": "streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/ConsumerProtocolUtils.java" },
"region": { "startLine": 60 }
}}]
}
This is the trace I find most impressive. Its source is opaque assignment metadata that one consumer group member computes for others, delivered through Kafka's group coordination protocol, passed through the Streams assignor, and decoded several calls deep in Streams internals. The engine followed it across the clients and Streams modules, through the AssignmentInfo decoders that read counts from the stream, and down to the reads they drive.
CVE-2026-114349: one produce request, broker-wide loss of compaction
The bug
DefaultRecord has two readFrom overloads. The ByteBuffer version checks the declared size against the bytes available before allocating. The InputStream version, used for compressed batches, allocates first:
int sizeOfBodyInBytes = ByteUtils.readVarint(input); // attacker-controlled
ByteBuffer recordBuffer = ByteBuffer.allocate(sizeOfBodyInBytes); // line 286
int bytesRead = Utils.readFully(input, recordBuffer);
if (bytesRead != sizeOfBodyInBytes)
throw new InvalidRecordException(...);
The size varint lives inside the compressed data. Kafka's limits on record and request size, and its client quotas, all apply to the compressed bytes on the wire, and the batch CRC is computed by the producer, so a crafted batch is CRC-valid. A batch of about 90 bytes can declare a record body of about 2 GiB.
Where it reaches
Validating the finding meant tracing every caller of the stream decoder. When a batch's compression matches the topic's configuration, the broker validates it in place, skipping over record bodies without allocating them, so the crafted record is accepted, written to disk, and replicated to followers byte for byte. Every component that later decodes the record attempts the allocation:
- Consumers crash on every fetch of the record, along with tools built on the standard consumer, including
kafka-console-consumer.shandkafka-dump-log.sh. - Broker request handlers crash during the produce request itself when a batch's compression differs from the topic's
compression.type, because the broker must re-encode the records. On brokers started with the common hardening flag-XX:+ExitOnOutOfMemoryError, one request terminates the broker JVM. - The log cleaner reads every record on compacted topics to decide which keys to keep. An
OutOfMemoryErrorescapes its error handling and terminates the cleaner thread, which is not restarted. With the default single cleaner thread, compaction stops for every compacted topic on the broker, including the internal__consumer_offsetsand__transaction_statetopics. The condition persists across restarts until the record is removed.
Reproduction
I built the batch by hand, computed its CRC32C, and sent it in a raw Produce request, about 150 lines of Java, against apache/kafka:4.2.0. All three failure modes reproduced. On a compacted topic, the cleaner thread died within seconds:
INFO Cleaner 0: Building offset map for compact-attack-0...
ERROR [kafka-log-cleaner-thread-0]: Error due to
java.lang.OutOfMemoryError: Java heap space
at org.apache.kafka.common.record.DefaultRecord.readFrom(DefaultRecord.java:285)
...
at org.apache.kafka.storage.internals.log.Cleaner.buildOffsetMapForSegment(Cleaner.java:733)
...
INFO [kafka-log-cleaner-thread-0]: Stopped
The broker stayed up and kept passing health checks while compaction stopped. The clearest signal is the kafka.log:type=LogCleaner,name=DeadThreadCount metric, which reports 1 once the thread dies.
CVE-2026-111551: crashing consumers and Kafka Streams applications
ListDeserializer
The method the engine flagged allocates a list sized by the first integer in the record:
private List<Integer> deserializeNullIndexList(final DataInputStream dis) throws IOException {
int nullIndexListSize = dis.readInt(); // line 152
List<Integer> nullIndexList = new ArrayList<>(nullIndexListSize); // line 153
...
}
ArrayList's size constructor allocates its backing array immediately, so a size of 0x7FFFFFFF throws OutOfMemoryError: Requested array size exceeds VM limit before a single element is read. The main deserialize method sizes its result list the same way at line 179, giving the class two allocation sites with the same shape.
A stock KafkaProducer triggers it with a 9-byte record value:
00 7F FF FF FF 00 00 00 00Every consumer that reads the topic with ListDeserializer crashes on its next poll, and again after every restart, because the record stays in the topic. The reach extends to Kafka Streams pipelines with collection-typed values, connectors, and any application wrapping ListSerde.
Kafka Streams assignment decoding
Following the engine's trace into AssignmentInfo led to decodeActiveTasks, where a count read from the assignment sizes a list:
final int count = in.readInt(); // line 389
assignmentInfo.activeTasks = new ArrayList<>(count); // line 390
Four sibling decoders in the same file share the pattern: decodeStandbyTasks, decodePartitionsByHost, readTopicPartitions, and decodeTopicIndexAndGet.
The bytes come from the consumer group leader. In Kafka's group protocol, the coordinator elects one member as leader, and the leader's assignor computes every member's assignment. Kafka Streams registers its assignor under the name stream, and a group only uses an assignor every member supports. A plain Kafka consumer, with no Streams runtime, can register its own assignor named stream, join a Streams application's consumer group, and, once elected leader, return any bytes it likes as every member's assignment. The only access required is the Group:Read ACL on that group.
My reproducer used a 60-line custom assignor returning 12 bytes for every member: assignment version 11, supported version 11, and an active task count of 0x7FFFFFFF.
00 00 00 0B 00 00 00 0B 7F FF FF FFEvery stream thread in the victim application crashed during the rebalance:
java.lang.OutOfMemoryError: Requested array size exceeds VM limit
at java.base/java.util.ArrayList.<init>(Unknown Source)
at org.apache.kafka.streams.processor.internals.assignment.AssignmentInfo.decodeActiveTasks(AssignmentInfo.java:390)
...
at org.apache.kafka.streams.processor.internals.StreamsPartitionAssignor.onAssignment(StreamsPartitionAssignor.java:1464)
...
at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:912)
Because the attacker stays in the group, each restart triggers another rebalance and another crash. The attack worked with both the eager and cooperative rebalance protocols. This path matters most in multi-tenant clusters, where consumer group ACLs are often granted by wildcard across application namespaces.
Four more paths
The Kafka maintainers extended the CVE-2026-111551 fix to four more Kafka Streams deserialization paths with the same defect: ProcessorRecordContext (header count), ChangedDeserializer (varint value length), CombinedKeySchema (foreign-key length, which could also produce a negative size), and ProcessorMetadata (key length). These read from a Streams application's internal changelog and repartition topics and from offset metadata committed to its consumer group, so the advisory treats writing to those as an attack surface too.
Why size limits do not help
The common thread across both CVEs is that the allocation size comes from a field inside the input, while every limit measures the input's own length. The payloads are 90 bytes, 9 bytes, and 12 bytes. max.message.bytes, request size limits, and quotas all measure the input, so none of them bounds an allocation the input merely describes. The fix in every case is to check the declared size against the input that is present, before allocating: a list of N four-byte integers cannot fit in fewer than 4N bytes.
Remediation
- Upgrade to Apache Kafka 4.2.2, 4.3.2, or 4.4.0. This addresses both CVEs.
- Set
max.decompressed.message.bytesat the broker or topic level. Upgrading alone does not protect against CVE-2026-114349, because this setting defaults to an effectively unlimited value of about 2 GiB for backward compatibility. Choose a value at or slightly above the largest legitimate decompressed record your applications produce, and comfortably below the broker heap divided by the number of request-handler and cleaner threads. - Restrict write and group access. Until you can upgrade, limit
Writeaccess to trusted producers. Restrict write access to each Streams application's internal topics and consumer group to that application's own principal, and restrict who can produce to topics consumed withListSerde. Quotas andmax.message.bytesdo not mitigate either issue. - Alert on
kafka.log:type=LogCleaner,name=DeadThreadCountgreater than zero, so a stopped cleaner is caught before internal compacted topics grow without bound.
If a crafted record was written to a topic before the limit was set, treat it as a hazard for consumers until it is removed or ages out of the log.
Disclosure
I reported the findings to the Apache Kafka security team with end-to-end reproducers for every failure mode on May 15, 2026. The maintainers fixed the record decoder and added the max.decompressed.message.bytes limit under CVE-2026-114349, combined the deserializer findings and extended their fix to four more Kafka Streams paths under CVE-2026-111551, and published both advisories on October 28, 2026.
Takeaway
Unbounded allocation from an untrusted size field is a simple bug to describe and a hard one to find at scale. The size field never looks dangerous where it is read, and the allocation never looks dangerous where it happens. The risk only appears when you connect the two, and in Kafka the connection runs through compressed record streams, the public deserializer interface, and the consumer group protocol.
That connection is what Endor Labs' AI SAST made. In one scan, it traced untrusted bytes from three different entry points to three allocation sites across the clients and Streams modules, verified each flow against its scorecard, and pointed at the code paths the Kafka maintainers went on to fix. From those findings came two CVEs, six fixed Kafka Streams and client deserialization paths, a hardened record decoder, and a new broker setting that lets operators bound decompressed record size.
What's next?
When you're ready to take the next step in securing your software supply chain, here are 3 ways Endor Labs can help:
.avif)
