checking newNode size to make sure its not 0 (#520)

* checking newNode size to make sure its not 0
linkedin · Sep 3, 2024 · cfa6f94 · cfa6f94
1 parent 05c9ff8
commit cfa6f94
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 0 deletions.
diff --git a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
@@ -990,6 +990,10 @@ public Node leastLoadedNode(long now) {
                 newNodes.add(new Node(nodeId--, address.getHostString(), address.getPort()));
             }
 
+            if (newNodes.size() == 0) {
+                return null;
+            }
+
             int offset = this.randOffset.nextInt(newNodes.size());
             Node node = newNodes.get(offset);
             log.trace("Resolved bootstrap server again, randomly picked node {} as least loaded node from the resolved node set", node);

diff --git a/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java b/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java
@@ -18,6 +18,7 @@
 
 import java.util.PriorityQueue;
 import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.errors.AuthenticationException;
@@ -631,6 +632,19 @@ public void testResolveBootstrapInLeastLoadedNode() {
         assertNotEquals(node, clusterClient.leastLoadedNode(time.milliseconds()));
     }
 
+    @Test
+    public void noLeastLoadedNode() {
+        NetworkClient nc = new NetworkClient(selector, clusterMetadataUpdater, "mock-cluster-md", Integer.MAX_VALUE,
+            0, 0, 64 * 1024, 64 * 1024,
+            defaultRequestTimeoutMs, connectionSetupTimeoutMsTest, connectionSetupTimeoutMaxMsTest, time, true, new ApiVersions(), new LogContext(),
+            LeastLoadedNodeAlgorithm.VANILLA, new ArrayList<>());
+        nc.ready(node, time.milliseconds());
+        assertFalse(client.isReady(node, time.milliseconds()));
+        assertThrows(ConfigException.class, () -> nc.leastLoadedNode(time.milliseconds()));
+
+        assertEquals(null, nc.leastLoadedNode(time.milliseconds()));
+    }
+
     @Test
     public void testLeastLoadedNode() {
         client.ready(node, time.milliseconds());