hachikuji commented on a change in pull request #10184:
URL: https://github.com/apache/kafka/pull/10184#discussion_r585086425
##########
File path: core/src/main/scala/kafka/server/ControllerApis.scala
##########
@@ -154,17 +162,179 @@ 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.forEach(maybeAppendToTopicNamesToResolve)
+
+ request.topics.forEach {
+ topic => if (topic.name == null) {
+ if (topic.topicId.equals(ZERO_UUID)) {
+ appendResponse(null, ZERO_UUID, new ApiError(INVALID_REQUEST,
+ "Neither topic name nor id were specified."))
+ } else {
+ maybeAppendToIdsToResolve(topic.topicId)
+ }
+ } else {
+ if (topic.topicId.equals(ZERO_UUID)) {
+ maybeAppendToTopicNamesToResolve(topic.name)
+ } else {
+ appendResponse(topic.name, topic.topicId, new
ApiError(INVALID_REQUEST,
+ "You may not specify both topic name and topic id."))
+ }
+ }
+ }
+
+ val idToName = new util.HashMap[Uuid, String]
+ val unknownTopicNameErrors = new util.HashMap[String, ApiError]
+ def maybeAppendToIdToName(id: Uuid, name: String): Unit = {
+ if (duplicatedTopicIds.contains(id) || idToName.put(id, name) != null) {
+ appendResponse(name, id, new ApiError(INVALID_REQUEST,
+ "The same topic was specified by name and by id."))
+ idToName.remove(id)
+ duplicatedTopicIds.add(id)
+ }
+ }
+ controller.findTopicIds(topicNamesToResolve).get().forEach { (name,
idOrError) =>
+ if (idOrError.isError)
+ unknownTopicNameErrors.put(name, idOrError.error)
+ else
+ maybeAppendToIdToName(idOrError.result, name)
+ }
+
+ /**
+ * There are 6 error cases to handle here if we don't have permission to
delete:
+ *
+ * 1. ID provided, topic missing => UNKNOWN_TOPIC_ID
+ * 2. ID provided, topic present, describeable =>
TOPIC_AUTHORIZATION_FAILED
+ * 3. ID provided, topic present, undescribeable => UNKNOWN_TOPIC_ID
+ * 4. name provided, topic missing, undescribable =>
UNKNOWN_TOPIC_OR_PARTITION
Review comment:
These cases seem wrong. It should be the following:
```
* 4. name provided, topic missing, undescribable =>
TOPIC_AUTHORIZATION_FAILED
* 5. name provided, topic missing, describable =>
UNKNOWN_TOPIC_OR_PARTITION
* 6. name provided, topic exists, undescribable =>
TOPIC_AUTHORIZATION_FAILED
```
If the client does not have describe permission, then it should get
`TOPIC_AUTHORIZATION_FAILED` regardless of existence.
##########
File path:
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
##########
@@ -809,6 +824,27 @@ private QuorumController(LogContext logContext,
() -> replicationControl.unregisterBroker(brokerId));
}
+ @Override
+ public CompletableFuture<Map<String, ResultOrError<Uuid>>>
findTopicIds(Collection<String> names) {
+ if (names.isEmpty()) return
CompletableFuture.completedFuture(Collections.emptyMap());
+ return appendReadEvent("findTopicIds",
+ () -> replicationControl.findTopicIds(lastCommittedOffset, names));
+ }
+
+ @Override
+ public CompletableFuture<Map<Uuid, ResultOrError<String>>>
findTopicNames(Collection<Uuid> ids) {
+ if (ids.isEmpty()) return
CompletableFuture.completedFuture(Collections.emptyMap());
+ return appendReadEvent("findTopicNames",
+ () -> replicationControl.findTopicNames(lastCommittedOffset, ids));
+ }
+
+ @Override
+ public CompletableFuture<Map<Uuid, ApiError>>
deleteTopics(Collection<Uuid> ids) {
+ if (ids.isEmpty()) return
CompletableFuture.completedFuture(Collections.emptyMap());
+ return appendWriteEvent("deleteTopics",
+ () -> replicationControl.deleteTopics(ids));
Review comment:
Is the deletion of topic configurations implicit based on the
DeleteTopic record? I know we discussed this, but I'm unsure what the final
outcome was. I don't see any logic for this in the broker listener, but the
implementation looks incomplete anyway.
----------------------------------------------------------------
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
