[
https://issues.apache.org/jira/browse/KAFKA-20416?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18108061#comment-18108061
]
Bill Bejeck commented on KAFKA-20416:
-------------------------------------
Hey [~fabianbell] applogies for the delay in the investigation.
>From the information provided, here's an (unproven) theory as to what happened
Two independent consequences of a broker patch have to land on the same task in
the same window — either alone is survivable, but together they silently empty
a store.
Race A — transient null committed offset. During group-coordinator
resignation/failover on the broker being patched, a concurrent requireStable
OffsetFetch can observe the group as momentarily absent and return
committedOffset = -1 / Errors.NONE — a successful "no committed offset", not a
retriable error. AdminClient.listConsumerGroupOffsets maps that -1 to a null
OffsetAndMetadata. (This transient-null path is reasoned from the coordinator's
resignation ordering, not reproduced; retention expiry and __consumer_offsets
truncation are alternative ways the same -1/NONE arises.)
Race B — EOS state wipe. In the same window, an EOS AddOffsetsToTxn timeout
raises TaskCorruptedException; the task is closed dirty and its state directory
(and checkpoint) is wiped.
The intersection (StoreChangelogReader, apache/kafka 3.9): after the wipe the
store re-initializes, and the transient null committed offset is coerced to 0
in committedOffsetForChangelogs (line 736):
// line 754 ← the poisoned value
.collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue() == null ? 0L :
e.getValue().offset()));
0 is non-null, so it passes the guard at line 867 and:
// line 874
changelogMetadata.restoreEndOffset = Math.min(endOffset, committedOffset); //
min(logEndOffset, 0) = 0
// line 876 → logs: "End offset for changelog <topic> initialized as 0."
hasRestoredToEnd is immediately true → the wiped store is declared fully
restored while empty → KTable returns null for keys that exist in the topic.
(Without the wipe, the on-disk store survives a 0 ceiling; without the 0, the
wiped store re-restores fully — both are required.)
But this should only affect tables that use their source topic as the changelog
(changelogAsSource — source-topic reuse via topology.optimization, or global
stores). Those are the only stores that fetch the group's committed offset and
feed it into the restore ceiling (initializeChangelogs lines 851-852). A
dedicated changelog uses committedOffset = Long.MAX_VALUE (line 865 in
initializeChangelogs), and never touches line 754's value.
But there's one inconsistency to clear up. The configs shared here show
topology.optimization = none, but the 3/10 logs show StoreChangelogReader
restoring prod-vss-vehicle-state-V7-11 as a changelog on a normal StreamThread.
A Streams-managed changelog is always named <application.id>-<store>-changelog
— and the other two changelogs in these logs follow exactly that
(enrichment-prod-Processors-V7-vehicle-processors-task-storage-changelog,
…-aggregate-store-changelog), which also fixes application.id =
enrichment-prod-Processors-V7. The only way a plain source topic
(prod-vss-vehicle-state-V7, no app-id prefix, no -changelog suffix) becomes a
store's changelog is source-topic reuse (reuse.ktable.source.topics / global
store). And that's gated: in InternalStreamsBuilder, reuseKTableSourceTopics()
runs only when optimizationConfigs.contains(REUSE_KTABLE_SOURCE_TOPICS) —
impossible under none. So none cannot have been the config that produced the
3/10 logs.
Given the logs are dated 2026-03-10 and the ticket ~2026-04-08: was there a
redeploy or config change in between, and what was topology.optimization set to
on 3/10? The most likely explanation is that source-topic reuse was enabled at
incident time (so the app was exposed to the double race above) and none is a
later change — possibly the mitigation itself, since flipping
topology.optimization to none converts prod-vss-vehicle-state from a reused
source-topic changelog (poisonable) into a dedicated one (immune).
I'll provide a PR soon for a fix
> RocksDB loses entries during broker patches.
> --------------------------------------------
>
> Key: KAFKA-20416
> URL: https://issues.apache.org/jira/browse/KAFKA-20416
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Affects Versions: 3.9.1
> Environment: Broker: MSK with kafka.m7g.2xlarge instances
> Application: EKS with m7i.2xlarge instances and Bottlerocket OS 1.57.0
> Reporter: Fabian Bell
> Priority: Major
> Attachments: logs
>
>
> h2. Problem:
> We discovered a strange behaviour on our production environment. We use a
> KTable to look up data from a topic we write to.
>
> {code:java}
> builder.table(topicName, Consumed.with(keySerde, valueSerde),
> Materialized.as(storeName)) {code}
>
> When we access the store in the processor, we observed that the store
> returned null values for keys that have non-null entries in the topic that
> backs the KTable after an MSK security patch. We never tombstone an entry in
> our topic nor have a delete retention activated.
> This only happens for some of our instances.
> We see the following stream logs:
>
> {code:java}
> Committing task(s) 0_14 failed.
> Detected the states of tasks [0_14] are corrupted. Will close the task as
> dirty and re-create and bootstrap from scratch.
> Active task(s) got corrupted. Triggering a rebalance.
> End offset for changelog our-topic-14 initialized as 16596290.
> Restoration in progress for 1 partitions. {our-topic-14: position=0,
> end=16596290, totalRestored=0}
> State transition from RUNNING to PARTITIONS_REVOKED
> No followup rebalance was requested, resetting the rebalance schedule.
> partition revocation took 80 ms.
> State transition from PARTITIONS_REVOKED to PARTITIONS_ASSIGNED
> State transition from PARTITIONS_ASSIGNED to RUNNING {code}
>
> This all happens within a few seconds, and the `Restoration in progress ...`
> log is the only one we can see. A full restoration usually takes like 30 min.
> The error message of the commit failure is
> {code:java}
> o.a.k.c.e.TimeoutException: Timeout expired after 60000ms while awaiting
> AddOffsetsToTxn {code}
> We can fix this situation by clearing the state directory and forcing a full
> restoration.
> h2. Context:
> Each instance has its own persistent state directory. The configured state
> directory does not change.
> Processing Guarantee: exactly_once_v2
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
