You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by gw...@apache.org on 2015/11/10 19:27:00 UTC
kafka git commit: KAFKA-2792: Don't wait for a response to the leave
group message when closing the new consumer.
Repository: kafka
Updated Branches:
refs/heads/trunk fe11488a7 -> ae5a5d7c0
KAFKA-2792: Don't wait for a response to the leave group message when closing the new consumer.
Author: Ewen Cheslack-Postava <me...@ewencp.org>
Reviewers: Onur Karaman, Gwen Shapira
Closes #480 from ewencp/kafka-2792-fix-blocking-consumer-close
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/ae5a5d7c
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/ae5a5d7c
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/ae5a5d7c
Branch: refs/heads/trunk
Commit: ae5a5d7c08bb634576a414f6f2864c5b8a7e58a3
Parents: fe11488
Author: Ewen Cheslack-Postava <me...@ewencp.org>
Authored: Tue Nov 10 10:26:51 2015 -0800
Committer: Gwen Shapira <cs...@gmail.com>
Committed: Tue Nov 10 10:26:51 2015 -0800
----------------------------------------------------------------------
.../apache/kafka/clients/consumer/KafkaConsumer.java | 2 +-
.../consumer/internals/AbstractCoordinator.java | 13 +++++--------
.../consumer/internals/ConsumerCoordinatorTest.java | 4 ++--
3 files changed, 8 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/ae5a5d7c/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
index d3616f9..89b2f0b 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
@@ -760,7 +760,7 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
try {
log.debug("Unsubscribed all topics or patterns and assigned partitions");
this.subscriptions.unsubscribe();
- this.coordinator.maybeLeaveGroup(false);
+ this.coordinator.maybeLeaveGroup();
this.metadata.needMetadataForAllTopics(false);
} finally {
release();
http://git-wip-us.apache.org/repos/asf/kafka/blob/ae5a5d7c/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 5b5c8a5..a12c6c1 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
@@ -540,18 +540,18 @@ public abstract class AbstractCoordinator implements Closeable {
@Override
public void close() {
client.disableWakeups();
- maybeLeaveGroup(true);
+ maybeLeaveGroup();
}
/**
* Leave the current group and reset local generation/memberId.
*/
- public void maybeLeaveGroup(boolean awaitResponse) {
+ public void maybeLeaveGroup() {
client.unschedule(heartbeatTask);
if (!coordinatorUnknown() && generation > 0) {
// this is a minimal effort attempt to leave the group. we do not
// attempt any resending if the request fails or times out.
- sendLeaveGroupRequest(awaitResponse);
+ sendLeaveGroupRequest();
}
this.generation = OffsetCommitRequest.DEFAULT_GENERATION_ID;
@@ -559,7 +559,7 @@ public abstract class AbstractCoordinator implements Closeable {
rejoinNeeded = true;
}
- private void sendLeaveGroupRequest(boolean awaitResponse) {
+ private void sendLeaveGroupRequest() {
LeaveGroupRequest request = new LeaveGroupRequest(groupId, memberId);
RequestFuture<Void> future = client.send(coordinator, ApiKeys.LEAVE_GROUP, request)
.compose(new LeaveGroupResponseHandler());
@@ -574,10 +574,7 @@ public abstract class AbstractCoordinator implements Closeable {
}
});
- if (awaitResponse)
- client.poll(future);
- else
- client.poll(future, 0);
+ client.poll(future, 0);
}
private class LeaveGroupResponseHandler extends CoordinatorResponseHandler<LeaveGroupResponse, Void> {
http://git-wip-us.apache.org/repos/asf/kafka/blob/ae5a5d7c/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
index 86ac6b3..500aaed 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
@@ -418,7 +418,7 @@ public class ConsumerCoordinatorTest {
leaveRequest.groupId().equals(groupId);
}
}, new LeaveGroupResponse(Errors.NONE.code()).toStruct());
- coordinator.maybeLeaveGroup(false);
+ coordinator.maybeLeaveGroup();
assertTrue(received.get());
assertEquals(JoinGroupRequest.UNKNOWN_MEMBER_ID, coordinator.memberId);
assertEquals(OffsetCommitRequest.DEFAULT_GENERATION_ID, coordinator.generation);
@@ -672,7 +672,7 @@ public class ConsumerCoordinatorTest {
// now switch to manual assignment
client.prepareResponse(new LeaveGroupResponse(Errors.NONE.code()).toStruct());
subscriptions.unsubscribe();
- coordinator.maybeLeaveGroup(false);
+ coordinator.maybeLeaveGroup();
subscriptions.assignFromUser(Arrays.asList(tp));
// the client should not reuse generation/memberId from auto-subscribed generation