Default FetchRequestManager.isUnavailable to return false and maybeThrowAuthFailure to perform no ops

Author: phuchong.tran

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

@@ -55,15 +55,22 @@ public class FetchRequestManager extends AbstractFetch implements RequestManager
         this.networkClientDelegate = networkClientDelegate;
     }
 
+    /**
+     * 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) {
-        return networkClientDelegate.isUnavailable(node);
+        return false;
     }
 
+    /**
+     * Authentication failure will be checked at a later stage, so we do nothing here
+     * @param node {@link Node} to check for a previous {@link AuthenticationException}; if found it is thrown
+     */
     @Override
-    protected void maybeThrowAuthFailure(Node node) {
-        networkClientDelegate.maybeThrowAuthFailure(node);
-    }
+    protected void maybeThrowAuthFailure(Node node) {}
 
     /**
      * {@inheritDoc}

Diff for: 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());
 
         time.sleep(500);
-        assertEquals(1, sendFetches());
+        assertEquals(0, sendFetches());
     }
 
     @Test