AndrewJSchofield commented on code in PR #14564: URL: https://github.com/apache/kafka/pull/14564#discussion_r1371441595
########## clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java: ########## @@ -685,6 +711,13 @@ private void completeBatch(ProducerBatch batch, ProduceResponse.PartitionRespons "to request metadata update now", batch.topicPartition, error.exception(response.errorMessage).toString()); } + if (error.exception() instanceof NotLeaderOrFollowerException || error.exception() instanceof FencedLeaderEpochException) { + log.debug("For {}, received error {}, with leaderIdAndEpoch {}", batch.topicPartition, error, response.currentLeader); + if (partitionsWithUpdatedLeaderInfo != null && response.currentLeader.leaderId() != -1 && response.currentLeader.leaderEpoch() != -1) { Review Comment: This line could be more legible with some parentheses and splitting the condition across multiple lines. ########## clients/src/main/java/org/apache/kafka/clients/MetadataCache.java: ########## @@ -150,7 +150,7 @@ MetadataCache mergeWith(String newClusterId, // We want the most recent topic ID. We start with the previous ID stored for retained topics and then // update with newest information from the MetadataResponse. We always take the latest state, removing existing // topic IDs if the latest state contains the topic name but not a topic ID. - Map<String, Uuid> newTopicIds = topicIds.entrySet().stream() + Map<String, Uuid> newTopicIds = this.topicIds.entrySet().stream() Review Comment: I would rename the argument currently called `topicIds` too, just to try to prevent a similar bug being re-introduced by accident. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org