m1a2st commented on code in PR #23318:
URL: https://github.com/apache/kafka/pull/23318#discussion_r4166908724
##########
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:
Good point, I have addressed it!
--
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]
