chia7712 commented on a change in pull request #10223:
URL: https://github.com/apache/kafka/pull/10223#discussion_r585628385
##########
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:
> 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?
That makes sense to me. We have to make sure this rule is applied to #10184
:)
----------------------------------------------------------------
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]
