apoorvmittal10 commented on code in PR #17796:
URL: https://github.com/apache/kafka/pull/17796#discussion_r1868505445
##########
core/src/main/java/kafka/server/share/SharePartitionManager.java:
##########
@@ -665,17 +670,78 @@ public void handleFencedSharePartitionException(
// The share partition is fenced hence remove the partition from
map and let the client retry.
// But surface the error to the client so client might take some
action i.e. re-fetch
// the metadata and retry the fetch on new leader.
- SharePartition sharePartition =
partitionCacheMap.remove(sharePartitionKey);
- if (sharePartition != null) {
- sharePartition.markFenced();
- }
+ removeSharePartitionFromCache(sharePartitionKey,
partitionCacheMap, replicaManager);
}
}
private SharePartitionKey sharePartitionKey(String groupId,
TopicIdPartition topicIdPartition) {
return new SharePartitionKey(groupId, topicIdPartition);
}
+ private static void removeSharePartitionFromCache(SharePartitionKey
sharePartitionKey,
+ Map<SharePartitionKey, SharePartition> map, ReplicaManager
replicaManager) {
+ SharePartition sharePartition = map.remove(sharePartitionKey);
+ if (sharePartition != null) {
+ sharePartition.markFenced();
+
replicaManager.removeListener(sharePartitionKey.topicIdPartition().topicPartition(),
sharePartition.listener());
+ }
+ }
+
+ /**
+ * The SharePartitionListener is used to listen for partition events. The
share partition is associated with
+ * the topic-partition, we need to handle the partition events for the
share partition.
+ * <p>
+ * The partition cache map stores share partitions against share partition
key which comprises
+ * group and topic-partition. Instead of maintaining a separate map for
topic-partition to share partitions,
+ * we can maintain the share partition key in the listener and create a
new listener for each share partition.
+ */
+ static class SharePartitionListener implements PartitionListener {
+
+ private final SharePartitionKey sharePartitionKey;
+ private final ReplicaManager replicaManager;
+ private final Map<SharePartitionKey, SharePartition> partitionCacheMap;
+
+ SharePartitionListener(
+ SharePartitionKey sharePartitionKey,
+ ReplicaManager replicaManager,
+ Map<SharePartitionKey, SharePartition> partitionCacheMap
+ ) {
+ this.sharePartitionKey = sharePartitionKey;
+ this.replicaManager = replicaManager;
+ this.partitionCacheMap = partitionCacheMap;
+ }
+
+ @Override
+ public void onFailed(TopicPartition topicPartition) {
+ log.debug("The share partition failed listener is invoked for the
topic-partition: {}, share-partition: {}",
+ topicPartition, sharePartitionKey);
+ onUpdate(topicPartition);
+ }
+
+ @Override
+ public void onDeleted(TopicPartition topicPartition) {
+ log.debug("The share partition delete listener is invoked for the
topic-partition: {}, share-partition: {}",
+ topicPartition, sharePartitionKey);
+ onUpdate(topicPartition);
+ }
+
+ @Override
+ public void onBecomingFollower(TopicPartition topicPartition) {
+ log.debug("The share partition leader change listener is invoked
for the topic-partition: {}, share-partition: {}",
Review Comment:
Done.
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -6798,6 +6798,105 @@ class ReplicaManagerTest {
assertEquals(Double.NaN, maxMetric.metricValue)
}
+ @Test
+ def testBecomeFollowerInvokeOnBecomingFollowerListener(): Unit = {
+ val localId = 1
+ val topicPartition = new TopicPartition("foo", 0)
+ val replicaManager = setupReplicaManagerWithMockedPurgatories(new
MockTimer(time), localId)
+ // Attach listener to partition.
+ val offsetCheckpoints = new
LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints.asJava)
+ replicaManager.createPartition(topicPartition).createLogIfNotExists(isNew
= false, isFutureReplica = false, offsetCheckpoints, None)
+ val listener = new MockPartitionListener
+ assertTrue(replicaManager.maybeAddListener(topicPartition, listener))
+ listener.verify()
+
+ try {
+ // Make the local replica the leader
+ val leaderTopicsDelta = topicsCreateDelta(localId, true)
+ val leaderMetadataImage = imageFromTopics(leaderTopicsDelta.apply())
+
+ replicaManager.applyDelta(leaderTopicsDelta, leaderMetadataImage)
+
+ // Check the state of that partition and fetcher
+ val HostedPartition.Online(leaderPartition) =
replicaManager.getPartition(topicPartition)
+ assertTrue(leaderPartition.isLeader)
+ assertEquals(0, leaderPartition.getLeaderEpoch)
+ // On becoming follower listener should not be invoked yet.
+ listener.verify()
+
+ // Change the local replica to follower
+ val followerTopicsDelta =
topicsChangeDelta(leaderMetadataImage.topics(), localId, false)
+ val followerMetadataImage = imageFromTopics(followerTopicsDelta.apply())
+ replicaManager.applyDelta(followerTopicsDelta, followerMetadataImage)
+
+ // On becoming follower listener should be invoked.
+ listener.verify(expectedFollower = true)
+
+ // Check the state of that partition.
+ val HostedPartition.Online(followerPartition) =
replicaManager.getPartition(topicPartition)
+ assertFalse(followerPartition.isLeader)
+ assertEquals(1, followerPartition.getLeaderEpoch)
+ } finally {
+ replicaManager.shutdown(checkpointHW = false)
+ }
+ }
+
+ @Test
+ def testBecomeFollowerNotInvokeOnBecomingFollowerListenerOnZk(): Unit = {
Review Comment:
Done, removed.
--
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]
