You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "chia7712 (via GitHub)" <gi...@apache.org> on 2026/10/11 16:59:44 UTC
Re: [PR] KAFKA-21142: fix partition creation when handling snapshot delta during migration [kafka]
chia7712 commented on code in PR #23610:
URL: https://github.com/apache/kafka/pull/23610#discussion_r4242330390
##########
metadata/src/test/java/org/apache/kafka/metadata/migration/KRaftMigrationZkWriterTest.java:
##########
@@ -98,9 +98,66 @@ public void iterateTopics(EnumSet<TopicVisitorInterest> interests, TopicVisitor
writer.handleSnapshot(image, (opType, opLog, operation) ->
operation.apply(ZkMigrationLeadershipState.EMPTY)
);
- assertEquals(topicClient.updatedTopics.get("foo").size(), 3);
+ assertEquals(3, topicClient.updatedTopics.get("foo").size());
+ assertEquals(topicClient.deletedTopicPartitions.get("foo"), Collections.singleton(3));
+ // Partition 1 is missing from ZK, so it must be created rather than updated
+ assertEquals(topicClient.newTopicPartitions.get("foo"), Collections.singleton(1));
Review Comment:
`Collections.singleton(1)` should be the first argument
##########
metadata/src/test/java/org/apache/kafka/metadata/migration/KRaftMigrationZkWriterTest.java:
##########
@@ -98,9 +98,66 @@ public void iterateTopics(EnumSet<TopicVisitorInterest> interests, TopicVisitor
writer.handleSnapshot(image, (opType, opLog, operation) ->
operation.apply(ZkMigrationLeadershipState.EMPTY)
);
- assertEquals(topicClient.updatedTopics.get("foo").size(), 3);
+ assertEquals(3, topicClient.updatedTopics.get("foo").size());
+ assertEquals(topicClient.deletedTopicPartitions.get("foo"), Collections.singleton(3));
Review Comment:
ditto
##########
core/src/main/scala/kafka/zk/migration/ZkTopicMigrationClient.scala:
##########
@@ -77,26 +78,31 @@ class ZkTopicMigrationClient(zkClient: KafkaZkClient) extends TopicMigrationClie
.setReplicas(replicaList)
.setAddingReplicas(replicaAssignment.addingReplicas.map(Integer.valueOf).asJava)
.setRemovingReplicas(replicaAssignment.removingReplicas.map(Integer.valueOf).asJava)
- leaderIsrAndControllerEpochs.get(topicPartition) match {
+ val recordToVisit = leaderIsrAndControllerEpochs.get(topicPartition) match {
case Some(leaderIsrAndEpoch) =>
- record
+ Some(record
.setIsr(leaderIsrAndEpoch.leaderAndIsr.isr.map(Integer.valueOf).asJava)
.setLeader(leaderIsrAndEpoch.leaderAndIsr.leader)
.setLeaderEpoch(leaderIsrAndEpoch.leaderAndIsr.leaderEpoch)
.setPartitionEpoch(leaderIsrAndEpoch.leaderAndIsr.partitionEpoch)
- .setLeaderRecoveryState(leaderIsrAndEpoch.leaderAndIsr.leaderRecoveryState.value())
- case None =>
+ .setLeaderRecoveryState(leaderIsrAndEpoch.leaderAndIsr.leaderRecoveryState.value()))
+ case None if visitPartitionsWithoutState =>
Review Comment:
This fix makes sense to me, but the flag `visitPartitionsWithoutState` is a bit mysterious. Since synthesizing the record is the magic from `ZkMigrationClient`, maybe we can add a default method `visitPartitionWithoutState` to `TopicVisitor`? `ZkTopicMigrationClient` just calls it when the state is missing, and `ZkMigrationClient` can hide the magic inside its visitor.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org