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

Phuc-Hong-Tran · 2023-12-15T03:10:31Z

Change:

*Refactored AbstractFetch so only Fetcher will check for node status when creating fetch requests.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

Phuc-Hong-Tran · 2023-12-15T03:12:16Z

@philipnee, I closed the other PR since closing and opening it again doesn't trigger the CI to start. Here is the context for this PR: #14406 (comment)

Phuc-Hong-Tran · 2023-12-15T18:16:06Z

@philipnee, all builds succeed, test failures looks unrelated. PTAL, thanks.

Phuc-Hong-Tran · 2023-12-28T04:18:55Z

@philipnee @kirktrue, Please take a look if you guys have some free time. Thanks in advance.

philipnee

Hi @Phuc-Hong-Tran - thanks for taking time raising this PR. I apologize for not writing the the jira ticket well here, so I recommend to rethink how you would approach this refactor. In short - we don't need to handle isUnavailable and auth failure in the request manager, so it is ok to make the default implementation to just return false, and perform no ops for both methods. The reason is networkClientDelegate already handle these scenario, just in a slight different order to the LegacyKafkaConsumer.

philipnee · 2024-01-02T17:10:15Z

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<Node, FetchSessionHandler.FetchRequestData> prepareCloseFetchSessionRequests() {
+    protected Map<Node, FetchSessionHandler.FetchRequestData> prepareCloseFetchSessionRequests(boolean checkNodeAvailability) {


final boolean checkNodeAvailability

philipnee · 2024-01-02T17:11:38Z

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

@@ -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;


final boolean

i would also call this isFetchTargetAvailable

let's not use inline if. this makes the code harder to read.

philipnee · 2024-01-02T17:13:05Z

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

-                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) {


is it possible to avoid nested if?

philipnee · 2024-01-02T17:14:20Z

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

        return pollInternal(
-                prepareFetchRequests(),
+                prepareFetchRequests(checkNodeAvailability),


just pass false. the var above is useless.

philipnee · 2024-01-02T17:14:35Z

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

        return pollInternal(
-                prepareCloseFetchSessionRequests(),
+                prepareCloseFetchSessionRequests(checkNodeAvailability),


philipnee · 2024-01-02T17:15:25Z

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

@@ -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;


same as above

philipnee · 2024-01-02T17:17:16Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java

@@ -781,10 +781,10 @@ public void testFetchSkipsBlackedOutNodes() {
        Node node = initialUpdateResponse.brokers().iterator().next();

        client.backoff(node, 500);
-        assertEquals(0, sendFetches());
+        assertEquals(1, sendFetches());


why are you changing the tests?

Since we're not checking the node availablity using the networkClientDelegate in FetchRequestManager, the request that was made would pass through the check and made it way to unsentRequest. After the backoff the request should be sent already.

Should I not change the test and make new one instead?

philipnee · 2024-01-02T17:17:58Z

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

@@ -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) {


final boolean

Phuc-Hong-Tran · 2024-01-03T14:09:25Z

@philipnee, Thanks for the comments. When you said I should rethink about the approach, are you suggesting that I should start over from the idea phase?

philipnee · 2024-01-03T19:34:43Z

hi @Phuc-Hong-Tran - yes, the abstractFetch implementation is based on the LegacyKafkaConsumer and therefore requires connection probing. We don't need that in the AsyncKafkaConsumer as it is being done right before sending out the requests.

Phuc-Hong-Tran · 2024-01-04T00:26:07Z

I understand, will get another PR out ASAP.

…rowAuthFailure to perform no ops

Phuc-Hong-Tran · 2024-01-04T14:46:19Z

@philipnee, PTAL, thanks in advance.

Phuc-Hong-Tran · 2024-01-05T20:41:35Z

@philipnee, is there anything else that I should change?

philipnee · 2024-02-08T18:10:32Z

hi @Phuc-Hong-Tran - thanks for the PR. Could you also clean up the isUnavailable and maybeThrowAuthFailure methods in the networkClientDelegate. I believe they aren't being used anywhere.

Phuc-Hong-Tran · 2024-02-09T13:17:46Z

@philipnee Will do

…entDelegate

Phuc-Hong-Tran · 2024-02-09T13:18:43Z

@philipnee All done

kirktrue · 2024-03-26T18:18:04Z

@Phuc-Hong-Tran—is this PR ready for review?

Phuc-Hong-Tran · 2024-03-26T21:33:41Z

@kirktrue, it's ready for review

github-actions · 2024-06-25T03:33:45Z

This PR is being marked as stale since it has not had any activity in 90 days. If you would like to keep this PR alive, please ask a committer for review. If the PR has merge conflicts, please update it with the latest from trunk (or appropriate release branch)

If this PR is no longer valid or desired, please feel free to close it. If no activity occurs in the next 30 days, it will be automatically closed.

AndrewJSchofield · 2024-07-23T08:22:55Z

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

+     * Node's availability will be checked at a later stage, so we default to return false.
+     * @param node {@link Node} to check for availability
+     * @return
+     */
    @Override
    protected boolean isUnavailable(Node node) {


This does seem a bit like half a refactor. Would it be possible to remove isUnavailable and maybeThrowAuthFailure entirely from AbstractFetch?

github-actions · 2025-04-05T03:40:19Z

This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.

If you are having difficulty finding a reviewer, please reach out on the [mailing list](https://kafka.apache.org/contact).

If this PR is no longer valid or desired, please feel free to close it. If no activity occurs in the next 30 days, it will be automatically closed.

philipnee suggested changes Jan 2, 2024

View reviewed changes

Default FetchRequestManager.isUnavailable to return false and maybeTh…

39fd2a8

…rowAuthFailure to perform no ops

Phuc-Hong-Tran force-pushed the trunk branch from d996563 to 39fd2a8 Compare January 4, 2024 14:27

Re-commit to trigger CI

362557f

Phuc-Hong-Tran force-pushed the trunk branch from 1b51887 to 362557f Compare January 8, 2024 16:19

Phuc-Hong-Tran requested a review from philipnee February 2, 2024 08:01

Removed isUnavailable and maybeThrowAuthFailure methods in NetworkCli…

5bd791d

…entDelegate

github-actions bot added the stale Stale PRs label Jun 25, 2024

mjsax added the consumer label Jul 22, 2024

philipnee added the KIP-848 The Next Generation of the Consumer Rebalance Protocol label Jul 23, 2024

AndrewJSchofield requested changes Jul 23, 2024

View reviewed changes

kirktrue added the clients label Sep 26, 2024

github-actions bot removed the stale Stale PRs label Jan 5, 2025

github-actions bot added the stale Stale PRs label Apr 5, 2025

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

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

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

Phuc-Hong-Tran commented Dec 15, 2023

Phuc-Hong-Tran commented Dec 15, 2023 •

edited

Loading

Phuc-Hong-Tran commented Dec 15, 2023

Phuc-Hong-Tran commented Dec 28, 2023

philipnee left a comment

philipnee Jan 2, 2024

philipnee Jan 2, 2024

philipnee Jan 2, 2024

philipnee Jan 2, 2024

philipnee Jan 2, 2024

philipnee Jan 2, 2024

philipnee Jan 2, 2024

Phuc-Hong-Tran Jan 3, 2024

philipnee Jan 2, 2024

Phuc-Hong-Tran commented Jan 3, 2024 •

edited

Loading

philipnee commented Jan 3, 2024

Phuc-Hong-Tran commented Jan 4, 2024

Phuc-Hong-Tran commented Jan 4, 2024

Phuc-Hong-Tran commented Jan 5, 2024

philipnee commented Feb 8, 2024

Phuc-Hong-Tran commented Feb 9, 2024

Phuc-Hong-Tran commented Feb 9, 2024

kirktrue commented Mar 26, 2024

Phuc-Hong-Tran commented Mar 26, 2024

github-actions bot commented Jun 25, 2024

AndrewJSchofield Jul 23, 2024

github-actions bot commented Apr 5, 2025

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

Are you sure you want to change the base?

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

Conversation

Phuc-Hong-Tran commented Dec 15, 2023

Committer Checklist (excluded from commit message)

Phuc-Hong-Tran commented Dec 15, 2023 • edited Loading

Phuc-Hong-Tran commented Dec 15, 2023

Phuc-Hong-Tran commented Dec 28, 2023

philipnee left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Phuc-Hong-Tran commented Jan 3, 2024 • edited Loading

philipnee commented Jan 3, 2024

Phuc-Hong-Tran commented Jan 4, 2024

Phuc-Hong-Tran commented Jan 4, 2024

Phuc-Hong-Tran commented Jan 5, 2024

philipnee commented Feb 8, 2024

Phuc-Hong-Tran commented Feb 9, 2024

Phuc-Hong-Tran commented Feb 9, 2024

kirktrue commented Mar 26, 2024

Phuc-Hong-Tran commented Mar 26, 2024

github-actions bot commented Jun 25, 2024

Choose a reason for hiding this comment

github-actions bot commented Apr 5, 2025

Phuc-Hong-Tran commented Dec 15, 2023 •

edited

Loading

Phuc-Hong-Tran commented Jan 3, 2024 •

edited

Loading