diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java index 07c82907fef51..acf5a68f9e51c 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java @@ -372,7 +372,7 @@ Node selectReadReplica(final TopicPartition partition, final Node leaderReplica, } } - protected Map prepareCloseFetchSessionRequests() { + protected Map prepareCloseFetchSessionRequests(boolean checkNodeAvailability) { final Cluster cluster = metadata.fetch(); Map fetchable = new HashMap<>(); @@ -385,7 +385,9 @@ protected Map prepareCloseFetchSessi // skip sending the close request. final Node fetchTarget = cluster.nodeById(fetchTargetNodeId); - if (fetchTarget == null || isUnavailable(fetchTarget)) { + boolean fetchTargetAvailability = checkNodeAvailability ? (fetchTarget == null || isUnavailable(fetchTarget)) : fetchTarget == null; + + if (fetchTargetAvailability) { log.debug("Skip sending close session request to broker {} since it is not reachable", fetchTarget); return; } @@ -403,7 +405,7 @@ protected Map prepareCloseFetchSessi * Create fetch requests for all nodes for which we have assigned partitions * that have no existing requests in flight. */ - protected Map prepareFetchRequests() { + protected Map prepareFetchRequests(boolean checkNodeAvailability) { // Update metrics in case there was an assignment change metricsManager.maybeUpdateAssignment(subscriptions); @@ -428,37 +430,53 @@ protected Map prepareFetchRequests() // Use the preferred read replica if set, otherwise the partition's leader Node node = selectReadReplica(partition, leaderOpt.get(), currentTimeMs); - if (isUnavailable(node)) { - maybeThrowAuthFailure(node); - - // If we try to send during the reconnect backoff window, then the request is just - // going to be failed anyway before being sent, so skip sending the request for now - log.trace("Skipping fetch for partition {} because node {} is awaiting reconnect backoff", partition, node); - } else if (nodesWithPendingFetchRequests.contains(node.id())) { - log.trace("Skipping fetch for partition {} because previous request to {} has not been processed", partition, node); + if (checkNodeAvailability) { + if (isUnavailable(node)) { + maybeThrowAuthFailure(node); + + // If we try to send during the reconnect backoff window, then the request is just + // going to be failed anyway before being sent, so skip sending the request for now + log.trace("Skipping fetch for partition {} because node {} is awaiting reconnect backoff", partition, node); + } else if (nodesWithPendingFetchRequests.contains(node.id())) { + log.trace("Skipping fetch for partition {} because previous request to {} has not been processed", partition, node); + } else { + buildFetchRequests(fetchable, node, topicIds, partition, position); + } } else { - // if there is a leader and no in-flight requests, issue a new fetch - FetchSessionHandler.Builder builder = fetchable.computeIfAbsent(node, k -> { - FetchSessionHandler fetchSessionHandler = sessionHandlers.computeIfAbsent(node.id(), n -> new FetchSessionHandler(logContext, n)); - return fetchSessionHandler.newBuilder(); - }); - Uuid topicId = topicIds.getOrDefault(partition.topic(), Uuid.ZERO_UUID); - FetchRequest.PartitionData partitionData = new FetchRequest.PartitionData(topicId, - position.offset, - FetchRequest.INVALID_LOG_START_OFFSET, - fetchConfig.fetchSize, - position.currentLeader.epoch, - Optional.empty()); - builder.add(partition, partitionData); - - log.debug("Added {} fetch request for partition {} at position {} to node {}", fetchConfig.isolationLevel, - partition, position, node); + if (nodesWithPendingFetchRequests.contains(node.id())) { + log.trace("Skipping fetch for partition {} because previous request to {} has not been processed", partition, node); + } else { + buildFetchRequests(fetchable, node, topicIds, partition, position); + } } } - return fetchable.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().build())); } + private void buildFetchRequests(Map fetchable, + Node node, + Map topicIds, + TopicPartition partition, + SubscriptionState.FetchPosition position) { + + // if there is a leader and no in-flight requests, issue a new fetch + FetchSessionHandler.Builder builder = fetchable.computeIfAbsent(node, k -> { + FetchSessionHandler fetchSessionHandler = sessionHandlers.computeIfAbsent(node.id(), n -> new FetchSessionHandler(logContext, n)); + return fetchSessionHandler.newBuilder(); + }); + Uuid topicId = topicIds.getOrDefault(partition.topic(), Uuid.ZERO_UUID); + FetchRequest.PartitionData partitionData = new FetchRequest.PartitionData(topicId, + position.offset, + FetchRequest.INVALID_LOG_START_OFFSET, + fetchConfig.fetchSize, + position.currentLeader.epoch, + Optional.empty()); + builder.add(partition, partitionData); + + log.debug("Added {} fetch request for partition {} at position {} to node {}", fetchConfig.isolationLevel, + partition, position, node); + } + // Visible for testing protected FetchSessionHandler sessionHandler(int node) { return sessionHandlers.get(node); diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchRequestManager.java index d99fa77f0f9e7..6343a4a8b497f 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchRequestManager.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchRequestManager.java @@ -70,8 +70,10 @@ protected void maybeThrowAuthFailure(Node node) { */ @Override public PollResult poll(long currentTimeMs) { + boolean checkNodeAvailability = false; + return pollInternal( - prepareFetchRequests(), + prepareFetchRequests(checkNodeAvailability), this::handleFetchSuccess, this::handleFetchFailure ); @@ -82,8 +84,10 @@ public PollResult poll(long currentTimeMs) { */ @Override public PollResult pollOnClose() { + boolean checkNodeAvailability = false; + return pollInternal( - prepareCloseFetchSessionRequests(), + prepareCloseFetchSessionRequests(checkNodeAvailability), this::handleCloseFetchSessionSuccess, this::handleCloseFetchSessionFailure ); diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index 51b2532e48b68..0ef80d0e4d037 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -102,7 +102,9 @@ public void clearBufferedDataForUnassignedPartitions(Collection * @return number of fetches sent */ public synchronized int sendFetches() { - final Map fetchRequests = prepareFetchRequests(); + boolean checkNodeAvailability = true; + + final Map fetchRequests = prepareFetchRequests(checkNodeAvailability); sendFetchesInternal( fetchRequests, (fetchTarget, data, clientResponse) -> { @@ -119,8 +121,10 @@ public synchronized int sendFetches() { } protected void maybeCloseFetchSessions(final Timer timer) { + boolean checkNodeAvailability = true; + final List> requestFutures = sendFetchesInternal( - prepareCloseFetchSessionRequests(), + prepareCloseFetchSessionRequests(checkNodeAvailability), this::handleCloseFetchSessionSuccess, this::handleCloseFetchSessionFailure ); diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java index e080f63c1a81e..b4eeaa0d43606 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java @@ -780,10 +780,10 @@ public void testFetchSkipsBlackedOutNodes() { Node node = initialUpdateResponse.brokers().iterator().next(); client.backoff(node, 500); - assertEquals(0, sendFetches()); + assertEquals(1, sendFetches()); time.sleep(500); - assertEquals(1, sendFetches()); + assertEquals(0, sendFetches()); } @Test