lianetm commented on code in PR #22279:
URL: https://github.com/apache/kafka/pull/22279#discussion_r4348831485
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerMetadata.java:
##########
@@ -107,6 +107,18 @@ synchronized void clearTransientTopics() {
this.transientTopics.clear();
}
+ /**
+ * Topics this consumer currently needs metadata for: subscribed topics,
topics of assigned
+ * partitions, and transient topics registered temporarily by
offset-related APIs. This is the
+ * same set of topics {@link #retainTopic(String, boolean, long)} keeps,
so a metadata error
+ * about a topic outside this set cannot concern any of this consumer's
pending operations.
+ */
+ public synchronized Set<String> subscribedAssignedAndTransientTopics() {
Review Comment:
uhm this piece of logic seems to overlap with bits that already exist in
metadata and subscription state, wonder if we should review them all to avoid
ending up with different ways to identify the same (if a topic's metadata is
relevant)
E.g, `metadataTopics` (doesn't it already tell us the same as the new
"subscribedOrAssignedTopics"?), also `retainTopic` (this is what we already use
to know if the client keeps the metadata entry from response or drops it, also
`needsMetadata`
Ideally I would aim for a hook here on the metadata component, that just
puts together all the hooks we already have to determine if a topic is still
"relevant" (so we can be sure it's ok to throw the exception". Maybe a
combination of isTransient || needsMetadata || [regex, this seems a missing bit
to consider in the PR?]...Still not 100% sure this is all but for sure it's
part of it
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java:
##########
@@ -452,12 +454,18 @@ private boolean maybeFailOnMetadataError(List<?> events) {
return false;
Optional<Exception> metadataError =
networkClientDelegate.getAndClearMetadataError();
+ if (metadataError.isEmpty())
+ return false;
- if (metadataError.isPresent()) {
- filteredEvents.forEach(e ->
e.onMetadataError(metadataError.get()));
- return true;
- } else {
+ Set<String> relevantTopics =
applicationEventProcessor.subscribedAssignedAndTransientTopics();
+ if (isErrorProvablyUnrelated(metadataError.get(), relevantTopics)) {
+ log.debug("Dropping metadata error deemed unrelated to currently
relevant topics {}",
+ relevantTopics, metadataError.get());
Review Comment:
I'm still nervous about this :)
E.g., I don't see we consider regex. Client-side old regex, metadata for all
topics is resolved on the client to compute it, would we drop auth errors here?
Similar thought for the new R2j regex
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
