cmccabe commented on a change in pull request #10343:
URL: https://github.com/apache/kafka/pull/10343#discussion_r609098540
##########
File path:
metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
##########
@@ -518,4 +523,89 @@ public void testDeleteTopics() throws Exception {
Long.MAX_VALUE, Collections.singleton("foo")));
assertEmptyTopicConfigs(ctx, "foo");
}
+
+
+ @Test
+ public void testCreatePartitions() throws Exception {
+ ReplicationControlTestContext ctx = new
ReplicationControlTestContext();
+ ReplicationControlManager replicationControl = ctx.replicationControl;
+ CreateTopicsRequestData request = new CreateTopicsRequestData();
+ request.topics().add(new CreatableTopic().setName("foo").
+ setNumPartitions(3).setReplicationFactor((short) 2));
+ request.topics().add(new CreatableTopic().setName("bar").
+ setNumPartitions(4).setReplicationFactor((short) 2));
+ request.topics().add(new CreatableTopic().setName("quux").
+ setNumPartitions(2).setReplicationFactor((short) 2));
+ request.topics().add(new CreatableTopic().setName("foo2").
+ setNumPartitions(2).setReplicationFactor((short) 2));
+ registerBroker(0, ctx);
+ unfenceBroker(0, ctx);
+ registerBroker(1, ctx);
+ unfenceBroker(1, ctx);
+ ControllerResult<CreateTopicsResponseData> createTopicResult =
+ replicationControl.createTopics(request);
+ ctx.replay(createTopicResult.records());
+ List<CreatePartitionsTopic> topics = new ArrayList<>();
+ topics.add(new CreatePartitionsTopic().
+ setName("foo").setCount(5).setAssignments(null));
+ topics.add(new CreatePartitionsTopic().
+ setName("bar").setCount(3).setAssignments(null));
+ topics.add(new CreatePartitionsTopic().
+ setName("baz").setCount(3).setAssignments(null));
+ topics.add(new CreatePartitionsTopic().
+ setName("quux").setCount(2).setAssignments(null));
+ ControllerResult<List<CreatePartitionsTopicResult>>
createPartitionsResult =
+ replicationControl.createPartitions(topics);
+ assertEquals(Arrays.asList(new CreatePartitionsTopicResult().
+ setName("foo").
+ setErrorCode(NONE.code()).
+ setErrorMessage(null),
+ new CreatePartitionsTopicResult().
+ setName("bar").
+ setErrorCode(INVALID_PARTITIONS.code()).
+ setErrorMessage("The topic bar currently has 4 partition(s); 3
would not be an increase."),
+ new CreatePartitionsTopicResult().
+ setName("baz").
+ setErrorCode(UNKNOWN_TOPIC_OR_PARTITION.code()).
+ setErrorMessage(null),
+ new CreatePartitionsTopicResult().
+ setName("quux").
+ setErrorCode(INVALID_PARTITIONS.code()).
+ setErrorMessage("Topic already has 2 partition(s).")),
+ createPartitionsResult.response());
+ ctx.replay(createPartitionsResult.records());
+ List<CreatePartitionsTopic> topics2 = new ArrayList<>();
+ topics2.add(new CreatePartitionsTopic().
Review comment:
Part of the goal here is to verify that the batch operation works,
though. If all the cases were split into separate functions we would not get
that validation.
--
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.
For queries about this service, please contact Infrastructure at:
[email protected]
