chia7712 commented on code in PR #23318:
URL: https://github.com/apache/kafka/pull/23318#discussion_r4162879220
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -501,25 +408,29 @@ abstract class AbstractFetcherThread(name: String,
}
/**
- * Returns initial partition fetch state based on current state and the
provided `initialFetchState`.
- * From IBP 2.7 onwards, we can rely on truncation based on diverging data
returned in fetch responses.
- * For older versions, we can skip the truncation step iff the leader epoch
matches the existing epoch.
+ * Returns the initial partition fetch state based on the current state and
the provided
+ * `initialFetchState`.
+ *
+ * If the current state already has the same leader epoch, it is preserved.
Otherwise, the
+ * partition starts from `initOffset`.
+ *
+ * When `latestEpoch` is present, the partition starts in the Fetching
state. Any log divergence is
+ * reported by the leader as a diverging epoch in the fetch response and
handled by
+ * `truncateOnFetchResponse`.
+ *
+ * If `latestEpoch` is empty (for example, when the log contains only
records in a message format
+ * older than v2), the fetch request carries no last fetched epoch and the
leader cannot report a
+ * diverging epoch. In this case, the partition starts in the Truncating
state, so that
+ * `maybeTruncate()` truncates its log to `initOffset`, which is the high
watermark.
*/
private def partitionFetchState(tp: TopicPartition, initialFetchState:
InitialFetchState, currentState: PartitionFetchState): PartitionFetchState = {
if (currentState != null && currentState.currentLeaderEpoch ==
initialFetchState.currentLeaderEpoch) {
currentState
- } else if (initialFetchState.initOffset < 0) {
- fetchOffsetAndTruncate(tp, initialFetchState.topicId,
initialFetchState.currentLeaderEpoch)
- } else if (leader.isTruncationOnFetchSupported) {
- // With old message format, `latestEpoch` will be empty and we use
Truncating state
- // to truncate to high watermark.
+ } else {
val lastFetchedEpoch = latestEpoch(tp)
val state = if (lastFetchedEpoch.isPresent) ReplicaState.FETCHING else
ReplicaState.TRUNCATING
- new PartitionFetchState(initialFetchState.topicId.toJava,
initialFetchState.initOffset, Optional.empty(),
initialFetchState.currentLeaderEpoch,
- state, lastFetchedEpoch)
- } else {
Review Comment:
It seems to be a bug for a future log with LEO > HWM, which is common when
the future log has caught up but the current replica still has uncommitted
data. The first fetch will cause the exception shown below.
```java
if (fetchOffset != futureLog.logEndOffset)
throw new IllegalStateException("Offset mismatch for the future
replica %s: fetched offset = %d, log end offset = %d.".format(
topicPartition, fetchOffset, futureLog.logEndOffset))
```
It was handled by the truncating phase (`getOffsetTruncationState` overrides
the initial offset), but that phase is now removed by this PR. I think we need
to update `ReplicaManager` to let `InitialFetchState` use
`initialFetchOffset(futureLog)` instead of HWM, same as the follower path.
```java
futureReplicasAndInitialOffset.put(topicPartition,
InitialFetchState(topicIds(topicPartition.topic), leader,
partition.getLeaderEpoch, futureLog.highWatermark))
val initialFetchState = InitialFetchState(topicId.toScala, new
BrokerEndPoint(config.brokerId, "localhost", -1),
partition.getLeaderEpoch, futureLog.highWatermark)
```
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -144,144 +142,71 @@ abstract class AbstractFetcherThread(name: String,
}
/**
- * Builds offset for leader epoch requests for partitions that are in the
truncating phase based
- * on latest epochs of the future replicas (the one that is fetching)
+ * Truncate the log of every partition in the truncating state to its fetch
offset, which is the
+ * high watermark. A partition only enters that state when the follower has
no leader epoch, i.e.
Review Comment:
The empty log also enters this state, right?
--
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]
