kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From j...@apache.org
Subject kafka git commit: KAFKA-4139; Reset findCoordinatorFuture when brokers are unavailable
Date Thu, 08 Sep 2016 16:42:32 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 69ebf6f7b -> 6cf2cb6f2


KAFKA-4139; Reset findCoordinatorFuture when brokers are unavailable

Author: Rajini Sivaram <rajinisivaram@googlemail.com>

Reviewers: Ismael Juma, Jason Gustafson

Closes #1831 from rajinisivaram/KAFKA-4139


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/6cf2cb6f
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/6cf2cb6f
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/6cf2cb6f

Branch: refs/heads/trunk
Commit: 6cf2cb6f294d0d4766f57e9da660fb4efac7d258
Parents: 69ebf6f
Author: Rajini Sivaram <rajinisivaram@googlemail.com>
Authored: Thu Sep 8 09:42:22 2016 -0700
Committer: Jason Gustafson <jason@confluent.io>
Committed: Thu Sep 8 09:42:22 2016 -0700

----------------------------------------------------------------------
 .../consumer/internals/AbstractCoordinator.java | 31 ++++++++++----------
 .../internals/AbstractCoordinatorTest.java      | 17 +++++++++++
 2 files changed, 32 insertions(+), 16 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/6cf2cb6f/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
index f2e15ca..0766f3d 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
@@ -207,8 +207,16 @@ public abstract class AbstractCoordinator implements Closeable {
     }
 
     protected synchronized RequestFuture<Void> lookupCoordinator() {
-        if (findCoordinatorFuture == null)
-            findCoordinatorFuture = sendGroupCoordinatorRequest();
+        if (findCoordinatorFuture == null) {
+            // find a node to ask about the coordinator
+            Node node = this.client.leastLoadedNode();
+            if (node == null) {
+                // TODO: If there are no brokers left, perhaps we should use the bootstrap
set
+                // from configuration?
+                return RequestFuture.noBrokersAvailable();
+            } else
+                findCoordinatorFuture = sendGroupCoordinatorRequest(node);
+        }
         return findCoordinatorFuture;
     }
 
@@ -496,21 +504,12 @@ public abstract class AbstractCoordinator implements Closeable {
      * one of the brokers. The returned future should be polled to get the result of the
request.
      * @return A request future which indicates the completion of the metadata request
      */
-    private RequestFuture<Void> sendGroupCoordinatorRequest() {
+    private RequestFuture<Void> sendGroupCoordinatorRequest(Node node) {
         // initiate the group metadata request
-        // find a node to ask about the coordinator
-        Node node = this.client.leastLoadedNode();
-        if (node == null) {
-            // TODO: If there are no brokers left, perhaps we should use the bootstrap set
-            // from configuration?
-            return RequestFuture.noBrokersAvailable();
-        } else {
-            // create a group  metadata request
-            log.debug("Sending coordinator request for group {} to broker {}", groupId, node);
-            GroupCoordinatorRequest metadataRequest = new GroupCoordinatorRequest(this.groupId);
-            return client.send(node, ApiKeys.GROUP_COORDINATOR, metadataRequest)
-                    .compose(new GroupCoordinatorResponseHandler());
-        }
+        log.debug("Sending coordinator request for group {} to broker {}", groupId, node);
+        GroupCoordinatorRequest metadataRequest = new GroupCoordinatorRequest(this.groupId);
+        return client.send(node, ApiKeys.GROUP_COORDINATOR, metadataRequest)
+                     .compose(new GroupCoordinatorResponseHandler());
     }
 
     private class GroupCoordinatorResponseHandler extends RequestFutureAdapter<ClientResponse,
Void> {

http://git-wip-us.apache.org/repos/asf/kafka/blob/6cf2cb6f/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java
index 77f9df5..777b67f 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java
@@ -43,6 +43,7 @@ import java.util.List;
 import java.util.Map;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
@@ -133,6 +134,22 @@ public class AbstractCoordinatorTest {
         }
     }
 
+    @Test
+    public void testLookupCoordinator() throws Exception {
+        mockClient.setNode(null);
+        RequestFuture<Void> noBrokersAvailableFuture = coordinator.lookupCoordinator();
+        assertTrue("Failed future expected", noBrokersAvailableFuture.failed());
+
+        mockClient.setNode(node);
+        RequestFuture<Void> future = coordinator.lookupCoordinator();
+        assertFalse("Request not sent", future.isDone());
+        assertTrue("New request sent while one is in progress", future == coordinator.lookupCoordinator());
+
+        mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));
+        coordinator.ensureCoordinatorReady();
+        assertTrue("New request not sent after previous completed", future != coordinator.lookupCoordinator());
+    }
+
     private Struct groupCoordinatorResponse(Node node, Errors error) {
         GroupCoordinatorResponse response = new GroupCoordinatorResponse(error.code(), node);
         return response.toStruct();


Mime
View raw message