Skip to content

KAFKA-15556: Remove NetworkClientDelegate methods isUnavailable, maybeThrowAuthFailure, and tryConnect #14979

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Phuc-Hong-Tran wants to merge 1 commit into from
Closed

KAFKA-15556: Remove NetworkClientDelegate methods isUnavailable, maybeThrowAuthFailure, and tryConnect #14979

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
74 changes: 46 additions & 28 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -372,7 +372,7 @@ Node selectReadReplica(final TopicPartition partition, final Node leaderReplica,
}
}

protected Map<Node, FetchSessionHandler.FetchRequestData> prepareCloseFetchSessionRequests() {
protected Map<Node, FetchSessionHandler.FetchRequestData> prepareCloseFetchSessionRequests(boolean checkNodeAvailability) {
final Cluster cluster = metadata.fetch();
Map<Node, FetchSessionHandler.Builder> fetchable = new HashMap<>();

Expand All @@ -385,7 +385,9 @@ protected Map<Node, FetchSessionHandler.FetchRequestData> 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;
}
Expand All @@ -403,7 +405,7 @@ protected Map<Node, FetchSessionHandler.FetchRequestData> prepareCloseFetchSessi
* Create fetch requests for all nodes for which we have assigned partitions
* that have no existing requests in flight.
*/
protected Map<Node, FetchSessionHandler.FetchRequestData> prepareFetchRequests() {
protected Map<Node, FetchSessionHandler.FetchRequestData> prepareFetchRequests(boolean checkNodeAvailability) {
// Update metrics in case there was an assignment change
metricsManager.maybeUpdateAssignment(subscriptions);

Expand All @@ -428,37 +430,53 @@ protected Map<Node, FetchSessionHandler.FetchRequestData> 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<Node, FetchSessionHandler.Builder> fetchable,
Node node,
Map<String, Uuid> 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);
Expand Down
8 changes: 6 additions & 2 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchRequestManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
);
Expand All @@ -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
);
Expand Down
8 changes: 6 additions & 2 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,9 @@ public void clearBufferedDataForUnassignedPartitions(Collection<TopicPartition>
* @return number of fetches sent
*/
public synchronized int sendFetches() {
final Map<Node, FetchSessionHandler.FetchRequestData> fetchRequests = prepareFetchRequests();
boolean checkNodeAvailability = true;

final Map<Node, FetchSessionHandler.FetchRequestData> fetchRequests = prepareFetchRequests(checkNodeAvailability);
sendFetchesInternal(
fetchRequests,
(fetchTarget, data, clientResponse) -> {
Expand All @@ -119,8 +121,10 @@ public synchronized int sendFetches() {
}

protected void maybeCloseFetchSessions(final Timer timer) {
boolean checkNodeAvailability = true;

final List<RequestFuture<ClientResponse>> requestFutures = sendFetchesInternal(
prepareCloseFetchSessionRequests(),
prepareCloseFetchSessionRequests(checkNodeAvailability),
this::handleCloseFetchSessionSuccess,
this::handleCloseFetchSessionFailure
);
Expand Down
4 changes: 2 additions & 2 deletions ...ts/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down