cmccabe commented on a change in pull request #10184:
URL: https://github.com/apache/kafka/pull/10184#discussion_r583084940
##########
File path: core/src/main/scala/kafka/server/ControllerApis.scala
##########
@@ -195,6 +201,147 @@ class ControllerApis(val requestChannel: RequestChannel,
requestThrottleMs => createResponseCallback(requestThrottleMs))
}
+ def handleDeleteTopics(request: RequestChannel.Request): Unit = {
+ val responses = deleteTopics(request.body[DeleteTopicsRequest].data(),
+ request.context.apiVersion(),
+ authHelper.authorize(request.context, DELETE, CLUSTER, CLUSTER_NAME),
+ names => authHelper.filterByAuthorized(request.context, DESCRIBE, TOPIC,
names)(n => n),
+ names => authHelper.filterByAuthorized(request.context, DELETE, TOPIC,
names)(n => n))
+ requestHelper.sendResponseMaybeThrottle(request, throttleTimeMs => {
+ val responseData = new DeleteTopicsResponseData().
+ setResponses(new DeletableTopicResultCollection(responses.iterator())).
+ setThrottleTimeMs(throttleTimeMs)
+ new DeleteTopicsResponse(responseData)
+ })
+ }
+
+ def deleteTopics(request: DeleteTopicsRequestData,
+ apiVersion: Int,
+ hasClusterAuth: Boolean,
+ getDescribableTopics: Iterable[String] => Set[String],
+ getDeletableTopics: Iterable[String] => Set[String]):
util.List[DeletableTopicResult] = {
+ if (!config.deleteTopicEnable) {
+ if (apiVersion < 3) {
+ throw new InvalidRequestException("Topic deletion is disabled.")
+ } else {
+ throw new TopicDeletionDisabledException()
+ }
+ }
+ val responses = new util.ArrayList[DeletableTopicResult]
+ val duplicatedTopicNames = new util.HashSet[String]
+ val topicNamesToResolve = new util.HashSet[String]
+ val topicIdsToResolve = new util.HashSet[Uuid]
+ val duplicatedTopicIds = new util.HashSet[Uuid]
+
+ def appendResponse(name: String, id: Uuid, error: ApiError): Unit = {
+ responses.add(new DeletableTopicResult().
+ setName(name).
+ setTopicId(id).
+ setErrorCode(error.error().code()).
+ setErrorMessage(error.message()))
+ }
+
+ def maybeAppendToTopicNamesToResolve(name: String): Unit = {
+ if (duplicatedTopicNames.contains(name) ||
!topicNamesToResolve.add(name)) {
+ appendResponse(name, ZERO_UUID, new ApiError(INVALID_REQUEST,
"Duplicate topic name."))
+ topicNamesToResolve.remove(name)
+ duplicatedTopicNames.add(name)
+ }
+ }
+
+ def maybeAppendToIdsToResolve(id: Uuid): Unit = {
+ if (duplicatedTopicIds.contains(id) || !topicIdsToResolve.add(id)) {
+ appendResponse(null, id, new ApiError(INVALID_REQUEST, "Duplicate
topic ID."))
+ topicIdsToResolve.remove(id)
+ duplicatedTopicIds.add(id)
+ }
+ }
+
+
request.topicNames().iterator().asScala.foreach(maybeAppendToTopicNamesToResolve(_))
+
+ request.topics().iterator().asScala.foreach {
+ case topic => if (topic.name() == null) {
Review comment:
ok
----------------------------------------------------------------
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
