KAFKA-20876: Windowed restore optimisation gives up when endOffset-1 is a transaction control record - #23086
Open
alanlau28 wants to merge 13 commits into
Open
KAFKA-20876: Windowed restore optimisation gives up when endOffset-1 is a transaction control record#23086alanlau28 wants to merge 13 commits into
endOffset-1 is a transaction control record#23086alanlau28 wants to merge 13 commits into
Conversation
Three defects stopped the KAFKA-13499 windowed restore optimisation from ever executing under exactly-once. All are reachable on trunk with enable.transactional.statestores at its default of false, since the EOS unclean close still wipes the task directory and forces a from-scratch restore. 1. PlainToHeadersWindowStoreAdapter holds its delegate in a private field rather than as a WrappedStateStore, so extractRetentionPeriod's unwrap walk ends on the adapter and resolves -1. That fails the gate and silently skips the optimisation for every stream-stream join store. Measured on a 4.3 soak carrying a backport: 441 and 379 non-finite retentions, 0 optimised seeks. 2. The head-timestamp probe seeked endOffset-1 and gave up on an empty poll. Under EOS that offset is a transaction control record -- confirmed by dumping a live segment: the last batch is isControl=true, count=1, size=78, and the pattern recurs at every transaction boundary. Markers occupy an offset but are never delivered to a consumer, so the poll returns nothing. Replaced with a bounded backward probe. It starts at endOffset-32 rather than -1, which costs nothing in accuracy because the probe takes the newest record of the returned batch. 3. The probe shared one poll across all unresolved partitions, so an empty result could merely mean another partition's fetch landed first -- doubling that partition's step-back on false evidence and spending a shared attempt budget on its behalf. Partitions starved into log-start seeks with zero margin and were lapped into OffsetOutOfRangeException. Each partition is now probed alone, with the others paused. Note poll() updates fetch positions for the whole assignment, so every probed partition is seeked up front; without that the first poll throws NoOffsetForPartitionException under auto.offset.reset=none. Tests: StoreChangelogReaderTest 47, SeekFallback 5, adapter 2, unwrap probe 1, soak topology probe 1, TxnStoreWipeAB 3, and a 7-test EOS-v2 integration suite against a real cluster. Each fix was verified to fail its test when reverted.
…e back Every probe attempt re-seeks before polling, which cancels a fetch that was in flight. An empty poll therefore does not distinguish "no record at this offset" from "the fetch has not landed yet", and the step-back doubles on that false evidence. Soak evidence that this is happening: with the probe starting at endOffset-32, ~46% of probes still walk out to backUsed 512-2048, and probeMs medians sit at 1.0-1.5s against ~0.5s for the previous build. Long runs of control records would explain that, but so would cancelled fetches, and the two are indistinguishable from the current instrumentation. Polling a second time at the same position separates them. If the deep step-backs are self-inflicted, backUsed should collapse to 32 and probeMs should fall; if they persist, the runs of control records are real. Note the existing tests pass either way -- this changes timing behaviour that neither MockConsumer nor a loopback EmbeddedKafkaCluster reproduces, so the result has to come from the soak.
… probing Individual polling removed the starvation but lost amortisation: 5 partitions cost 22 attempts and 2013ms against 1 attempt and 3ms for a shared poll that is simply polled more than once. The shared poll always could serve every partition; the defect was giving up after the first empty result, which only means a fetch has not landed. Also fixes the regression test's observable: a partition's position after restore reflects records it has since consumed, not where it was seeked, so the test now records seekToBeginning calls directly.
… one empty poll When restoring a windowed store with no checkpoint, StoreChangelogReader seeks to latestTimestamp - retentionPeriod rather than log-start, since data older than the retention is discarded on write. It learned latestTimestamp by seeking to endOffset - 1 and polling once; an empty poll fell back to seekToBeginning. Two reasons that poll comes back empty without the offset being at fault: - Under EOS the last offset is almost always a transaction control record, which occupies an offset but is never delivered to a consumer. Measured on a soak, endOffset - 1 returned a record on 0 of 195 probes, and dumping a segment shows the last batch is isControl=true with the pattern recurring at every transaction boundary. - One poll returns as soon as any fetch lands, so a partition can be empty because another was served first or because its own fetch has not arrived. The probe now starts at endOffset - 32 and polls repeatedly at each position before stepping back, bounded by PROBE_MAX_ATTEMPTS and the log's beginning. It takes the newest record of the returned batch, so a deeper start costs nothing in accuracy. Falling back to seekToBeginning remains the behaviour when no data record can be found. PlainToHeadersWindowStoreAdapter also now reports the retention of the store it adapts. It holds its delegate in a private field rather than as a WrappedStateStore, so extractRetentionPeriod's unwrap walk terminated on the adapter and resolved -1, silently skipping the optimisation for every stream-stream join store. shouldRetryProbePollBeforeFallingBackToLogStart fails without the change.
The probe stepped back from 32, re-seeking after three polls whether or not those polls had waited on a fetch, so a partition could be widened away from the answer it was about to give and end up restoring from log start. Start at 128, which answers the large majority outright, and give each window until it stops resolving anyone before widening to 512 and 2048.
alanlau28
force-pushed
the
KAFKA-20876-probe-retry
branch
from
August 9, 2026 15:03
f527149 to
bc2e02f
Compare
bbejeck
reviewed
Aug 11, 2026
| // usually a transaction control record, which is never delivered to a consumer, so probing | ||
| // there is a guaranteed empty poll. The first window answers the large majority; only the | ||
| // partitions it cannot answer for pay to widen. | ||
| private static final long[] PROBE_WINDOWS = {128L, 512L, 2048L}; |
Member
There was a problem hiding this comment.
You may have had this before but consider adding a shallower rung: PROBE_WINDOWS = {8L, 128L, 512L, 2048L}. The probe fetches the entire window, but we only need one record from it — so starting smaller cuts traffic.
I know this is a reversal of a comment I had earlier but some local testing convinced me otherwise.
UPDATE: I've been thinking about this and I'm not sure - if the record with the correct ending timestamp is in this first rung, then we don't pull nearly the same amount of data - but on the other hand if it's not we're increasing the load on the broker, so I'm leaning on keeping this as is to keep broker load to a minimum
Member
|
@alanlau28 with merging #23037 we should rebase this PR as I think some of the refactoring may have reached this PR. Overall this is looking good to me but I think we need to harden this some to guard against a constant probing due crash-looping.
|
…robe The lookups inherited default.api.timeout.ms, and a task that is corrupted, wiped and re-registered in a loop probed in full on every iteration.
A re-seek discards the fetch already in flight for that partition, and its replacement waits behind it because the consumer keeps only one fetch in flight per broker. At five poll timeouts a window could not absorb that, so the widest window -- the one most likely to answer -- ran out of budget.
bbejeck
reviewed
Aug 12, 2026
| // probe is an optimisation that must not hold the restore thread that long. Generous, because | ||
| // expiry sends every windowed partition in the batch to log start: the bound is here to cap a | ||
| // stall, not to react to a slow broker. | ||
| private static final Duration OFFSET_LOOKUP_TIMEOUT = Duration.ofSeconds(60); |
Member
There was a problem hiding this comment.
Let's actually remove this variable (and it's comment) and go with the non-overloaded calls then they will use the same timeouts as the others which is default.api.timeout.ms which is 60 seconds
Comment on lines
1258
to
1266
| } catch (final TimeoutException e) { | ||
| log.debug("Could not seek by timestamp for changelog partitions {}, falling back to seek-to-beginning", | ||
| windowedPartitionsRetention.keySet(), e); | ||
| seekToBeginningPartitions.addAll(windowedPartitionsRetention.keySet()); | ||
| } catch (final KafkaException e) { | ||
| log.warn("Failed to seek by timestamp for changelog partitions {}, falling back to seek-to-beginning", | ||
| windowedPartitionsRetention.keySet(), e); | ||
| seekToBeginningPartitions.addAll(windowedPartitionsRetention.keySet()); | ||
| } finally { |
Member
There was a problem hiding this comment.
The probeFailedAtMs is only updated from seekByRetentionFromPolledRecords, which is the last statement of the try at 1257 — so neither catch arms the backoff. A probe that fails by TimeoutException re-probes on every iteration
Drop the offset lookup bound so the calls inherit default.api.timeout.ms like the rest of the class, arm the probe backoff from the finally so a lookup that times out cannot re-probe on every iteration, and keep a backed-off partition off the "no usable retention period" warning path.
System.nanoTime has an arbitrary origin, so comparing against an absolute deadline can wrap; on wrap the window would skip its wait entirely and fall back to log start, which is the behaviour this was added to prevent.
Contributor
There was a problem hiding this comment.
Pull request overview
Improves the Kafka Streams windowed-store restore optimization when there is no checkpoint by making the “latest timestamp” probe more robust (avoiding empty polls when endOffset - 1 is a transactional control record, and avoiding premature fallback to log-start).
Changes:
- Replace the single-offset probe with a backward “window” probe that can widen and keep polling while partitions are still resolving.
- Add a probe-failure backoff to avoid repeatedly paying the probe cost in tight re-register loops.
- Add targeted unit tests covering empty/idle polls, slow/unready consumers, per-partition resolution pacing, and backoff behavior.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.
| File | Description |
|---|---|
| streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java | Implements backward-window probing, polling rules, and probe-failure backoff before falling back to log-start restores. |
| streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java | Adds tests validating probe retry behavior, polling semantics, and backoff; includes a small comment cleanup nit. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Member
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Jira: https://issues.apache.org/jira/browse/KAFKA-20876
When restoring a windowed store with no checkpoint,
StoreChangelogReaderseeks tolatestTimestamp - retentionPeriodrather than log-start, since data older than the retention is discarded
on write. It learns
latestTimestampby seeking toendOffset - 1andpolling once in
seekNewPartitions; an empty poll falls back toseekToBeginning.The probe now reads a window back from the end rather than a single
offset, widens it when that window holds no record, and keeps polling a
window for as long as it is still resolving partitions.
Reviewers: Bill Bejeck bbejeck@apache.org