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-5730; Consumer should invoke async commit callback before sync commit returns
Date Thu, 17 Aug 2017 22:05:37 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk efe4f6540 -> 334a30bcf


KAFKA-5730; Consumer should invoke async commit callback before sync commit returns

Author: Jason Gustafson <jason@confluent.io>

Reviewers: Randall Hauch <rhauch@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 <jason@confluent.io>
Authored: Thu Aug 17 15:04:03 2017 -0700
Committer: Jason Gustafson <jason@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);


Mime
View raw message