jsancio commented on a change in pull request #11216:
URL: https://github.com/apache/kafka/pull/11216#discussion_r689874304
##########
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:
> There is an issue with the test where I already included the new topic
foo in the image before applying the delta...
I think the reason why this inconsistency workout for the test is because
`TopicsDelta` and `TopicDelta` use topic id to figure out the changes in the
`apply` method.
--
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
