jsancio commented on a change in pull request #11216:
URL: https://github.com/apache/kafka/pull/11216#discussion_r689871542
##########
File path: metadata/src/test/java/org/apache/kafka/image/TopicsImageTest.java
##########
@@ -126,6 +133,158 @@ private static TopicImage newTopicImage(String name, Uuid
id, PartitionRegistrat
IMAGE2 = new TopicsImage(newTopicsByIdMap(topics2),
newTopicsByNameMap(topics2));
}
+ private PartitionRegistration newPartition(int[] replicas) {
+ return new PartitionRegistration(replicas, replicas, Replicas.NONE,
Replicas.NONE, replicas[0], 1, 1);
+ }
+
+ @Test
+ public void testLocalReplicaChanges() {
+ int localId = 3;
+ Uuid newFooId = Uuid.fromString("0hHJ3X5ZQ-CFfQ5xgpj90w");
+
+ List<TopicImage> topics = new ArrayList<>(TOPIC_IMAGES1);
+ topics.add(
+ newTopicImage(
+ "foo",
+ newFooId,
+ newPartition(new int[] {0, 1, 3}),
+ newPartition(new int[] {3, 1, 2}),
+ newPartition(new int[] {0, 1, 3}),
+ newPartition(new int[] {3, 1, 2}),
+ newPartition(new int[] {0, 1, 2}),
+ newPartition(new int[] {0, 1, 2})
+ )
+ );
+
+ TopicsImage image = new TopicsImage(newTopicsByIdMap(topics),
newTopicsByNameMap(topics));
+
+ List<ApiMessageAndVersion> topicRecords = new
ArrayList<>(DELTA1_RECORDS);
+ // foo-0 - follower to leader
+ topicRecords.add(
+ new ApiMessageAndVersion(
+ new PartitionChangeRecord()
+ .setTopicId(newFooId)
+ .setPartitionId(0)
+ .setLeader(3),
+ PARTITION_CHANGE_RECORD.highestSupportedVersion()
+ )
+ );
+ // foo-1 - leader to follower
+ topicRecords.add(
+ new ApiMessageAndVersion(
+ new PartitionChangeRecord()
+ .setTopicId(newFooId)
+ .setPartitionId(1)
+ .setLeader(1),
+ PARTITION_CHANGE_RECORD.highestSupportedVersion()
+ )
+ );
+ // foo-2 - follower to removed
+ topicRecords.add(
+ new ApiMessageAndVersion(
+ new PartitionChangeRecord()
+ .setTopicId(newFooId)
+ .setPartitionId(2)
+ .setIsr(Arrays.asList(0, 1, 2))
+ .setReplicas(Arrays.asList(0, 1, 2)),
+ PARTITION_CHANGE_RECORD.highestSupportedVersion()
+ )
+ );
+ // foo-3 - leader to removed
+ topicRecords.add(
+ new ApiMessageAndVersion(
+ new PartitionChangeRecord()
+ .setTopicId(newFooId)
+ .setPartitionId(3)
+ .setLeader(0)
+ .setIsr(Arrays.asList(0, 1, 2))
+ .setReplicas(Arrays.asList(0, 1, 2)),
+ PARTITION_CHANGE_RECORD.highestSupportedVersion()
+ )
+ );
+ // foo-4 - not replica to leader
+ topicRecords.add(
+ new ApiMessageAndVersion(
+ new PartitionChangeRecord()
+ .setTopicId(newFooId)
+ .setPartitionId(4)
+ .setLeader(3)
+ .setIsr(Arrays.asList(3, 1, 2))
+ .setReplicas(Arrays.asList(3, 1, 2)),
+ PARTITION_CHANGE_RECORD.highestSupportedVersion()
+ )
+ );
+ // foo-5 - not replica to follower
+ topicRecords.add(
+ new ApiMessageAndVersion(
+ new PartitionChangeRecord()
+ .setTopicId(newFooId)
+ .setPartitionId(5)
+ .setIsr(Arrays.asList(0, 1, 3))
+ .setReplicas(Arrays.asList(0, 1, 3)),
+ PARTITION_CHANGE_RECORD.highestSupportedVersion()
+ )
+ );
+
+ /* Changes already include in DELTA1_RECORDS:
+ * foo - topic id deleted
+ * bar-0 - stay as follower with different partition epoch
+ * baz-0 - new topic to leader
+ */
+
+ // baz-1 - new topic to follower
+ topicRecords.add(
+ new ApiMessageAndVersion(
+ new PartitionRecord()
+ .setPartitionId(1)
+ .setTopicId(Uuid.fromString("tgHBnRglT5W_RlENnuG5vg"))
+ .setReplicas(Arrays.asList(4, 2, 3))
+ .setIsr(Arrays.asList(4, 2, 3))
+ .setLeader(4)
+ .setLeaderEpoch(2)
+ .setPartitionEpoch(1),
+ PARTITION_RECORD.highestSupportedVersion()
+ )
+ );
+
+ TopicsDelta delta = new TopicsDelta(image);
+ RecordTestUtils.replayAll(delta, topicRecords);
+
+ LocalReplicaChanges changes = delta.localChanges(localId);
+ assertEquals(
+ new HashSet<>(
+ Arrays.asList(
+ new TopicPartition("foo", 2),
+ new TopicPartition("foo", 3),
+ new TopicPartition("foo", 0), // from remove topic of the
old id
Review comment:
When I was working on this PR I thought about using `TopicIdPartition`
instead of `TopidPartition` to make this obvious but that is a bigger change to
the `ReplicaManager`. I can do that after 3.0.
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
