cmccabe commented on a change in pull request #10343:
URL: https://github.com/apache/kafka/pull/10343#discussion_r612656387
##########
File path:
metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -1007,6 +999,128 @@ int bestLeader(int[] replicas, int[] isr, boolean
unclean) {
return ControllerResult.of(records, null);
}
+ ControllerResult<List<CreatePartitionsTopicResult>>
+ createPartitions(List<CreatePartitionsTopic> topics) {
+ List<ApiMessageAndVersion> records = new ArrayList<>();
+ List<CreatePartitionsTopicResult> results = new ArrayList<>();
+ for (CreatePartitionsTopic topic : topics) {
+ ApiError apiError = ApiError.NONE;
+ try {
+ createPartitions(topic, records);
+ } catch (ApiException e) {
+ apiError = ApiError.fromThrowable(e);
+ } catch (Exception e) {
+ log.error("Unexpected createPartitions error for {}", topic,
e);
+ apiError = ApiError.fromThrowable(e);
+ }
+ results.add(new CreatePartitionsTopicResult().
+ setName(topic.name()).
+ setErrorCode(apiError.error().code()).
+ setErrorMessage(apiError.message()));
+ }
+ return new ControllerResult<>(records, results, true);
+ }
+
+ void createPartitions(CreatePartitionsTopic topic,
+ List<ApiMessageAndVersion> records) {
+ Uuid topicId = topicsByName.get(topic.name());
+ if (topicId == null) {
+ throw new UnknownTopicOrPartitionException();
+ }
+ TopicControlInfo topicInfo = topics.get(topicId);
+ if (topicInfo == null) {
+ throw new UnknownTopicOrPartitionException();
+ }
+ if (topic.count() == topicInfo.parts.size()) {
Review comment:
Yeah, for now I'm just being consistent with the current implementation.
I think eventually we will want a retransmission cache to solve all these
problems at once.
--
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:
us...@infra.apache.org
