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 2017/08/17 22:05:37 UTC
kafka git commit: KAFKA-5730;
Consumer should invoke async commit callback before sync commit
returns
Repository: kafka
Updated Branches:
refs/heads/trunk efe4f6540 -> 334a30bcf
KAFKA-5730; Consumer should invoke async commit callback before sync commit returns
Author: Jason Gustafson <ja...@confluent.io>
Reviewers: Randall Hauch <rh...@gmail.com>, Ewen Cheslack-Postava <me...@ewencp.org>
Closes #3666 from hachikuji/KAFKA-5730
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/334a30bc
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/334a30bc
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/334a30bc
Branch: refs/heads/trunk
Commit: 334a30bcfffc3759aeded08f2089cea4ed6e9937
Parents: efe4f65
Author: Jason Gustafson <ja...@confluent.io>
Authored: Thu Aug 17 15:04:03 2017 -0700
Committer: Jason Gustafson <ja...@confluent.io>
Committed: Thu Aug 17 15:04:03 2017 -0700
----------------------------------------------------------------------
.../org/apache/kafka/clients/KafkaClient.java | 2 +-
.../org/apache/kafka/clients/NetworkClient.java | 8 +++
.../kafka/clients/consumer/KafkaConsumer.java | 16 +++++
.../consumer/internals/AbstractCoordinator.java | 10 +--
.../consumer/internals/ConsumerCoordinator.java | 5 ++
.../internals/ConsumerNetworkClient.java | 15 ++++-
.../apache/kafka/clients/NetworkClientTest.java | 14 ++--
.../internals/ConsumerCoordinatorTest.java | 68 ++++++++++++++++++++
.../internals/ConsumerNetworkClientTest.java | 22 +++++++
9 files changed, 147 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/334a30bc/clients/src/main/java/org/apache/kafka/clients/KafkaClient.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/KafkaClient.java b/clients/src/main/java/org/apache/kafka/clients/KafkaClient.java
index 5bca261..2faebfd 100644
--- a/clients/src/main/java/org/apache/kafka/clients/KafkaClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/KafkaClient.java
@@ -86,7 +86,7 @@ public interface KafkaClient extends Closeable {
List<ClientResponse> poll(long timeout, long now);
/**
- * Diconnects the connection to a particular node, if there is one.
+ * Disconnects the connection to a particular node, if there is one.
* Any pending ClientRequests for this connection will receive disconnections.
*
* @param nodeId The id of the node
http://git-wip-us.apache.org/repos/asf/kafka/blob/334a30bc/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
index 60b1598..4fe55ae 100644
--- a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
@@ -226,6 +226,11 @@ public class NetworkClient implements KafkaClient {
return false;
}
+ // Visible for testing
+ boolean canConnect(Node node, long now) {
+ return connectionStates.canConnect(node.idString(), now);
+ }
+
/**
* Disconnects the connection to a particular node, if there is one.
* Any pending ClientRequests for this connection will receive disconnections.
@@ -234,6 +239,9 @@ public class NetworkClient implements KafkaClient {
*/
@Override
public void disconnect(String nodeId) {
+ if (connectionStates.isDisconnected(nodeId))
+ return;
+
selector.close(nodeId);
List<ApiKeys> requestTypes = new ArrayList<>();
long now = time.milliseconds();
http://git-wip-us.apache.org/repos/asf/kafka/blob/334a30bc/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 4ddd648..f1351b7 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
@@ -1119,6 +1119,9 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
* <p>
* This is a synchronous commits and will block until either the commit succeeds or an unrecoverable error is
* encountered (in which case it is thrown to the caller).
+ * <p>
+ * Note that asynchronous offset commits sent previously with the {@link #commitAsync(OffsetCommitCallback)}
+ * (or similar) are guaranteed to have their callbacks invoked prior to completion of this method.
*
* @throws org.apache.kafka.clients.consumer.CommitFailedException if the commit failed and cannot be retried.
* This can only occur if you are using automatic group management with {@link #subscribe(Collection)},
@@ -1152,6 +1155,9 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
* <p>
* This is a synchronous commits and will block until either the commit succeeds or an unrecoverable error is
* encountered (in which case it is thrown to the caller).
+ * <p>
+ * Note that asynchronous offset commits sent previously with the {@link #commitAsync(OffsetCommitCallback)}
+ * (or similar) are guaranteed to have their callbacks invoked prior to completion of this method.
*
* @param offsets A map of offsets by partition with associated metadata
* @throws org.apache.kafka.clients.consumer.CommitFailedException if the commit failed and cannot be retried.
@@ -1194,6 +1200,11 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
* <p>
* This is an asynchronous call and will not block. Any errors encountered are either passed to the callback
* (if provided) or discarded.
+ * <p>
+ * Offsets committed through multiple calls to this API are guaranteed to be sent in the same order as
+ * the invocations. Corresponding commit callbacks are also invoked in the same order. Additionally note that
+ * offsets committed through this API are guaranteed to complete before a subsequent call to {@link #commitSync()}
+ * (and variants) returns.
*
* @param callback Callback to invoke when the commit completes
*/
@@ -1217,6 +1228,11 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
* <p>
* This is an asynchronous call and will not block. Any errors encountered are either passed to the callback
* (if provided) or discarded.
+ * <p>
+ * Offsets committed through multiple calls to this API are guaranteed to be sent in the same order as
+ * the invocations. Corresponding commit callbacks are also invoked in the same order. Additionally note that
+ * offsets committed through this API are guaranteed to complete before a subsequent call to {@link #commitSync()}
+ * (and variants) returns.
*
* @param offsets A map of offsets by partition with associate metadata. This map will be copied internally, so it
* is safe to mutate the map after returning.
http://git-wip-us.apache.org/repos/asf/kafka/blob/334a30bc/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 0f594df..74ef20a 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
@@ -21,8 +21,8 @@ import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.errors.DisconnectException;
import org.apache.kafka.common.errors.GroupAuthorizationException;
-import org.apache.kafka.common.errors.CoordinatorNotAvailableException;
import org.apache.kafka.common.errors.IllegalGenerationException;
+import org.apache.kafka.common.errors.InterruptException;
import org.apache.kafka.common.errors.RebalanceInProgressException;
import org.apache.kafka.common.errors.RetriableException;
import org.apache.kafka.common.errors.UnknownMemberIdException;
@@ -59,7 +59,6 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
-import org.apache.kafka.common.errors.InterruptException;
/**
* AbstractCoordinator implements group management for a single group member by interacting with
@@ -247,8 +246,6 @@ public abstract class AbstractCoordinator implements Closeable {
// 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?
log.debug("No broker available to send GroupCoordinator request for group {}", groupId);
return RequestFuture.noBrokersAvailable();
} else
@@ -651,7 +648,10 @@ public abstract class AbstractCoordinator implements Closeable {
protected synchronized void coordinatorDead() {
if (this.coordinator != null) {
log.info("Marking the coordinator {} dead for group {}", this.coordinator, groupId);
- client.failUnsentRequests(this.coordinator, CoordinatorNotAvailableException.INSTANCE);
+
+ // Disconnect from the coordinator to ensure that there are no in-flight requests remaining.
+ // Pending callbacks will be invoked with a DisconnectException.
+ client.disconnect(this.coordinator);
this.coordinator = null;
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/334a30bc/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
index d39b369..5ba9ccb 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
@@ -604,6 +604,11 @@ public final class ConsumerCoordinator extends AbstractCoordinator {
RequestFuture<Void> future = sendOffsetCommitRequest(offsets);
client.poll(future, remainingMs);
+ // We may have had in-flight offset commits when the synchronous commit began. If so, ensure that
+ // the corresponding callbacks are invoked prior to returning in order to preserve the order that
+ // the offset commits were applied.
+ invokeCompletedOffsetCommitCallbacks();
+
if (future.succeeded()) {
if (interceptors != null)
interceptors.onCommit(offsets);
http://git-wip-us.apache.org/repos/asf/kafka/blob/334a30bc/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java
index 84e9a81..bb7f9f2 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java
@@ -263,8 +263,7 @@ public class ConsumerNetworkClient implements Closeable {
}
/**
- * Poll for network IO and return immediately. This will not trigger wakeups,
- * nor will it execute any delayed tasks.
+ * Poll for network IO and return immediately. This will not trigger wakeups.
*/
public void pollNoWakeup() {
poll(0, time.milliseconds(), null, true);
@@ -374,6 +373,16 @@ public class ConsumerNetworkClient implements Closeable {
}
}
+ public void disconnect(Node node) {
+ synchronized (this) {
+ failUnsentRequests(node, DisconnectException.INSTANCE);
+ client.disconnect(node.idString());
+ }
+
+ // We need to poll to ensure callbacks from in-flight requests on the disconnected socket are fired
+ pollNoWakeup();
+ }
+
private void failExpiredRequests(long now) {
// clear all expired unsent requests and fail their corresponding futures
Collection<ClientRequest> expiredRequests = unsent.removeExpiredRequests(now, unsentExpiryMs);
@@ -383,7 +392,7 @@ public class ConsumerNetworkClient implements Closeable {
}
}
- public void failUnsentRequests(Node node, RuntimeException e) {
+ private void failUnsentRequests(Node node, RuntimeException e) {
// clear unsent requests to node and fail their corresponding futures
synchronized (this) {
Collection<ClientRequest> unsentRequests = unsent.remove(node);
http://git-wip-us.apache.org/repos/asf/kafka/blob/334a30bc/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java b/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java
index 77960e1..069fb7a 100644
--- a/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java
@@ -22,14 +22,13 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.network.NetworkReceive;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.types.Struct;
-import org.apache.kafka.common.record.RecordBatch;
import org.apache.kafka.common.record.MemoryRecords;
+import org.apache.kafka.common.record.RecordBatch;
import org.apache.kafka.common.requests.ApiVersionsResponse;
import org.apache.kafka.common.requests.MetadataRequest;
import org.apache.kafka.common.requests.ProduceRequest;
import org.apache.kafka.common.requests.ResponseHeader;
import org.apache.kafka.common.utils.MockTime;
-import org.apache.kafka.common.utils.Time;
import org.apache.kafka.test.DelayedReceive;
import org.apache.kafka.test.MockSelector;
import org.apache.kafka.test.TestUtils;
@@ -270,14 +269,21 @@ public class NetworkClientTest {
public void testCallDisconnect() throws Exception {
awaitReady(client, node);
assertTrue("Expected NetworkClient to be ready to send to node " + node.idString(),
- client.isReady(node, Time.SYSTEM.milliseconds()));
+ client.isReady(node, time.milliseconds()));
assertFalse("Did not expect connection to node " + node.idString() + " to be failed",
client.connectionFailed(node));
client.disconnect(node.idString());
assertFalse("Expected node " + node.idString() + " to be disconnected.",
- client.isReady(node, Time.SYSTEM.milliseconds()));
+ client.isReady(node, time.milliseconds()));
assertTrue("Expected connection to node " + node.idString() + " to be failed after disconnect",
client.connectionFailed(node));
+ assertFalse(client.canConnect(node, time.milliseconds()));
+
+ // ensure disconnect does not reset blackout period if already disconnected
+ time.sleep(reconnectBackoffMsTest);
+ assertTrue(client.canConnect(node, time.milliseconds()));
+ client.disconnect(node.idString());
+ assertTrue(client.canConnect(node, time.milliseconds()));
}
private static class TestCallbackHandler implements RequestCompletionHandler {
http://git-wip-us.apache.org/repos/asf/kafka/blob/334a30bc/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 3c1b411..18e18ed 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
@@ -58,6 +58,7 @@ import org.junit.Before;
import org.junit.Test;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
@@ -951,6 +952,37 @@ public class ConsumerCoordinatorTest {
}
@Test
+ public void testCoordinatorDisconnectAfterNotCoordinatorError() {
+ testInFlightRequestsFailedAfterCoordinatorMarkedDead(Errors.NOT_COORDINATOR);
+ }
+
+ @Test
+ public void testCoordinatorDisconnectAfterCoordinatorNotAvailableError() {
+ testInFlightRequestsFailedAfterCoordinatorMarkedDead(Errors.COORDINATOR_NOT_AVAILABLE);
+ }
+
+ private void testInFlightRequestsFailedAfterCoordinatorMarkedDead(Errors error) {
+ client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));
+ coordinator.ensureCoordinatorReady();
+
+ // Send two async commits and fail the first one with an error.
+ // This should cause a coordinator disconnect which will cancel the second request.
+
+ MockCommitCallback firstCommitCallback = new MockCommitCallback();
+ MockCommitCallback secondCommitCallback = new MockCommitCallback();
+ coordinator.commitOffsetsAsync(Collections.singletonMap(t1p, new OffsetAndMetadata(100L)), firstCommitCallback);
+ coordinator.commitOffsetsAsync(Collections.singletonMap(t1p, new OffsetAndMetadata(100L)), secondCommitCallback);
+
+ client.respond(offsetCommitResponse(Collections.singletonMap(t1p, error)));
+ consumerClient.pollNoWakeup();
+ coordinator.invokeCompletedOffsetCommitCallbacks();
+
+ assertTrue(coordinator.coordinatorUnknown());
+ assertTrue(firstCommitCallback.exception instanceof RetriableCommitFailedException);
+ assertTrue(secondCommitCallback.exception instanceof RetriableCommitFailedException);
+ }
+
+ @Test
public void testAutoCommitDynamicAssignment() {
final String consumerId = "consumer";
@@ -1212,6 +1244,42 @@ public class ConsumerCoordinatorTest {
coordinator.commitOffsetsSync(Collections.singletonMap(t1p, new OffsetAndMetadata(100L)), Long.MAX_VALUE);
}
+ @Test
+ public void testAsyncCommitCallbacksInvokedPriorToSyncCommitCompletion() throws Exception {
+ client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));
+ coordinator.ensureCoordinatorReady();
+
+ final List<OffsetAndMetadata> committedOffsets = Collections.synchronizedList(new ArrayList<OffsetAndMetadata>());
+ final OffsetAndMetadata firstOffset = new OffsetAndMetadata(0L);
+ final OffsetAndMetadata secondOffset = new OffsetAndMetadata(1L);
+
+ coordinator.commitOffsetsAsync(Collections.singletonMap(t1p, firstOffset), new OffsetCommitCallback() {
+ @Override
+ public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
+ committedOffsets.add(firstOffset);
+ }
+ });
+
+ // Do a synchronous commit in the background so that we can send both responses at the same time
+ Thread thread = new Thread() {
+ @Override
+ public void run() {
+ coordinator.commitOffsetsSync(Collections.singletonMap(t1p, secondOffset), 10000);
+ committedOffsets.add(secondOffset);
+ }
+ };
+
+ thread.start();
+
+ client.waitForRequests(2, 5000);
+ client.respond(offsetCommitResponse(Collections.singletonMap(t1p, Errors.NONE)));
+ client.respond(offsetCommitResponse(Collections.singletonMap(t1p, Errors.NONE)));
+
+ thread.join();
+
+ assertEquals(Arrays.asList(firstOffset, secondOffset), committedOffsets);
+ }
+
@Test(expected = KafkaException.class)
public void testCommitUnknownTopicOrPartition() {
client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));
http://git-wip-us.apache.org/repos/asf/kafka/blob/334a30bc/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClientTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClientTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClientTest.java
index b46b657..8e71dd5 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClientTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClientTest.java
@@ -22,6 +22,7 @@ import org.apache.kafka.clients.MockClient;
import org.apache.kafka.clients.NetworkClient;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.Node;
+import org.apache.kafka.common.errors.DisconnectException;
import org.apache.kafka.common.errors.WakeupException;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.requests.HeartbeatRequest;
@@ -81,6 +82,27 @@ public class ConsumerNetworkClientTest {
}
@Test
+ public void testDisconnectWithUnsentRequests() {
+ RequestFuture<ClientResponse> future = consumerClient.send(node, heartbeat());
+ assertTrue(consumerClient.hasPendingRequests(node));
+ assertFalse(client.hasInFlightRequests(node.idString()));
+ consumerClient.disconnect(node);
+ assertTrue(future.failed());
+ assertTrue(future.exception() instanceof DisconnectException);
+ }
+
+ @Test
+ public void testDisconnectWithInFlightRequests() {
+ RequestFuture<ClientResponse> future = consumerClient.send(node, heartbeat());
+ consumerClient.pollNoWakeup();
+ assertTrue(consumerClient.hasPendingRequests(node));
+ assertTrue(client.hasInFlightRequests(node.idString()));
+ consumerClient.disconnect(node);
+ assertTrue(future.failed());
+ assertTrue(future.exception() instanceof DisconnectException);
+ }
+
+ @Test
public void doNotBlockIfPollConditionIsSatisfied() {
NetworkClient mockNetworkClient = EasyMock.mock(NetworkClient.class);
ConsumerNetworkClient consumerClient = new ConsumerNetworkClient(mockNetworkClient, metadata, time, 100, 1000);