ramu11 opened a new pull request, #23298:
URL: https://github.com/apache/kafka/pull/23298
## What is the bug?
When a Kafka Streams state store finishes restoring its changelog,
`ProcessorStateManager` tracks the restored position using last-applied-offset
semantics, while the changelog restore boundary and Kafka consumer position use
next-fetch-offset semantics.
For an empty changelog with a high log end offset, there may be no restored
records to update the state-store offset. As a result, the restored offset can
remain unset and `changelogOffsets()` reports it as `0`, even though
restoration has completed at a much higher offset.
For example:
Changelog LEO = 20,000
Records restored = 0
Expected advertised offset = 20,000
Actual offset before this fix = 0
This incorrect offset propagates into the task offset sum and ultimately
into the high-availability task lag calculation:
restore completion
↓
ProcessorStateManager / changelogOffsets()
↓
TaskManager.taskOffsetSums()
↓
StateDirectory.sumOfChangelogOffsets()
↓
ClientState.computeTaskLags()
↓
HighAvailabilityTaskAssignor.acceptable()
With the incorrect offset, a task can appear to have approximately `20,000`
records of recovery lag even though its changelog restore has already
completed. If that exceeds `acceptable.recovery.lag`, the task can be
considered unacceptable for assignment.
## What does this change do?
This patch adds a package-private helper:
ProcessorStateManager.advanceRestoredOffsetTo()
The helper accepts a next-fetch offset and converts it to the internal
last-applied-offset representation.
For example:
next-fetch offset = 20,000
last-applied offset = 19,999
changelogOffsets() = 20,000
The helper also:
* Does nothing for `nextOffsetToFetch <= 0`.
* Never moves an existing restored offset backwards.
* Only advances the offset when the target boundary is ahead of the current
restored position.
* Uses the same persistent changelog-offset tracking used by the existing
restore/checkpoint path.
## Restore-boundary handling
For ACTIVE tasks, the restored offset is advanced only after the changelog
has actually completed restoration.
Importantly, remaining-record accounting is performed before synthesizing
the final restored offset.
This ordering is required for offset gaps such as transaction markers or
compacted records. If the restored offset were advanced first, the
remaining-record calculation could become zero prematurely and trailing offset
holes would not be accounted for correctly.
The ACTIVE path also uses `restoreEndOffset` rather than the live consumer
position. The consumer may already have fetched records beyond the restore
boundary that are buffered but have not yet been applied to the state store.
Advertising the live position in that situation would incorrectly claim that
those records had been restored.
## STANDBY handling
The STANDBY path follows the same distinction between fetched and actually
restored data.
The offset is not advanced when:
* The poll is empty but lag is still positive.
* Records remain buffered and have not yet been applied.
* The restore boundary is not initialized (`restoreEndOffset == 0`).
* The consumer position/lag lookup times out.
* The requested target would move the restored offset backwards.
For dedicated changelogs, catch-up is based on `currentLag() == 0` and the
consumer's next-fetch position.
For source changelogs, the committed restore limit remains the upper
boundary; the standby does not advertise progress beyond that limit.
## Tests
The regression tests exercise the actual restore and task-lag paths rather
than only testing the helper in isolation.
### KAFKA-14302 regression
`StoreChangelogReaderTest.shouldReportCaughtUpOffsetAfterRestoringEmptyChangelogWithHighEndOffset`
Uses a real `ProcessorStateManager` and `StoreChangelogReader` to reproduce
the empty-changelog/high-LEO scenario.
It verifies that:
LEO = 20,000
restored records = 0
advertised offset = 20,000
last-applied offset = 19,999
instead of the previous advertised offset of `0`.
### Task-lag regression
`StoreChangelogReaderCaughtUpOffsetTest.shouldReportZeroTaskLagAfterEmptyHighEndOffsetRestore`
Carries the restored offset through the task offset-sum and
`ClientState.computeTaskLags()` path.
It verifies that the fixed restore produces:
task lag = 0
where the old behavior would produce:
task lag = 20,000
and therefore could exceed the configured acceptable recovery lag.
### Additional regression coverage
The tests also cover:
* ACTIVE restore using `restoreEndOffset` rather than live consumer position.
* Empty restore with `restoreEndOffset == 0`.
* ACTIVE restore remaining behind the end offset.
* STANDBY empty polls while lag remains positive.
* STANDBY source changelog with buffered records.
* STANDBY not advancing beyond the committed restore limit.
* Timeout handling.
* Last-applied/next-fetch offset conversion.
* Monotonicity of restored offsets.
* Checkpoint round-trip behavior.
* Trailing offset-hole accounting and restore ordering.
* Existing non-empty restore behavior.
The broader Streams tests covering `StreamTask`, `StandbyTask`,
`TaskManager`, and `HighAvailabilityTaskAssignor` also pass.
## Scope
This is intentionally a small restore-boundary fix.
There are no public API changes, no KIP changes, and no changes to the
existing `changelogOffsets()` representation or checkpoint semantics.
The fix only ensures that once restoration has actually reached its known
boundary, the corresponding restored offset is correctly reflected in the
task's offset sum.
## Testing
The following tests were run successfully:
ProcessorStateManagerTest
StoreChangelogReaderTest
StoreChangelogReaderCaughtUpOffsetTest
StreamTaskTest
StandbyTaskTest
TaskManagerTest
HighAvailabilityTaskAssignorTest
Additionally:
git diff --cached --check
passed with no whitespace errors.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
