sepuri sai krishna created KAFKA-20770:
------------------------------------------
Summary: LocalLeaderEndPoint returns leader epoch 0 instead of -1
when the epoch cannot be resolved, inconsistent with the ListOffsets-based path
Key: KAFKA-20770
URL: https://issues.apache.org/jira/browse/KAFKA-20770
Project: Kafka
Issue Type: Bug
Components: core
Affects Versions: 4.3.0
Reporter: sepuri sai krishna
LocalLeaderEndPoint (used by ReplicaAlterLogDirsThread) and
RemoteLeaderEndPoint (used by ReplicaFetcherThread) implement the same
LeaderEndPoint contract, but they disagree on what leader epoch to report when
the epoch for an offset cannot be resolved from the leader epoch cache.
The remote path goes through ListOffsets:
UnifiedLog#fetchEarliestPendingUploadOffset leaves the epoch unset when
LeaderEpochFileCache#epochForOffset returns empty, ReplicaManager only sets the
response leaderEpoch when present, and the ListOffsetsResponse schema default
is -1 (NO_LEADER_EPOCH). So RemoteLeaderEndPoint reports epoch -1, meaning
"unknown".
The local path instead fabricates epoch 0: fetchEarliestPendingUploadOffset in
core/src/main/scala/kafka/server/LocalLeaderEndPoint.scala ends with
epoch.orElse(0), and the sibling methods fetchEarliestOffset, fetchLatestOffset
and fetchEarliestLocalOffset use the same orElse(0) fallback. Note that
fetchEarliestPendingUploadOffset also returns fetchEarliestOffset's result
wholesale in its "no segments have been uploaded yet" branch, so the fabricated
epoch 0 can leak out of the pending-upload API through that branch as well.
This matters because the epoch returned by these methods is consumed by
TierStateMachine#buildRemoteLogAuxState, which treats it as meaningful: epoch 0
short-circuits the earlier-epoch lookup ("no need to fetch from earlier
epoch"), while any other value drives an OffsetsForLeaderEpoch probe and an
epoch-based remote log segment metadata lookup
(fetchRemoteLogSegmentMetadata(topicPartition, targetEpoch, offset)). A
fabricated 0 therefore claims the first leader epoch rather than admitting the
epoch is unknown, and the two fetcher paths behave differently for identical
log state.
In practice the fallback is hard to hit on a healthy broker, because message
format v2 always populates the epoch cache and
LeaderEpochFileCache#truncateFromStart clamps the first entry to the log start
offset, so epochForOffset resolves for any offset at or above logStartOffset.
It becomes reachable when the leader epoch cache is empty or truncated above
the queried offset (for example after checkpoint loss or corruption). Even then
the consequence is a misleading epoch handed to TierStateMachine rather than
data loss, hence Minor priority. The main goal of this fix is correctness and
consistency between the two LeaderEndPoint implementations.
Identified by code inspection on trunk while comparing the two implementations
introduced/extended by KIP-1023 (KAFKA-15433, shipped in 4.3.0), and confirmed
with unit tests. The existing tests in LocalLeaderEndPointTest did not catch
this because they computed the expected epoch using the same
epochForOffset(...).orElse(0) expression as the production code.
Proposed fix: replace the orElse(0) fallback with orElse(UNDEFINED_EPOCH) (-1,
org.apache.kafka.common.requests.OffsetsForLeaderEpochResponse.UNDEFINED_EPOCH,
already imported in the file) in all four methods, matching the
ListOffsets-based path. Behavior is unchanged whenever the epoch is resolvable.
Also replace the tautological epoch expectations in LocalLeaderEndPointTest
with concrete values and add tests covering a non-zero resolved epoch and the
unresolvable-epoch case.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
