dajac commented on a change in pull request #8933:
URL: https://github.com/apache/kafka/pull/8933#discussion_r456273174
##########
File path: core/src/main/scala/kafka/server/AdminManager.scala
##########
@@ -295,34 +312,44 @@ class AdminManager(val config: KafkaConfig,
throw new InvalidPartitionsException(s"Topic already has
$oldNumPartitions partitions.")
}
- val newPartitionsAssignment = Option(newPartition.assignments)
- .map { assignmentMap =>
- val assignments = assignmentMap.asScala.map {
- createPartitionAssignment =>
createPartitionAssignment.brokerIds.asScala.map(_.toInt)
- }
- val unknownBrokers = assignments.flatten.toSet -- allBrokerIds
- if (unknownBrokers.nonEmpty)
- throw new InvalidReplicaAssignmentException(
- s"Unknown broker(s) in replica assignment:
${unknownBrokers.mkString(", ")}.")
-
- if (assignments.size != numPartitionsIncrement)
- throw new InvalidReplicaAssignmentException(
- s"Increasing the number of partitions by
$numPartitionsIncrement " +
- s"but ${assignments.size} assignments provided.")
-
- assignments.zipWithIndex.map { case (replicas, index) =>
- existingAssignment.size + index -> replicas
- }.toMap
+ val newPartitionsAssignment = Option(newPartition.assignments).map {
assignmentMap =>
+ val assignments = assignmentMap.asScala.map {
+ createPartitionAssignment =>
createPartitionAssignment.brokerIds.asScala.map(_.toInt)
+ }
+ val unknownBrokers = assignments.flatten.toSet -- allBrokerIds
+ if (unknownBrokers.nonEmpty)
+ throw new InvalidReplicaAssignmentException(
+ s"Unknown broker(s) in replica assignment:
${unknownBrokers.mkString(", ")}.")
+
+ if (assignments.size != numPartitionsIncrement)
+ throw new InvalidReplicaAssignmentException(
+ s"Increasing the number of partitions by $numPartitionsIncrement
" +
+ s"but ${assignments.size} assignments provided.")
+
+ assignments.zipWithIndex.map { case (replicas, index) =>
+ existingAssignment.size + index -> replicas
+ }.toMap
}
- val updatedReplicaAssignment = adminZkClient.addPartitions(topic,
existingAssignment, allBrokers,
- newPartition.count, newPartitionsAssignment, validateOnly =
validateOnly)
- CreatePartitionsMetadata(topic, updatedReplicaAssignment.keySet,
ApiError.NONE)
+ val assignmentForNewPartitions =
adminZkClient.createNewPartitionsAssignment(
+ topic, existingAssignment, allBrokers, newPartition.count,
newPartitionsAssignment)
+
+ if (validateOnly) {
+ CreatePartitionsMetadata(topic, (existingAssignment ++
assignmentForNewPartitions).keySet)
Review comment:
That's a great question. If I remember correctly, someone brought this
up in the DISCUSS thread a while back.
The main issue is that we can't use the `throttleTimeMs` field to give back
this information to the client because the java client would consider this as a
quota violation and would throttle the connection. If we want to do this, we
would need to find another way to convey this to the client. I don't think that
it is worth it.
Moreover, as the validation of the request doesn't not entail any creations
nor deletions, I would find strange to actually return
`ThrottlingQuotaExceededException` with a `throttleTimeMs` to the client.
`ThrottlingQuotaExceededException` would actually mean "the validation has
passed but you would have been throttled if you would have executed that
request" which is not intuitive if you care about validating the request only.
----------------------------------------------------------------
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
