If we pass false to getTopicMetadata, it generates UNKNOWN_TOPIC_OR_PARTITION when the topic is removed, right? If so, does client-side need to handle such error? For example, KafkaAdminClient#listTopics should filter out those "nonexistent" topics (https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java#L1717). Otherwise, users want to get all "existent" topics but response say a_topic is "nonexistent".

Hi @chia7712 , thank you for bringing up this issue and it's totally valid!
In order to preserve its behavior (not including UNKNOWN_TOPIC_OR_PARTITION in the response to fetch all topic metadata request), we implemented the below logic which filters out all entries in the response with UNKNOWN_TOPIC_OR_PARTITION if the metadata request is to get metadata for all topisc.

val completeTopicMetadata = (if (metadataRequest.isAllTopics) {
    opicMetadata.filter(_.errorCode() != Errors.UNKNOWN_TOPIC_OR_PARTITION.code())
} else {
    topicMetadata
}) ++ unauthorizedForCreateTopicMetadata ++ unauthorizedForDescribeTopicMetadata

We have added a unit test as well. Hope to get some feedback from you soon!

Could we do this filter early? For example, how about filtering getTopicMetadata directly? That makes all fixes be in one place and following authorized ops can skip "unknown partition/topic" as well.

This is a good point. We should pass a boolean to getTopicMetadata indicating that it's an "allTopics" request and have that method handle everything.

Hi @chia7712 @ijuma ,

This is Joseph and I'm @Lincong 's colleague working on this patch with him. I think this is a good point and just updated the PR accordingly

@ijuma I will merge this PR tomorrow if there is no objection.

Could we remove this @Before code? Changing modifier of metadataCache from "val" to "var" is enough.

hey @chia7712 ,

I add this to let each test case to have a clean, uncorrupted MetadataCache. Is there any historical context that we should avoid that cleanup?

It seems to me following code is able to generate a clean, uncorrupted MetadataCache for each test case.

var metadataCache = new MetadataCache(brokerId)

Also, it is simpler than @Before block.

Hey @chia7712 you're right. I did some reading and found that I misunderstood JUnit's behavior. Just updated that part and rebased latest trunk.

Could we not avoid this altogether instead of doing the filter later?

@ijuma I slightly changed it (and rebased latest) so the processing is a branch in the middle about isFetchAllMetadata.

It's still a map and filter structure because I don't want to sacrifice the readability, but it's handled in the middle instead of at the returning statement, which I think is more elegant and readable than the original version.

Let me know what you think!

Also, I added the assertion of no UNKNOWN_TOPIC_OR_PARTITION into the unit test

I think if you use flatMap, you don't need the filter and it's still readable. Thoughts?

@ijuma That's a good point. I've updated it.

Is Some/None more readable than List/Nil?

@chia7712 Thanks, you're right. Just updated accordingly and rebased latest trunk

We don't usually include JIRA references in the code unless it's a very complex issue. I think the comment here could be something like:

"A metadata request for all topics should never result in topic auto creation. A topic may be deleted between the creation of topics and topicResponses, so we make sure to always return None for this case."

Thanks @ijuma ,

To follow the convention in Kafka, I just updated the comments. Let me know if you have there are any other thoughts!