KAFKA-15615: Improve handling of fetching during metadata updates

- Move await a metadata update logic from AbstractFetch to FetchCollector

Author: appchemist

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java

@@ -200,9 +200,6 @@ protected void handleFetchSuccess(final Node fetchTarget,
                     if (partitionData.currentLeader().leaderId() != -1 && partitionData.currentLeader().leaderEpoch() != -1) {
                         partitionsWithUpdatedLeaderInfo.put(partition, new Metadata.LeaderIdAndEpoch(
                             Optional.of(partitionData.currentLeader().leaderId()), Optional.of(partitionData.currentLeader().leaderEpoch())));
-                    } else {
-                        requestMetadataUpdate(metadata, subscriptions, partition);
-                        subscriptions.awaitUpdate(partition);
                     }
                 }
 

clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchCollector.java

@@ -333,6 +333,10 @@ private void handleInitializeErrors(final CompletedFetch completedFetch, final E
                 error == Errors.FENCED_LEADER_EPOCH) {
             log.debug("Error in fetch for partition {}: {}", tp, error.exceptionName());
             requestMetadataUpdate(metadata, subscriptions, tp);
+            if (completedFetch.partitionData.currentLeader().leaderId() == -1 ||
+                    completedFetch.partitionData.currentLeader().leaderEpoch() == -1) {
+                subscriptions.awaitUpdate(tp);
+            }
         } else if (error == Errors.KAFKA_STORAGE_ERROR ||
                 error == Errors.OFFSET_NOT_AVAILABLE) {
             log.debug("Error in fetch for partition {}: {}", tp, error.exceptionName());