Leonard Solvik Lisoey created KAFKA-20841:
---------------------------------------------
Summary: CoordinatorLoaderImpl can read past bounded FileRecords
slices with preallocated logs, causing CorruptRecordException and FAILED
group-coordinator shards
Key: KAFKA-20841
URL: https://issues.apache.org/jira/browse/KAFKA-20841
Project: Kafka
Issue Type: Bug
Components: group-coordinator
Affects Versions: 4.2.0
Environment: Confluent Platform 8.2.1 (Apache Kafka 4.2 line), KRaft
mode. Broker banner "Kafka version: 8.2.1-ccs", commit
943bf3de3264ec138d5d3025ce3816486d3351ab. The relevant source files are
identical to Apache Kafka 4.2.0; the same missing slice limit is present on
trunk 2699fc40748da5ce9dc4b6aac41687fbc603ab1d (2026-07-24).
log.preallocate=true (non-default); offsets.topic.segment.bytes=104857600; 50
__consumer_offsets partitions; RF=3; offsets.load.buffer.size=5242880 (default).
OpenJDK 23.0.2; Linux 6.8.0-134-generic; log filesystem XFS on NVMe.
Standalone reproduction: kafka-clients-4.2.0 from Maven Central, OpenJDK 21,
macOS/APFS.
Reporter: Leonard Solvik Lisoey
Attachments: ReproTest.java
h2. Summary
{{CoordinatorLoaderImpl.toReadableMemoryRecords}} can read beyond the logical
end of the bounded {{FileRecords}} returned by {{LogSegment.read}}. The loader
obtains {{fileRecords.sizeInBytes()}}, but allocates or reuses a buffer of at
least {{loadBufferSize}}, clears it to full capacity, and calls
{{FileRecords.readInto(buffer, 0)}} without limiting the destination to the
logical slice size. {{readInto}} delegates to {{Utils.readFully}}, which reads
from the underlying channel until the buffer is full or physical EOF — the
slice's logical end is never enforced.
With {{log.preallocate=true}}, a freshly rolled active segment has physical
extent beyond its logical end, and the excess region is zero-readable. When
the logical slice is smaller than the destination buffer, the copy can
continue past the slice end into that region; {{MemoryRecords}} iteration then
interprets the zeros as a batch header and throws:
{code}
CorruptRecordException: Record size 0 is less than the minimum record overhead
(14)
{code}
even when the logical records in the bounded slice are valid. In the 4.x
coordinator runtime this failure transitions the shard to FAILED. No new
load is scheduled at the same partition epoch; a higher leader epoch or a
process restart creates a new load attempt — and an unremediated replica
can fail that attempt again. Meanwhile {{GroupCoordinatorService.listGroups}}
converts the failed shard's {{NotCoordinatorException}} into an empty list,
so the affected groups silently disappear from
{{kafka-consumer-groups --list}}.
This appears closely related to KAFKA-13664, which reported the same
preallocation-triggered exception and an over-read into zero-readable file
space in the legacy coordinator. In the 4.x runtime, the additional
observed consequence is a same-epoch-sticky FAILED shard, together with
the ListGroups behavior described above.
h2. Environment
* Observed on Confluent Platform 8.2.1 (Apache Kafka 4.2 line), KRaft mode
* Observed broker banner: {{Kafka version: 8.2.1-ccs}}, commit
{{943bf3de3264ec138d5d3025ce3816486d3351ab}}
* The relevant source files in that build were compared with Apache Kafka 4.2.0
and found identical; the same missing slice limit is present on Apache trunk at
commit {{2699fc40748da5ce9dc4b6aac41687fbc603ab1d}} (2026-07-24)
* {{log.preallocate=true}} (non-default),
{{offsets.topic.segment.bytes=104857600}}, 50 {{__consumer_offsets}}
partitions, RF=3, default {{offsets.load.buffer.size}} (5242880)
* Java version: OpenJDK 23.0.2; OS/kernel: Linux 6.8.0-134-generic; log
filesystem: XFS (NVMe)
* Broker stop mode during the observed recoveries: graceful (systemd)
h2. Expected vs actual behavior
{code}
Expected:
CoordinatorLoaderImpl copies and parses only the bytes inside the bounded
FileRecords result.
Actual:
When the destination buffer is larger than the bounded result, readInto can
continue beyond the slice end to physical EOF or the buffer limit, causing
preallocated zeros to be parsed as another record batch.
{code}
h2. Mechanism (source references)
Permalinks (Apache Kafka 4.2.0 unless noted):
* [CoordinatorLoaderImpl.java @
4.2.0|https://github.com/apache/kafka/blob/4.2.0/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorLoaderImpl.java]
* [CoordinatorLoaderImpl.java @ trunk
2699fc40|https://github.com/apache/kafka/blob/2699fc40748da5ce9dc4b6aac41687fbc603ab1d/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorLoaderImpl.java]
* [FileRecords.java @
4.2.0|https://github.com/apache/kafka/blob/4.2.0/clients/src/main/java/org/apache/kafka/common/record/FileRecords.java]
* [ByteBufferLogInputStream.java @
4.2.0|https://github.com/apache/kafka/blob/4.2.0/clients/src/main/java/org/apache/kafka/common/record/ByteBufferLogInputStream.java]
* [CoordinatorRuntime.java @
4.2.0|https://github.com/apache/kafka/blob/4.2.0/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java]
# {{CoordinatorLoaderImpl.doLoad}} reads with {{log.read(currentOffset,
loadBufferSize, FetchIsolation.LOG_END, true)}} and converts the result via
{{toReadableMemoryRecords}}.
# In the conversion, {{bytesNeeded = Math.max(loadBufferSize, sizeInBytes)}};
the buffer is allocated or reused, {{buffer.clear()}} restores limit =
capacity, and {{fileRecords.readInto(buffer, 0)}} is called. No
{{buffer.limit(sizeInBytes)}} is applied in either branch.
# {{FileRecords.readInto}} → {{Utils.readFully(channel, buffer, position)}}
loops until the buffer has no remaining capacity or the channel hits EOF. The
{{FileRecords}} slice's {{end}} is not consulted.
# {{ByteBufferLogInputStream.nextBatchSize}} reads the size field of the next
"batch" in the copied zeros and throws the CorruptRecordException above.
# Segment lifecycle explains why the failure is intermittent:
{{FileRecords.close()}} performs {{flush(); trim()}}, and
{{LogSegment.onBecomeInactiveSegment()}} trims on roll — so inactive and
gracefully-closed files carry no zero tail. A directly vulnerable geometry is a
*newly rolled, preallocated active segment* combined with a destination buffer
whose remaining capacity exceeds the logical slice size (a final logical slice
smaller than the destination buffer is the common case; a reused buffer
previously enlarged past {{loadBufferSize}} widens the exposure).
h2. Production observation (one incident day, summarized)
During a maintenance-window leadership reshuffle, 25 coordinator loads
failed across three brokers in four election waves — every failure with the
exact message above. 12 shards received no higher epoch and stayed FAILED
for up to ~4 h 13 min; the affected consumer groups received
NOT_COORDINATOR, and {{--list}}-based tooling showed them as absent.
No durable logical-record corruption was observed: every previously failed
partition later replayed successfully (all 41 successful loads in the same
windows completed normally — 0.1-4.2 s, 44k-656k records). The exact file
and slice geometry at the failure instant was not captured before restart.
{{FileRecords.close()}} trims the physical file to its logical size, which is
consistent with every observed post-restart load succeeding.
Sanitized log excerpts (broker ids retained, hosts removed):
{code}
[2026-07-17 09:04:51,377] INFO [GroupCoordinator id=10] Scheduling loading of
metadata from __consumer_offsets-35 with epoch 689
(org.apache.kafka.coordinator.common.runtime.CoordinatorRuntime)
[2026-07-17 09:04:53,797] ERROR [GroupCoordinator id=10] Failed to load
metadata from __consumer_offsets-35 with epoch 689 due to Record size 0 is less
than the minimum record overhead (14).
(org.apache.kafka.coordinator.common.runtime.CoordinatorRuntime)
org.apache.kafka.common.errors.CorruptRecordException: Record size 0 is less
than the minimum record overhead (14)
{code}
After the immediate failure-handling unload lines, no further coordinator
activity for that partition occurs until the broker restart approximately
two hours later. After restart, the same partition loads cleanly:
{code}
[2026-07-17 11:19:52,386] INFO [GroupCoordinator id=10] Finished loading of
metadata from __consumer_offsets-35 with epoch 691 in 674ms where 558ms was
spent in the scheduler. Loaded 65743 records which total to 10931884 bytes.
(org.apache.kafka.coordinator.common.runtime.CoordinatorRuntime)
{code}
Startup configuration dump confirms {{log.preallocate = true}}.
Full (sanitized) journald archives of the incident are available on request.
h2. Reproduction status
*Reproduced standalone on 2026-07-24* with a ~70-line program against the
official {{kafka-clients-4.2.0}} artifact from Maven Central (OpenJDK 21,
macOS/APFS — the same mechanism thus reproduces on a platform distinct
from the production Linux/XFS environment). The program mirrors the
relevant first-fetch allocation branch of {{toReadableMemoryRecords}}
(the method itself is private and lives in the coordinator-common
artifact); the mirrored allocate/{{readInto}} lines are byte-identical to
{{CoordinatorLoaderImpl.toReadableMemoryRecords}} at 4.2.0 and current trunk,
so this exercises the same defective sequence the coordinator runs. Setup:
{{FileRecords.open(file, mutable=true, fileAlreadyExists=false,
initFileSize=104857600, preallocate=true)}};
one valid 3-record batch appended (94 logical bytes; physical file
104,857,600 bytes); bounded {{slice(0, 94)}}; conversion with
{{loadBufferSize=5242880}}.
Result:
{code}
logical size = 94 bytes, physical file = 104857600 bytes (zero tail = 104857506)
--- PHASE 1: current loader conversion (no limit) ---
bytes copied into buffer: 5242880
PHASE 1 REPRODUCED: Record size 0 is less than the minimum record overhead (14)
--- PHASE 2: with buffer.limit(sizeInBytes) fix ---
bytes copied into buffer: 94
PHASE 2 FIX VERIFIED: iterated cleanly, records = 3
--- PHASE 3: preallocate=false workaround, defective conversion ---
logical = 94, physical = 94 (no tail)
bytes copied into buffer: 94
PHASE 3 WORKAROUND VERIFIED: defective conversion iterated cleanly, records = 3
{code}
Phase 3 verifies the operator workaround's core mechanism: with
{{preallocate=false}} the file has no zero tail, so the same unbounded
conversion is naturally bounded by physical EOF and completes cleanly.
The reproduction source ({{ReproTest.java}}) is attached. Not yet run: an
integration-level reproduction ({{__consumer_offsets}} with
{{preallocate=true}}, segment roll and leadership movement) and the
reused-enlarged-buffer variant.
h2. Suggested fix
Minimal, backportable — one line inserted immediately before the existing
{{readInto}} in {{toReadableMemoryRecords}}. It covers both the allocate and
reuse branches (both reach this call), and since {{bytesNeeded >= sizeInBytes}}
the new limit is always {{<= capacity}}:
{code:java}
// ... allocate-or-reuse of `buffer` unchanged ...
buffer.limit(sizeInBytes); // added: bound the copy to the logical
slice
fileRecords.readInto(buffer, 0);
{code}
A broader alternative is to make {{FileRecords.readInto}} clamp to
{{min(buffer.remaining(), sizeInBytes() - position)}} so the slice
abstraction is robust for all callers; that has a wider compatibility
surface and needs call-site auditing.
Additional regression coverage beyond the executed reproduction above: a
reused buffer previously enlarged past {{loadBufferSize}}, an empty slice, a
slice with non-zero start, and an integration test with
{{__consumer_offsets}}, {{preallocate=true}}, segment roll and leadership
movement.
Separately worth considering: a bounded same-epoch retry for retriable
load exceptions, and having ListGroups distinguish an unavailable shard
from an empty one. These are not required to fix the slice-boundary defect
and can be tracked separately.
h2. Related issue
Related issue: KAFKA-13664. We will add a reciprocal reference there if
this is retained as a separate issue; maintainers may instead prefer to
treat this as a continuation or duplicate.
h2. Workarounds for operators (until fixed)
* Set {{preallocate=false}} as a topic-level override on {{__consumer_offsets}}
(dynamic); note that existing preallocated tails persist until files are
closed/rolled, so a controlled rolling restart completes the mitigation. (Core
mechanism verified standalone — reproduction Phase 3.)
* Monitor the Kafka metric {{num-partitions}} in the group coordinator metrics
group with tag {{state=failed}} (> 0 alert); the exact JMX/exporter spelling
depends on the deployment. The state is otherwise easy to miss because
ListGroups masks it. (Gauge existence and state transitions are source-verified
in {{CoordinatorRuntimeMetricsImpl}}; the corresponding JMX ObjectName derived
from that source is
{{kafka.server:type=group-coordinator-metrics,name=num-partitions,state=failed}}.)
* A new load can be triggered by a higher leader epoch or by restarting the
current leader. Moving leadership to another unremediated replica may reproduce
the same failure. (Production-observed: 13 higher-epoch reloads succeeded
elsewhere, all 12 wedged shards healed on restart, and same-partition
re-failures occurred on unremediated replicas.)
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
