kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rsiva...@apache.org
Subject [kafka] branch trunk updated: MINOR: Skip sending fetches/offset lookups when awaiting the reconnect backoff (#4644)
Date Thu, 08 Mar 2018 11:00:25 GMT
This is an automated email from the ASF dual-hosted git repository.

rsivaram pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new 925d6a2  MINOR: Skip sending fetches/offset lookups when awaiting the reconnect backoff
(#4644)
925d6a2 is described below

commit 925d6a2ef363ee540e65f7ec3b699dce05aa591c
Author: Jason Gustafson <jason@confluent.io>
AuthorDate: Thu Mar 8 03:00:20 2018 -0800

    MINOR: Skip sending fetches/offset lookups when awaiting the reconnect backoff (#4644)
    
    Logging can get spammy during the reconnect blackout period because any requests we send
to ConsumerNetworkClient will immediately be failed when poll() returns. This patch checks
for connection failures prior to sending fetches and offset lookups and skips sending to any
failed nodes. Test cases added for both.
    
    Reviewers: Rajini Sivaram <rajinisivaram@googlemail.com>
---
 .../kafka/clients/ClusterConnectionStates.java     |  1 +
 .../org/apache/kafka/clients/NetworkClient.java    | 11 +++----
 .../consumer/internals/AbstractCoordinator.java    |  4 +--
 .../consumer/internals/ConsumerNetworkClient.java  | 31 +++++++++++++------
 .../kafka/clients/consumer/internals/Fetcher.java  | 29 ++++++++++++-----
 .../kafka/common/requests/FetchResponse.java       |  2 +-
 .../kafka/clients/ClusterConnectionStatesTest.java |  4 ++-
 .../java/org/apache/kafka/clients/MockClient.java  | 14 ++++++---
 .../clients/consumer/internals/FetcherTest.java    | 36 ++++++++++++++++++++++
 9 files changed, 100 insertions(+), 32 deletions(-)

diff --git a/clients/src/main/java/org/apache/kafka/clients/ClusterConnectionStates.java b/clients/src/main/java/org/apache/kafka/clients/ClusterConnectionStates.java
index 89e958d..7abcf12 100644
--- a/clients/src/main/java/org/apache/kafka/clients/ClusterConnectionStates.java
+++ b/clients/src/main/java/org/apache/kafka/clients/ClusterConnectionStates.java
@@ -144,6 +144,7 @@ final class ClusterConnectionStates {
     public void ready(String id) {
         NodeConnectionState nodeState = nodeState(id);
         nodeState.state = ConnectionState.READY;
+        nodeState.authenticationException = null;
         resetReconnectBackoff(nodeState);
     }
 
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 4b291b5..3a9cbc4 100644
--- a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
@@ -603,11 +603,13 @@ public class NetworkClient implements KafkaClient {
         nodesNeedingApiVersionsFetch.remove(nodeId);
         switch (disconnectState.state()) {
             case AUTHENTICATION_FAILED:
-                connectionStates.authenticationFailed(nodeId, now, disconnectState.exception());
-                log.error("Connection to node {} failed authentication due to: {}", nodeId,
disconnectState.exception().getMessage());
+                AuthenticationException exception = disconnectState.exception();
+                connectionStates.authenticationFailed(nodeId, now, exception);
+                metadataUpdater.handleAuthenticationFailure(exception);
+                log.error("Connection to node {} failed authentication due to: {}", nodeId,
exception.getMessage());
                 break;
             case AUTHENTICATE:
-                // This warning applies to older brokers which dont provide feedback on authentication
failures
+                // This warning applies to older brokers which don't provide feedback on
authentication failures
                 log.warn("Connection to node {} terminated during authentication. This may
indicate " +
                         "that authentication failed due to invalid credentials.", nodeId);
                 break;
@@ -625,9 +627,6 @@ public class NetworkClient implements KafkaClient {
             else if (request.header.apiKey() == ApiKeys.METADATA)
                 metadataUpdater.handleDisconnection(request.destination);
         }
-        AuthenticationException authenticationException = connectionStates.authenticationException(nodeId);
-        if (authenticationException != null)
-            metadataUpdater.handleAuthenticationFailure(authenticationException);
     }
 
     /**
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 c3edaa7..dd4bb70 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
@@ -228,7 +228,7 @@ public abstract class AbstractCoordinator implements Closeable {
                     client.awaitMetadataUpdate(remainingMs);
                 } else
                     throw future.exception();
-            } else if (coordinator != null && client.connectionFailed(coordinator))
{
+            } else if (coordinator != null && client.isUnavailable(coordinator))
{
                 // we found the coordinator, but the connection has failed, so mark
                 // it dead and backoff before retrying discovery
                 markCoordinatorUnknown();
@@ -637,7 +637,7 @@ public abstract class AbstractCoordinator implements Closeable {
      * @return the current coordinator or null if it is unknown
      */
     protected synchronized Node checkAndGetCoordinator() {
-        if (coordinator != null && client.connectionFailed(coordinator)) {
+        if (coordinator != null && client.isUnavailable(coordinator)) {
             markCoordinatorUnknown(true);
             return null;
         }
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 fb393a5..e1598fa 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
@@ -28,7 +28,6 @@ import org.apache.kafka.common.errors.InterruptException;
 import org.apache.kafka.common.errors.TimeoutException;
 import org.apache.kafka.common.errors.WakeupException;
 import org.apache.kafka.common.requests.AbstractRequest;
-import org.apache.kafka.common.requests.RequestHeader;
 import org.apache.kafka.common.utils.LogContext;
 import org.apache.kafka.common.utils.Time;
 import org.slf4j.Logger;
@@ -510,15 +509,29 @@ public class ConsumerNetworkClient implements Closeable {
         }
     }
 
+
+    /**
+     * Check if the code is disconnected and unavailable for immediate reconnection (i.e.
if it is in
+     * reconnect backoff window following the disconnect).
+     */
+    public boolean isUnavailable(Node node) {
+        lock.lock();
+        try {
+            return client.connectionFailed(node) && client.connectionDelay(node,
time.milliseconds()) > 0;
+        } finally {
+            lock.unlock();
+        }
+    }
+
     /**
-     * Find whether a previous connection has failed. Note that the failure state will persist
until either
-     * {@link #tryConnect(Node)} or {@link #send(Node, AbstractRequest.Builder)} has been
called.
-     * @param node Node to connect to if possible
+     * Check for an authentication error on a given node and raise the exception if there
is one.
      */
-    public boolean connectionFailed(Node node) {
+    public void maybeThrowAuthFailure(Node node) {
         lock.lock();
         try {
-            return client.connectionFailed(node);
+            AuthenticationException exception = client.authenticationException(node);
+            if (exception != null)
+                throw exception;
         } finally {
             lock.unlock();
         }
@@ -552,10 +565,8 @@ public class ConsumerNetworkClient implements Closeable {
             if (e != null) {
                 future.raise(e);
             } else if (response.wasDisconnected()) {
-                RequestHeader requestHeader = response.requestHeader();
-                int correlation = requestHeader.correlationId();
-                log.debug("Cancelled {} request {} with correlation id {} due to node {}
being disconnected",
-                        requestHeader.apiKey(), requestHeader, correlation, response.destination());
+                log.debug("Cancelled request with header {} due to node {} being disconnected",
+                        response.requestHeader(), response.destination());
                 future.raise(DisconnectException.INSTANCE);
             } else if (response.versionMismatch() != null) {
                 future.raise(response.versionMismatch());
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
index b46a3a2..6d8fb6c 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
@@ -576,10 +576,11 @@ public class Fetcher<K, V> implements SubscriptionState.Listener,
Closeable {
 
         Map<Node, Map<TopicPartition, Long>> timestampsToSearchByNode = groupListOffsetRequests(partitionResetTimestamps);
         for (Map.Entry<Node, Map<TopicPartition, Long>> entry : timestampsToSearchByNode.entrySet())
{
+            Node node = entry.getKey();
             final Map<TopicPartition, Long> resetTimestamps = entry.getValue();
             subscriptions.setResetPending(resetTimestamps.keySet(), time.milliseconds() +
requestTimeoutMs);
 
-            RequestFuture<ListOffsetResult> future = sendListOffsetRequest(entry.getKey(),
resetTimestamps, false);
+            RequestFuture<ListOffsetResult> future = sendListOffsetRequest(node, resetTimestamps,
false);
             future.addListener(new RequestFutureListener<ListOffsetResult>() {
                 @Override
                 public void onSuccess(ListOffsetResult result) {
@@ -667,11 +668,19 @@ public class Fetcher<K, V> implements SubscriptionState.Listener,
Closeable {
             PartitionInfo info = metadata.fetch().partition(tp);
             if (info == null) {
                 metadata.add(tp.topic());
-                log.debug("Partition {} is unknown for fetching offset", tp);
+                log.debug("Leader for partition {} is unknown for fetching offset", tp);
                 metadata.requestUpdate();
             } else if (info.leader() == null) {
-                log.debug("Leader for partition {} unavailable for fetching offset", tp);
+                log.debug("Leader for partition {} is unavailable for fetching offset", tp);
                 metadata.requestUpdate();
+            } else if (client.isUnavailable(info.leader())) {
+                client.maybeThrowAuthFailure(info.leader());
+
+                // The connection has failed and we need to await the blackout period before
we can
+                // try again. No need to request a metadata update since the disconnect will
have
+                // done so already.
+                log.debug("Leader {} for partition {} is unavailable for fetching offset
until reconnect backoff expires",
+                        info.leader(), tp);
             } else {
                 Node node = info.leader();
                 Map<TopicPartition, Long> topicData = timestampsToSearchByNode.get(node);
@@ -823,14 +832,21 @@ public class Fetcher<K, V> implements SubscriptionState.Listener,
Closeable {
      * that have no existing requests in flight.
      */
     private Map<Node, FetchSessionHandler.FetchRequestData> prepareFetchRequests()
{
-        // create the fetch info
         Cluster cluster = metadata.fetch();
         Map<Node, FetchSessionHandler.Builder> fetchable = new LinkedHashMap<>();
         for (TopicPartition partition : fetchablePartitions()) {
             Node node = cluster.leaderFor(partition);
             if (node == null) {
                 metadata.requestUpdate();
-            } else if (!this.client.hasPendingRequests(node)) {
+            } else if (client.isUnavailable(node)) {
+                client.maybeThrowAuthFailure(node);
+
+                // If we try to send during the reconnect blackout window, then the request
is just
+                // going to be failed anyway before being sent, so skip the send for now
+                log.trace("Skipping fetch for partition {} because node {} is awaiting reconnect
backoff", partition, node);
+            } else if (client.hasPendingRequests(node)) {
+                log.trace("Skipping fetch for partition {} because there is an in-flight
request to {}", partition, node);
+            } else {
                 // if there is a leader and no in-flight requests, issue a new fetch
                 FetchSessionHandler.Builder builder = fetchable.get(node);
                 if (builder == null) {
@@ -846,10 +862,9 @@ public class Fetcher<K, V> implements SubscriptionState.Listener,
Closeable {
                 long position = this.subscriptions.position(partition);
                 builder.add(partition, new FetchRequest.PartitionData(position, FetchRequest.INVALID_LOG_START_OFFSET,
                     this.fetchSize));
+
                 log.debug("Added {} fetch request for partition {} at offset {} to node {}",
isolationLevel,
                     partition, position, node);
-            } else {
-                log.trace("Skipping fetch for partition {} because there is an in-flight
request to {}", partition, node);
             }
         }
         Map<Node, FetchSessionHandler.FetchRequestData> reqs = new LinkedHashMap<>();
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
index ed0f5a3..d0380f1 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
@@ -152,7 +152,7 @@ public class FetchResponse extends AbstractResponse {
      */
     private static final Schema FETCH_RESPONSE_V6 = FETCH_RESPONSE_V5;
 
-    // FETCH_REESPONSE_V7 added incremental fetch responses and a top-level error code.
+    // FETCH_RESPONSE_V7 added incremental fetch responses and a top-level error code.
     public static final Field.Int32 SESSION_ID = new Field.Int32("session_id", "The fetch
session ID");
 
     private static final Schema FETCH_RESPONSE_V7 = new Schema(
diff --git a/clients/src/test/java/org/apache/kafka/clients/ClusterConnectionStatesTest.java
b/clients/src/test/java/org/apache/kafka/clients/ClusterConnectionStatesTest.java
index 3b246f4..03f4d02 100644
--- a/clients/src/test/java/org/apache/kafka/clients/ClusterConnectionStatesTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/ClusterConnectionStatesTest.java
@@ -19,6 +19,7 @@ package org.apache.kafka.clients;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
 import org.apache.kafka.common.errors.AuthenticationException;
@@ -141,9 +142,10 @@ public class ClusterConnectionStatesTest {
         time.sleep(connectionStates.connectionDelay(nodeId1, time.milliseconds()) + 1);
 
         assertTrue(connectionStates.canConnect(nodeId1, time.milliseconds()));
+        connectionStates.ready(nodeId1);
+        assertNull(connectionStates.authenticationException(nodeId1));
     }
 
-
     @Test
     public void testRemoveNode() {
         connectionStates.connecting(nodeId1, time.milliseconds());
diff --git a/clients/src/test/java/org/apache/kafka/clients/MockClient.java b/clients/src/test/java/org/apache/kafka/clients/MockClient.java
index faf9240..60af9bc 100644
--- a/clients/src/test/java/org/apache/kafka/clients/MockClient.java
+++ b/clients/src/test/java/org/apache/kafka/clients/MockClient.java
@@ -79,7 +79,7 @@ public class MockClient implements KafkaClient {
     private Node node = null;
     private final Set<String> ready = new HashSet<>();
     private final Map<Node, Long> blackedOut = new HashMap<>();
-    private final Map<Node, AuthenticationException> authenticationException = new
HashMap<>();
+    private final Map<Node, AuthenticationException> authenticationErrors = new HashMap<>();
     // Use concurrent queue for requests so that requests may be queried from a different
thread
     private final Queue<ClientRequest> requests = new ConcurrentLinkedDeque<>();
     // Use concurrent queue for responses so that responses may be updated during poll()
from a different thread.
@@ -106,14 +106,18 @@ public class MockClient implements KafkaClient {
 
     @Override
     public boolean ready(Node node, long now) {
-        if (isBlackedOut(node) || authenticationException(node) != null)
+        if (isBlackedOut(node))
             return false;
+        authenticationErrors.remove(node);
         ready.add(node.idString());
         return true;
     }
 
     @Override
     public long connectionDelay(Node node, long now) {
+        Long blackoutExpiration = blackedOut.get(node);
+        if (blackoutExpiration != null)
+            return Math.max(0, blackoutExpiration - now);
         return 0;
     }
 
@@ -122,7 +126,7 @@ public class MockClient implements KafkaClient {
     }
 
     public void authenticationFailed(Node node, long duration) {
-        authenticationException.put(node, (AuthenticationException) Errors.SASL_AUTHENTICATION_FAILED.exception());
+        authenticationErrors.put(node, (AuthenticationException) Errors.SASL_AUTHENTICATION_FAILED.exception());
         disconnect(node.idString());
         blackout(node, duration);
     }
@@ -147,7 +151,7 @@ public class MockClient implements KafkaClient {
 
     @Override
     public AuthenticationException authenticationException(Node node) {
-        return authenticationException.get(node);
+        return authenticationErrors.get(node);
     }
 
     @Override
@@ -389,7 +393,7 @@ public class MockClient implements KafkaClient {
         responses.clear();
         futureResponses.clear();
         metadataUpdates.clear();
-        authenticationException.clear();
+        authenticationErrors.clear();
     }
 
     public boolean hasPendingMetadataUpdates() {
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
index 27aba41..6de1186 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
@@ -199,6 +199,18 @@ public class FetcherTest {
     }
 
     @Test
+    public void testFetchSkipsBlackedOutNodes() {
+        subscriptions.assignFromUser(singleton(tp0));
+        subscriptions.seek(tp0, 0);
+
+        client.blackout(node, 500);
+        assertEquals(0, fetcher.sendFetches());
+
+        time.sleep(500);
+        assertEquals(1, fetcher.sendFetches());
+    }
+
+    @Test
     public void testFetcherIgnoresControlRecords() {
         subscriptions.assignFromUser(singleton(tp0));
         subscriptions.seek(tp0, 0);
@@ -962,6 +974,30 @@ public class FetcherTest {
     }
 
     @Test
+    public void testResetOffsetsSkipsBlackedOutConnections() {
+        subscriptions.assignFromUser(singleton(tp0));
+        subscriptions.requestOffsetReset(tp0, OffsetResetStrategy.EARLIEST);
+
+        // Check that we skip sending the ListOffset request when the node is blacked out
+        client.blackout(node, 500);
+        fetcher.resetOffsetsIfNeeded();
+        assertEquals(0, consumerClient.pendingRequestCount());
+        consumerClient.pollNoWakeup();
+        assertTrue(subscriptions.isOffsetResetNeeded(tp0));
+        assertEquals(OffsetResetStrategy.EARLIEST, subscriptions.resetStrategy(tp0));
+
+        time.sleep(500);
+        client.prepareResponse(listOffsetRequestMatcher(ListOffsetRequest.EARLIEST_TIMESTAMP),
+                listOffsetResponse(Errors.NONE, 1L, 5L));
+        fetcher.resetOffsetsIfNeeded();
+        consumerClient.pollNoWakeup();
+
+        assertFalse(subscriptions.isOffsetResetNeeded(tp0));
+        assertTrue(subscriptions.isFetchable(tp0));
+        assertEquals(5, subscriptions.position(tp0).longValue());
+    }
+
+    @Test
     public void testUpdateFetchPositionResetToEarliestOffset() {
         subscriptions.assignFromUser(singleton(tp0));
         subscriptions.requestOffsetReset(tp0, OffsetResetStrategy.EARLIEST);

-- 
To stop receiving notification emails like this one, please contact
rsivaram@apache.org.

Mime
View raw message