You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by jg...@apache.org on 2016/09/08 16:42:32 UTC
kafka git commit: KAFKA-4139;
Reset findCoordinatorFuture when brokers are unavailable
Repository: kafka
Updated Branches:
refs/heads/trunk 69ebf6f7b -> 6cf2cb6f2
KAFKA-4139; Reset findCoordinatorFuture when brokers are unavailable
Author: Rajini Sivaram <ra...@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 <ra...@googlemail.com>
Authored: Thu Sep 8 09:42:22 2016 -0700
Committer: Jason Gustafson <ja...@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();