This is an automated email from the ASF dual-hosted git repository.
rsivaram pushed a commit to branch 2.8
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/2.8 by this push:
new 9618c61 KAFKA-13207: Skip truncation on fetch response with diverging
epoch if partition removed from fetcher (#11221)
9618c61 is described below
commit 9618c61a2488fd8a1ecaf140c27f27e58a0efdf3
Author: Rajini Sivaram <[email protected]>
AuthorDate: Tue Aug 17 17:12:51 2021 +0100
KAFKA-13207: Skip truncation on fetch response with diverging epoch if
partition removed from fetcher (#11221)
AbstractFetcherThread#truncateOnFetchResponse is used with IBP 2.7 and
above to truncate partitions based on diverging epoch returned in fetch
responses. Truncation should only be performed for partitions that are still
owned by the fetcher and this check should be done while holding
partitionMapLock to ensure that any partitions removed from the fetcher thread
are not truncated. Truncation will be performed by any new fetcher that owns
the partition when it restarts fetching.
Reviewers: David Jacot <[email protected]>, Jason Gustafson
<[email protected]>
---
.../scala/kafka/server/AbstractFetcherThread.scala | 38 +++++++++++--------
.../kafka/server/AbstractFetcherThreadTest.scala | 43 ++++++++++++++++++++++
2 files changed, 66 insertions(+), 15 deletions(-)
diff --git a/core/src/main/scala/kafka/server/AbstractFetcherThread.scala
b/core/src/main/scala/kafka/server/AbstractFetcherThread.scala
index 5e716e2..492efa9 100755
--- a/core/src/main/scala/kafka/server/AbstractFetcherThread.scala
+++ b/core/src/main/scala/kafka/server/AbstractFetcherThread.scala
@@ -233,7 +233,8 @@ abstract class AbstractFetcherThread(name: String,
}
}
- protected def truncateOnFetchResponse(epochEndOffsets: Map[TopicPartition,
EpochEndOffset]): Unit = {
+ // Visibility for unit tests
+ protected[server] def truncateOnFetchResponse(epochEndOffsets:
Map[TopicPartition, EpochEndOffset]): Unit = {
inLock(partitionMapLock) {
val ResultWithPartitions(fetchOffsets, partitionsWithError) =
maybeTruncateToEpochEndOffsets(epochEndOffsets, Map.empty)
handlePartitionsWithErrors(partitionsWithError,
"truncateOnFetchResponse")
@@ -266,21 +267,28 @@ abstract class AbstractFetcherThread(name: String,
val partitionsWithError = mutable.HashSet.empty[TopicPartition]
fetchedEpochs.forKeyValue { (tp, leaderEpochOffset) =>
- Errors.forCode(leaderEpochOffset.errorCode) match {
- case Errors.NONE =>
- val offsetTruncationState = getOffsetTruncationState(tp,
leaderEpochOffset)
- if (doTruncate(tp, offsetTruncationState))
- fetchOffsets.put(tp, offsetTruncationState)
-
- case Errors.FENCED_LEADER_EPOCH =>
- val currentLeaderEpoch = latestEpochsForPartitions.get(tp)
- .map(epochEndOffset =>
Int.box(epochEndOffset.currentLeaderEpoch)).asJava
- if (onPartitionFenced(tp, currentLeaderEpoch))
+ if (partitionStates.contains(tp)) {
+ Errors.forCode(leaderEpochOffset.errorCode) match {
+ case Errors.NONE =>
+ val offsetTruncationState = getOffsetTruncationState(tp,
leaderEpochOffset)
+ if (doTruncate(tp, offsetTruncationState))
+ fetchOffsets.put(tp, offsetTruncationState)
+
+ case Errors.FENCED_LEADER_EPOCH =>
+ val currentLeaderEpoch = latestEpochsForPartitions.get(tp)
+ .map(epochEndOffset =>
Int.box(epochEndOffset.currentLeaderEpoch)).asJava
+ if (onPartitionFenced(tp, currentLeaderEpoch))
+ partitionsWithError += tp
+
+ case error =>
+ info(s"Retrying leaderEpoch request for partition $tp as the
leader reported an error: $error")
partitionsWithError += tp
-
- case error =>
- info(s"Retrying leaderEpoch request for partition $tp as the leader
reported an error: $error")
- partitionsWithError += tp
+ }
+ } else {
+ // Partitions may have been removed from the fetcher while the thread
was waiting for fetch
+ // response. Removed partitions are filtered out while holding
`partitionMapLock` to ensure that we
+ // don't update state for any partition that may have already been
migrated to another thread.
+ trace(s"Ignoring epoch offsets for partition $tp since it has been
removed from this fetcher thread.")
}
}
diff --git
a/core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala
b/core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala
index 9b9fe8e..3b6fa41 100644
--- a/core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala
+++ b/core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala
@@ -40,6 +40,7 @@ import
org.apache.kafka.common.requests.OffsetsForLeaderEpochResponse.{UNDEFINED
import org.apache.kafka.common.requests.FetchRequest
import org.apache.kafka.common.utils.Time
import org.junit.jupiter.api.Assertions._
+import org.junit.jupiter.api.Assumptions.assumeTrue
import org.junit.jupiter.api.{BeforeEach, Test}
import scala.jdk.CollectionConverters._
@@ -438,6 +439,48 @@ class AbstractFetcherThreadTest {
}
@Test
+ def testTruncationOnFetchSkippedIfPartitionRemoved(): Unit = {
+ assumeTrue(truncateOnFetch)
+ val partition = new TopicPartition("topic", 0)
+ var truncations = 0
+ val fetcher = new MockFetcherThread {
+ override def truncate(topicPartition: TopicPartition, truncationState:
OffsetTruncationState): Unit = {
+ truncations += 1
+ super.truncate(topicPartition, truncationState)
+ }
+ }
+ val replicaLog = Seq(
+ mkBatch(baseOffset = 0, leaderEpoch = 0, new SimpleRecord("a".getBytes)),
+ mkBatch(baseOffset = 1, leaderEpoch = 2, new SimpleRecord("b".getBytes)),
+ mkBatch(baseOffset = 2, leaderEpoch = 4, new SimpleRecord("c".getBytes)))
+
+ val replicaState = MockFetcherThread.PartitionState(replicaLog,
leaderEpoch = 5, highWatermark = 2L)
+ fetcher.setReplicaState(partition, replicaState)
+
+ // Verify that truncation based on fetch response is performed if
partition is owned by fetcher thread
+ fetcher.addPartitions(Map(partition -> initialFetchState(6L, leaderEpoch =
4)))
+ val endOffset = new EpochEndOffset()
+ .setPartition(partition.partition)
+ .setErrorCode(Errors.NONE.code)
+ .setLeaderEpoch(4)
+ .setEndOffset(3L)
+ fetcher.truncateOnFetchResponse(Map(partition -> endOffset))
+ assertEquals(1, truncations)
+
+ // Verify that truncation based on fetch response is not performed if
partition is removed from fetcher thread
+ val offsets = fetcher.removePartitions(Set(partition))
+ assertEquals(Set(partition), offsets.keySet)
+ assertEquals(3L, offsets(partition).fetchOffset)
+ val newEndOffset = new EpochEndOffset()
+ .setPartition(partition.partition)
+ .setErrorCode(Errors.NONE.code)
+ .setLeaderEpoch(4)
+ .setEndOffset(2L)
+ fetcher.truncateOnFetchResponse(Map(partition -> newEndOffset))
+ assertEquals(1, truncations)
+ }
+
+ @Test
def testFollowerFetchOutOfRangeHigh(): Unit = {
val partition = new TopicPartition("topic", 0)
val fetcher = new MockFetcherThread()
