You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2026/10/16 19:39:07 UTC
(kafka) branch 3.9 updated: KAFKA-21142 fix partition creation when handling snapshot delta during migration (#23610)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch 3.9
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.9 by this push:
new daebfbd9b37 KAFKA-21142 fix partition creation when handling snapshot delta during migration (#23610)
daebfbd9b37 is described below
commit daebfbd9b37c913937d5adfc2e493fe825e3da3a
Author: Gaurav Narula <ga...@apple.com>
AuthorDate: Fri Oct 16 20:38:58 2026 +0100
KAFKA-21142 fix partition creation when handling snapshot delta during migration (#23610)
A quorum leader crash during a KRaft migration can result in a scenario
where the `/state` ZNode for a topic-partition is not created after the
new leader is elected due to
`migrationClient.topicClient().updateTopicPartitions()` being invoked
instead of `migrationClient.topicClient().createTopicPartitions()`
This change updates the callsite in
`KraftMigrationZkWriter#handleSnapshot` and introduces a new variant to
`TopicMigrationClient#TopicVisitorInterest` enum to restrict the change
to when KRaft controller is propagating changes to ZK and keep the
existing behaviour prior to that.
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
.../main/scala/kafka/zk/ZkMigrationClient.scala | 21 ++-
.../zk/migration/ZkTopicMigrationClient.scala | 34 ++---
.../kafka/zk/migration/ZkMigrationClientTest.scala | 162 ++++++++++++++++++++-
.../metadata/migration/KRaftMigrationZkWriter.java | 6 +-
.../metadata/migration/TopicMigrationClient.java | 18 +++
.../migration/CapturingTopicMigrationClient.java | 1 +
.../migration/KRaftMigrationZkWriterTest.java | 65 ++++++++-
7 files changed, 277 insertions(+), 30 deletions(-)
diff --git a/core/src/main/scala/kafka/zk/ZkMigrationClient.scala b/core/src/main/scala/kafka/zk/ZkMigrationClient.scala
index 20afc9fb0c4..85953a321cf 100644
--- a/core/src/main/scala/kafka/zk/ZkMigrationClient.scala
+++ b/core/src/main/scala/kafka/zk/ZkMigrationClient.scala
@@ -28,8 +28,7 @@ import org.apache.kafka.common.metadata._
import org.apache.kafka.common.resource.ResourcePattern
import org.apache.kafka.common.security.scram.ScramCredential
import org.apache.kafka.common.{TopicIdPartition, Uuid}
-import org.apache.kafka.metadata.DelegationTokenData
-import org.apache.kafka.metadata.PartitionRegistration
+import org.apache.kafka.metadata.{DelegationTokenData, LeaderRecoveryState, PartitionRegistration}
import org.apache.kafka.metadata.migration.ConfigMigrationClient.ClientQuotaVisitor
import org.apache.kafka.metadata.migration.TopicMigrationClient.{TopicVisitor, TopicVisitorInterest}
import org.apache.kafka.metadata.migration._
@@ -188,6 +187,24 @@ class ZkMigrationClient(
partitionRegistration.addingReplicas.foreach(brokerIdConsumer.accept(_))
topicBatch.add(new ApiMessageAndVersion(record, 0.toShort))
}
+
+
+ /**
+ * ZooKeeper is the source of truth during the migration, so partitions without any state in ZK are
+ * migrated with a synthetic leader and ISR rather than being left out of the migrated metadata.
+ */
+ override def visitPartitionWithoutState(topicIdPartition: TopicIdPartition, partitionRecord: PartitionRecord): Unit = {
+ val replicaList = partitionRecord.replicas()
+ warn(s"Initializing partition $topicIdPartition with ISR={$replicaList} and leaderEpoch=0 since it " +
+ "has no partition state in ZK.")
+ partitionRecord
+ .setIsr(replicaList)
+ .setLeader(replicaList.get(0))
+ .setLeaderEpoch(0)
+ .setPartitionEpoch(0)
+ .setLeaderRecoveryState(LeaderRecoveryState.RECOVERED.value())
+ visitPartition(topicIdPartition, new PartitionRegistration(partitionRecord))
+ }
}
)
diff --git a/core/src/main/scala/kafka/zk/migration/ZkTopicMigrationClient.scala b/core/src/main/scala/kafka/zk/migration/ZkTopicMigrationClient.scala
index bba2aecae0a..ed5a4b1d874 100644
--- a/core/src/main/scala/kafka/zk/migration/ZkTopicMigrationClient.scala
+++ b/core/src/main/scala/kafka/zk/migration/ZkTopicMigrationClient.scala
@@ -28,7 +28,7 @@ import org.apache.kafka.common.metadata.PartitionRecord
import org.apache.kafka.common.{TopicIdPartition, TopicPartition, Uuid}
import org.apache.kafka.metadata.migration.TopicMigrationClient.TopicVisitorInterest
import org.apache.kafka.metadata.migration.{MigrationClientException, TopicMigrationClient, ZkMigrationLeadershipState}
-import org.apache.kafka.metadata.{LeaderRecoveryState, PartitionRegistration}
+import org.apache.kafka.metadata.PartitionRegistration
import org.apache.kafka.server.config.ConfigType
import org.apache.zookeeper.CreateMode
import org.apache.zookeeper.KeeperException.Code
@@ -77,26 +77,20 @@ 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 {
- case Some(leaderIsrAndEpoch) =>
- 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 =>
- warn(s"Could not find partition state in ZK for $topicPartition. Initializing this partition " +
- s"with ISR={$replicaList} and leaderEpoch=0.")
- record
- .setIsr(replicaList)
- .setLeader(replicaList.get(0))
- .setLeaderEpoch(0)
- .setPartitionEpoch(0)
- .setLeaderRecoveryState(LeaderRecoveryState.RECOVERED.value())
- }
logAndRethrow(this, s"Error in partition consumer. TopicPartition was $topicPartition.") {
- visitor.visitPartition(new TopicIdPartition(topicIdOpt.get, topicPartition), new PartitionRegistration(record))
+ leaderIsrAndControllerEpochs.get(topicPartition) match {
+ case Some(leaderIsrAndEpoch) =>
+ 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())
+ visitor.visitPartition(new TopicIdPartition(topicIdOpt.get, topicPartition), new PartitionRegistration(record))
+ case None =>
+ warn(s"Could not find partition state in ZK for $topicPartition.")
+ visitor.visitPartitionWithoutState(new TopicIdPartition(topicIdOpt.get, topicPartition), record)
+ }
}
}
}
diff --git a/core/src/test/scala/unit/kafka/zk/migration/ZkMigrationClientTest.scala b/core/src/test/scala/unit/kafka/zk/migration/ZkMigrationClientTest.scala
index bc4b5ec6944..d650412ac15 100644
--- a/core/src/test/scala/unit/kafka/zk/migration/ZkMigrationClientTest.scala
+++ b/core/src/test/scala/unit/kafka/zk/migration/ZkMigrationClientTest.scala
@@ -22,8 +22,9 @@ import kafka.coordinator.transaction.{ProducerIdManager, ZkProducerIdManager}
import org.apache.kafka.common.config.{ConfigResource, SslConfigs, TopicConfig}
import org.apache.kafka.common.errors.ControllerMovedException
import org.apache.kafka.common.metadata.{ConfigRecord, MetadataRecordType, PartitionRecord, ProducerIdsRecord, TopicRecord}
-import org.apache.kafka.common.{DirectoryId, TopicPartition, Uuid}
+import org.apache.kafka.common.{DirectoryId, TopicIdPartition, TopicPartition, Uuid}
import org.apache.kafka.image.{MetadataDelta, MetadataImage, MetadataProvenance}
+import org.apache.kafka.metadata.migration.TopicMigrationClient.{TopicVisitor, TopicVisitorInterest}
import org.apache.kafka.metadata.migration.{KRaftMigrationZkWriter, ZkMigrationLeadershipState}
import org.apache.kafka.metadata.{LeaderRecoveryState, PartitionRegistration}
import org.apache.kafka.server.config.ReplicationConfigs
@@ -33,7 +34,7 @@ import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows, assertTrue,
import org.junit.jupiter.api.Test
import java.util.Properties
-import scala.collection.Map
+import scala.collection.{Map, mutable}
import scala.jdk.CollectionConverters._
import scala.util.{Failure, Success}
@@ -513,4 +514,161 @@ class ZkMigrationClientTest extends ZkMigrationTestHarness {
assertEquals(expectedPartition.isr.toList, part.leaderAndIsr.isr)
}
}
+
+ /**
+ * A partition can be present in a topic's replica assignment without having any partition state, for
+ * instance when a controller fails in between writing the two. Verify that such partitions are visited
+ * through visitPartitionWithoutState rather than visitPartition, and that the other partitions and topics
+ * are visited as usual.
+ */
+ @Test
+ def testIterateTopicsWithMissingPartitionState(): Unit = {
+ val topicNames = createTopicsWithMissingPartitionState()
+
+ val visitor = new CapturingTopicVisitor()
+ migrationClient.topicClient().iterateTopics(
+ java.util.EnumSet.of(TopicVisitorInterest.TOPICS, TopicVisitorInterest.PARTITIONS),
+ visitor)
+ assertEquals(topicNames.toSet, visitor.visitedTopics.toSet)
+ assertEquals(
+ topicNames.map(new TopicPartition(_, 0)).toSet,
+ visitor.visitedPartitions.keySet.toSet)
+ assertEquals(
+ topicNames.map(new TopicPartition(_, 1)).toSet,
+ visitor.visitedPartitionsWithoutState.keySet.toSet)
+ topicNames.foreach { topicName =>
+ val record = visitor.visitedPartitionsWithoutState(new TopicPartition(topicName, 1))
+ assertEquals(1, record.partitionId)
+ assertEquals(Seq(1, 2, 3), record.replicas.asScala.map(_.toInt))
+ }
+ }
+
+ /**
+ * ZooKeeper is the source of truth during the migration, so a partition without any state in ZK must be
+ * migrated with a synthetic leader and ISR rather than being left out of the migrated metadata.
+ */
+ @Test
+ def testMigrateTopicsWithMissingPartitionState(): Unit = {
+ val topicNames = createTopicsWithMissingPartitionState()
+
+ val batches = new java.util.ArrayList[java.util.List[ApiMessageAndVersion]]()
+ migrationClient.readAllMetadata(batch => batches.add(batch), _ => ())
+ val partitionRecords = batches.asScala.flatMap(_.asScala).map(_.message).collect {
+ case record: PartitionRecord => record
+ }
+ val topicIds = zkClient.getTopicIdsForTopics(topicNames.toSet)
+
+ topicNames.foreach { topicName =>
+ val topicRecords = partitionRecords.filter(_.topicId == topicIds(topicName))
+ assertEquals(Set(0, 1), topicRecords.map(_.partitionId).toSet)
+ val synthesized = topicRecords.find(_.partitionId == 1).get
+ assertEquals(1, synthesized.leader)
+ assertEquals(Seq(1, 2, 3), synthesized.isr.asScala.map(_.toInt))
+ assertEquals(0, synthesized.leaderEpoch)
+ assertEquals(0, synthesized.partitionEpoch)
+ assertEquals(LeaderRecoveryState.RECOVERED.value, synthesized.leaderRecoveryState)
+ }
+ }
+
+ /**
+ * Create two topics with two partitions each, where only partition 0 of each topic has partition state.
+ */
+ private def createTopicsWithMissingPartitionState(): Seq[String] = {
+ val topicNames = Seq("test-a", "test-b")
+ topicNames.foreach { topicName =>
+ val assignment = Map(
+ new TopicPartition(topicName, 0) -> List(0, 1, 2),
+ new TopicPartition(topicName, 1) -> List(1, 2, 3)
+ )
+ zkClient.createTopicAssignment(topicName, Some(Uuid.randomUuid()), assignment)
+ zkClient.createTopicPartitionStatesRaw(Map(
+ new TopicPartition(topicName, 0) -> LeaderIsrAndControllerEpoch(
+ LeaderAndIsr(0, 5, List(0, 1, 2), LeaderRecoveryState.RECOVERED, -1), 1)
+ ), 0)
+ }
+ topicNames
+ }
+
+ /**
+ * Simulate a controller that failed after writing a new partition to the topic assignment in ZK, but
+ * before writing that partition's state. The next controller must create the missing partition state
+ * when it syncs the KRaft state to ZK (KAFKA-21142).
+ */
+ @Test
+ def testSnapshotCreatesMissingPartitionState(): Unit = {
+ val topicName = "test"
+ val topicId = Uuid.randomUuid()
+ val newPartition = new TopicPartition(topicName, 2)
+ val assignment = Map(
+ new TopicPartition(topicName, 0) -> List(0, 1, 2),
+ new TopicPartition(topicName, 1) -> List(1, 2, 3),
+ newPartition -> List(2, 3, 4)
+ )
+ zkClient.createTopicAssignment(topicName, Some(topicId), assignment)
+ zkClient.createTopicPartitionStatesRaw(Map(
+ new TopicPartition(topicName, 0) -> LeaderIsrAndControllerEpoch(
+ LeaderAndIsr(0, 5, List(0, 1, 2), LeaderRecoveryState.RECOVERED, -1), 1),
+ new TopicPartition(topicName, 1) -> LeaderIsrAndControllerEpoch(
+ LeaderAndIsr(1, 5, List(1, 2, 3), LeaderRecoveryState.RECOVERED, -1), 1)
+ ), 0)
+ assertTrue(zkClient.getTopicPartitionState(newPartition).isEmpty)
+
+ val delta = new MetadataDelta(MetadataImage.EMPTY)
+ delta.replay(new TopicRecord().setTopicId(topicId).setName(topicName))
+ assignment.foreach { case (topicPartition, replicas) =>
+ delta.replay(new PartitionRecord()
+ .setTopicId(topicId)
+ .setPartitionId(topicPartition.partition())
+ .setReplicas(replicas.map(int2Integer).asJava)
+ .setAddingReplicas(List.empty.asJava)
+ .setRemovingReplicas(List.empty.asJava)
+ .setIsr(replicas.map(int2Integer).asJava)
+ .setLeader(replicas.head)
+ .setLeaderEpoch(5)
+ .setPartitionEpoch(10)
+ .setLeaderRecoveryState(LeaderRecoveryState.RECOVERED.value()))
+ }
+
+ val kraftWriter = new KRaftMigrationZkWriter(migrationClient, fail(_))
+ kraftWriter.handleSnapshot(delta.apply(MetadataProvenance.EMPTY), (_, _, operation) => {
+ migrationState = operation(migrationState)
+ })
+
+ val newPartitionState = zkClient.getTopicPartitionState(newPartition)
+ assertTrue(newPartitionState.isDefined, s"Expected $newPartition to have been created in ZK")
+ newPartitionState.foreach { state =>
+ assertEquals(2, state.leaderAndIsr.leader)
+ assertEquals(5, state.leaderAndIsr.leaderEpoch)
+ assertEquals(List(2, 3, 4), state.leaderAndIsr.isr)
+ assertEquals(LeaderRecoveryState.RECOVERED, state.leaderAndIsr.leaderRecoveryState)
+ }
+ }
+
+ private class CapturingTopicVisitor extends TopicVisitor {
+ val visitedTopics = new mutable.ArrayBuffer[String]()
+ val visitedPartitions = new mutable.HashMap[TopicPartition, PartitionRegistration]()
+ val visitedPartitionsWithoutState = new mutable.HashMap[TopicPartition, PartitionRecord]()
+
+ override def visitTopic(
+ topicName: String,
+ topicId: Uuid,
+ assignments: java.util.Map[Integer, java.util.List[Integer]]
+ ): Unit = {
+ visitedTopics += topicName
+ }
+
+ override def visitPartition(
+ topicIdPartition: TopicIdPartition,
+ partitionRegistration: PartitionRegistration
+ ): Unit = {
+ visitedPartitions.put(topicIdPartition.topicPartition(), partitionRegistration)
+ }
+
+ override def visitPartitionWithoutState(
+ topicIdPartition: TopicIdPartition,
+ partitionRecord: PartitionRecord
+ ): Unit = {
+ visitedPartitionsWithoutState.put(topicIdPartition.topicPartition(), partitionRecord)
+ }
+ }
}
diff --git a/metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationZkWriter.java b/metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationZkWriter.java
index 702889df48b..3dda59f4429 100644
--- a/metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationZkWriter.java
+++ b/metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationZkWriter.java
@@ -214,7 +214,9 @@ public class KRaftMigrationZkWriter {
Map<Integer, PartitionRegistration> newTopicPartitions = new HashMap<>(topic.partitions());
// Compute KRaft partitions that are not in ZK
topicPartitionsInZk.forEach(newTopicPartitions::remove);
- newPartitions.put(topicId, newTopicPartitions);
+ if (!newTopicPartitions.isEmpty()) {
+ newPartitions.put(topicId, newTopicPartitions);
+ }
// Compute ZK partitions that are not in KRaft
topicPartitionsInZk.removeAll(topic.partitions().keySet());
@@ -256,7 +258,7 @@ public class KRaftMigrationZkWriter {
operationConsumer.accept(
UPDATE_PARTITION,
"Created additional partitions for Topic " + topic.name() + ", ID " + topicId,
- migrationState -> migrationClient.topicClient().updateTopicPartitions(
+ migrationState -> migrationClient.topicClient().createTopicPartitions(
Collections.singletonMap(topic.name(), partitionMap),
migrationState));
});
diff --git a/metadata/src/main/java/org/apache/kafka/metadata/migration/TopicMigrationClient.java b/metadata/src/main/java/org/apache/kafka/metadata/migration/TopicMigrationClient.java
index 5eafd72b298..11a62210154 100644
--- a/metadata/src/main/java/org/apache/kafka/metadata/migration/TopicMigrationClient.java
+++ b/metadata/src/main/java/org/apache/kafka/metadata/migration/TopicMigrationClient.java
@@ -19,6 +19,7 @@ package org.apache.kafka.metadata.migration;
import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.metadata.PartitionRecord;
import org.apache.kafka.metadata.PartitionRegistration;
import java.util.EnumSet;
@@ -38,6 +39,23 @@ public interface TopicMigrationClient {
default void visitPartition(TopicIdPartition topicIdPartition, PartitionRegistration partitionRegistration) {
}
+
+ /**
+ * Visit a partition which is included in a topic's replica assignment, but which has no partition state
+ * (leader and ISR) in ZooKeeper. Such partitions are left behind when a controller fails in between
+ * writing the topic assignment and the partition states.
+ * <p>
+ * The given record only has the topic ID, partition ID and replica assignment set. Callers for which
+ * ZooKeeper is the source of truth (i.e. the ZK to KRaft migration) should override this method, since
+ * skipping such a partition would drop it from the migrated metadata. Callers for which KRaft is the
+ * source of truth (i.e. the KRaft to ZK dual-write) should not, since the missing partition state is
+ * meant to be (re)created from the KRaft state rather than inferred from ZooKeeper.
+ * <p>
+ * Only invoked if {@link TopicVisitorInterest#PARTITIONS} is included in the interests.
+ */
+ default void visitPartitionWithoutState(TopicIdPartition topicIdPartition, PartitionRecord partitionRecord) {
+
+ }
}
void iterateTopics(EnumSet<TopicVisitorInterest> interests, TopicVisitor visitor);
diff --git a/metadata/src/test/java/org/apache/kafka/metadata/migration/CapturingTopicMigrationClient.java b/metadata/src/test/java/org/apache/kafka/metadata/migration/CapturingTopicMigrationClient.java
index e3b7cede7fc..619073f09a5 100644
--- a/metadata/src/test/java/org/apache/kafka/metadata/migration/CapturingTopicMigrationClient.java
+++ b/metadata/src/test/java/org/apache/kafka/metadata/migration/CapturingTopicMigrationClient.java
@@ -39,6 +39,7 @@ public class CapturingTopicMigrationClient implements TopicMigrationClient {
public void reset() {
createdTopics.clear();
+ newTopicPartitions.clear();
updatedTopicPartitions.clear();
deletedTopics.clear();
updatedTopics.clear();
diff --git a/metadata/src/test/java/org/apache/kafka/metadata/migration/KRaftMigrationZkWriterTest.java b/metadata/src/test/java/org/apache/kafka/metadata/migration/KRaftMigrationZkWriterTest.java
index 969914ca631..60c30c4c90a 100644
--- a/metadata/src/test/java/org/apache/kafka/metadata/migration/KRaftMigrationZkWriterTest.java
+++ b/metadata/src/test/java/org/apache/kafka/metadata/migration/KRaftMigrationZkWriterTest.java
@@ -57,7 +57,7 @@ public class KRaftMigrationZkWriterTest {
public void iterateTopics(EnumSet<TopicVisitorInterest> interests, TopicVisitor visitor) {
Map<Integer, List<Integer>> assignments = new HashMap<>();
assignments.put(0, Arrays.asList(2, 3, 4));
- assignments.put(1, Arrays.asList(3, 4, 5));
+ assignments.put(1, Arrays.asList(3, 4, 5)); // This one is not in ZK but in KRaft
assignments.put(2, Arrays.asList(2, 4, 5));
assignments.put(3, Arrays.asList(1, 2, 3)); // This one is not in KRaft
visitor.visitTopic("foo", TopicsImageTest.FOO_UUID, assignments);
@@ -98,9 +98,66 @@ public class KRaftMigrationZkWriterTest {
writer.handleSnapshot(image, (opType, opLog, operation) ->
operation.apply(ZkMigrationLeadershipState.EMPTY)
);
- assertEquals(topicClient.updatedTopics.get("foo").size(), 3);
- assertEquals(topicClient.deletedTopicPartitions.get("foo"), Collections.singleton(3));
- assertEquals(topicClient.updatedTopicPartitions.get("foo"), Collections.singleton(1));
+ assertEquals(3, topicClient.updatedTopics.get("foo").size());
+ assertEquals(Collections.singleton(3), topicClient.deletedTopicPartitions.get("foo"));
+ // Partition 1 is missing from ZK, so it must be created rather than updated
+ assertEquals(Collections.singleton(1), topicClient.newTopicPartitions.get("foo"));
+ assertTrue(topicClient.updatedTopicPartitions.isEmpty());
+ }
+
+ /**
+ * If ZK has extraneous partitions but is not missing any, no partitions should be created.
+ */
+ @Test
+ public void testOnlyExtraneousZkPartitions() {
+ CapturingTopicMigrationClient topicClient = new CapturingTopicMigrationClient() {
+ @Override
+ public void iterateTopics(EnumSet<TopicVisitorInterest> interests, TopicVisitor visitor) {
+ Map<Integer, List<Integer>> assignments = new HashMap<>();
+ assignments.put(0, Arrays.asList(2, 3, 4));
+ assignments.put(1, Arrays.asList(3, 4, 5));
+ assignments.put(2, Arrays.asList(2, 4, 5));
+ assignments.put(3, Arrays.asList(1, 2, 3)); // This one is not in KRaft
+ visitor.visitTopic("foo", TopicsImageTest.FOO_UUID, assignments);
+
+ IntStream.of(0, 1, 2, 3).forEach(partitionId ->
+ visitor.visitPartition(
+ new TopicIdPartition(TopicsImageTest.FOO_UUID, new TopicPartition("foo", partitionId)),
+ TopicsImageTest.IMAGE1.getPartition(TopicsImageTest.FOO_UUID, partitionId)
+ )
+ );
+ }
+ };
+
+ CapturingConfigMigrationClient configClient = new CapturingConfigMigrationClient();
+ CapturingMigrationClient migrationClient = CapturingMigrationClient.newBuilder()
+ .setBrokersInZk(0)
+ .setTopicMigrationClient(topicClient)
+ .setConfigMigrationClient(configClient)
+ .build();
+
+ KRaftMigrationZkWriter writer = new KRaftMigrationZkWriter(migrationClient, __ -> { });
+
+ MetadataImage image = new MetadataImage(
+ MetadataProvenance.EMPTY,
+ FeaturesImage.EMPTY,
+ ClusterImage.EMPTY,
+ TopicsImageTest.IMAGE1, // This includes "foo" with 3 partitions
+ ConfigurationsImage.EMPTY,
+ ClientQuotasImage.EMPTY,
+ ProducerIdsImage.EMPTY,
+ AclsImage.EMPTY,
+ ScramImage.EMPTY,
+ DelegationTokenImage.EMPTY
+ );
+
+ writer.handleSnapshot(image, (opType, opLog, operation) ->
+ operation.apply(ZkMigrationLeadershipState.EMPTY)
+ );
+ assertEquals(3, topicClient.updatedTopics.get("foo").size());
+ assertEquals(Collections.singleton(3), topicClient.deletedTopicPartitions.get("foo"));
+ assertTrue(topicClient.newTopicPartitions.isEmpty());
+ assertTrue(topicClient.updatedTopicPartitions.isEmpty());
}
/**