hachikuji commented on a change in pull request #10223:
URL: https://github.com/apache/kafka/pull/10223#discussion_r585602805
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -1884,20 +1884,26 @@ class KafkaApis(val requestChannel: RequestChannel,
val authorizedDeleteTopics =
authHelper.filterByAuthorized(request.context, DELETE, TOPIC,
results.asScala.filter(result => result.name() != null))(_.name)
results.forEach { topic =>
- val unresolvedTopicId = !(topic.topicId() == Uuid.ZERO_UUID) &&
topic.name() == null
- if (!config.usesTopicId &&
topicIdsFromRequest.contains(topic.topicId)) {
- topic.setErrorCode(Errors.UNSUPPORTED_VERSION.code)
- topic.setErrorMessage("Topic IDs are not supported on the server.")
- } else if (unresolvedTopicId)
- topic.setErrorCode(Errors.UNKNOWN_TOPIC_ID.code)
- else if (topicIdsFromRequest.contains(topic.topicId) &&
!authorizedDescribeTopics(topic.name))
- topic.setErrorCode(Errors.UNKNOWN_TOPIC_ID.code)
- else if (!authorizedDeleteTopics.contains(topic.name))
- topic.setErrorCode(Errors.TOPIC_AUTHORIZATION_FAILED.code)
- else if (!metadataCache.contains(topic.name))
- topic.setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code)
- else
- toDelete += topic.name
+ val unresolvedTopicId = topic.topicId() != Uuid.ZERO_UUID &&
topic.name() == null
+ if (!config.usesTopicId &&
topicIdsFromRequest.contains(topic.topicId)) {
+ topic.setErrorCode(Errors.UNSUPPORTED_VERSION.code)
+ topic.setErrorMessage("Topic IDs are not supported on the server.")
+ } else if (unresolvedTopicId) {
+ topic.setErrorCode(Errors.UNKNOWN_TOPIC_ID.code)
+ } else if (topicIdsFromRequest.contains(topic.topicId) &&
!authorizedDescribeTopics(topic.name)) {
+ // Because the client does not have Describe permission, the name
should
+ // not be returned in the response. Note, however, that we do not
consider
+ // the topicId itself to be sensitive, so there is no reason to
obscure
+ // this case with `UNKNOWN_TOPIC_ID`.
+ topic.setName(null)
+ topic.setErrorCode(Errors.TOPIC_AUTHORIZATION_FAILED.code)
+ } else if (!authorizedDeleteTopics.contains(topic.name)) {
Review comment:
I guess there are really two sub-cases. Here's how they are currently
handled:
```
name provided, topic missing, describable, deletable =>
UNKNOWN_TOPIC_OR_PARTITION
name provided, topic missing, describable, undeletable =>
TOPIC_AUTHORIZATION_FAILED
```
This seems defensible to me. The `UNKNOWN_TOPIC_OR_PARTITION` error will
cause the client to retry because of the possibility of stale metadata, but it
can't delete the topic anyway because of the authorization failure. It seems
better to fail fast?
----------------------------------------------------------------
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]
