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 2021/11/15 18:38:04 UTC
[kafka] branch 3.1 updated: MINOR: Log client disconnect events at INFO level (#11449)
This is an automated email from the ASF dual-hosted git repository.
jgus pushed a commit to branch 3.1
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.1 by this push:
new 79c4f0e MINOR: Log client disconnect events at INFO level (#11449)
79c4f0e is described below
commit 79c4f0e3f3f536d69e71a7d664216e1e31daba53
Author: Jason Gustafson <ja...@confluent.io>
AuthorDate: Wed Nov 10 09:56:55 2021 -0800
MINOR: Log client disconnect events at INFO level (#11449)
Client disconnects are crucial events for debugging. The fact that we only log them at DEBUG/TRACE means we rarely have them available outside of a testing context. This patch therefore increases verbosity to INFO level. In practice, we already have backoff configurations which should prevent these logs from getting too spammy.
Reviewers: Guozhang Wang <wa...@gmail.com>
---
.../org/apache/kafka/clients/InFlightRequests.java | 3 +-
.../org/apache/kafka/clients/NetworkClient.java | 41 +++++++++++++++-------
.../consumer/internals/AbstractCoordinator.java | 4 ++-
3 files changed, 33 insertions(+), 15 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java b/clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java
index efca453..6f5477e 100644
--- a/clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java
+++ b/clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java
@@ -158,8 +158,7 @@ final class InFlightRequests {
private Boolean hasExpiredRequest(long now, Deque<NetworkClient.InFlightRequest> deque) {
for (NetworkClient.InFlightRequest request : deque) {
- long timeSinceSend = Math.max(0, now - request.sendTimeMs);
- if (timeSinceSend > request.requestTimeoutMs)
+ if (request.timeElapsedSinceSendMs(now) > request.requestTimeoutMs)
return true;
}
return false;
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 baceaba..e7a0ac7 100644
--- a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
@@ -316,26 +316,34 @@ public class NetworkClient implements KafkaClient {
*/
@Override
public void disconnect(String nodeId) {
- if (connectionStates.isDisconnected(nodeId))
+ if (connectionStates.isDisconnected(nodeId)) {
+ log.debug("Client requested disconnect from node {}, which is already disconnected", nodeId);
return;
+ }
+ log.info("Client requested disconnect from node {}", nodeId);
selector.close(nodeId);
long now = time.milliseconds();
-
cancelInFlightRequests(nodeId, now, abortedSends);
-
connectionStates.disconnected(nodeId, now);
-
- if (log.isTraceEnabled()) {
- log.trace("Manually disconnected from {}. Aborted in-flight requests: {}.", nodeId, inFlightRequests);
- }
}
private void cancelInFlightRequests(String nodeId, long now, Collection<ClientResponse> responses) {
Iterable<InFlightRequest> inFlightRequests = this.inFlightRequests.clearAll(nodeId);
for (InFlightRequest request : inFlightRequests) {
- log.trace("Cancelled request {} {} with correlation id {} due to node {} being disconnected",
- request.header.apiKey(), request.request, request.header.correlationId(), nodeId);
+ if (log.isDebugEnabled()) {
+ log.debug("Cancelled in-flight {} request with correlation id {} due to node {} being disconnected " +
+ "(elapsed time since creation: {}ms, elapsed time since send: {}ms, request timeout: {}ms): {}",
+ request.header.apiKey(), request.header.correlationId(), nodeId,
+ request.timeElapsedSinceCreateMs(now), request.timeElapsedSinceSendMs(now),
+ request.requestTimeoutMs, request.request);
+ } else {
+ log.info("Cancelled in-flight {} request with correlation id {} due to node {} being disconnected " +
+ "(elapsed time since creation: {}ms, elapsed time since send: {}ms, request timeout: {}ms)",
+ request.header.apiKey(), request.header.correlationId(), nodeId,
+ request.timeElapsedSinceCreateMs(now), request.timeElapsedSinceSendMs(now),
+ request.requestTimeoutMs);
+ }
if (!request.isInternalRequest) {
if (responses != null)
@@ -355,6 +363,7 @@ public class NetworkClient implements KafkaClient {
*/
@Override
public void close(String nodeId) {
+ log.info("Client requested connection close from node {}", nodeId);
selector.close(nodeId);
long now = time.milliseconds();
cancelInFlightRequests(nodeId, now, null);
@@ -785,7 +794,7 @@ public class NetworkClient implements KafkaClient {
for (String nodeId : nodeIds) {
// close connection to the node
this.selector.close(nodeId);
- log.debug("Disconnecting from node {} due to request timeout.", nodeId);
+ log.info("Disconnecting from node {} due to request timeout.", nodeId);
processDisconnection(responses, nodeId, now, ChannelState.LOCAL_CLOSE);
}
}
@@ -807,7 +816,7 @@ public class NetworkClient implements KafkaClient {
List<String> nodes = connectionStates.nodesWithConnectionSetupTimeout(now);
for (String nodeId : nodes) {
this.selector.close(nodeId);
- log.debug(
+ log.info(
"Disconnecting from node {} due to socket connection setup timeout. " +
"The timeout value is {} ms.",
nodeId,
@@ -923,7 +932,7 @@ public class NetworkClient implements KafkaClient {
private void handleDisconnections(List<ClientResponse> responses, long now) {
for (Map.Entry<String, ChannelState> entry : this.selector.disconnected().entrySet()) {
String node = entry.getKey();
- log.debug("Node {} disconnected.", node);
+ log.info("Node {} disconnected.", node);
processDisconnection(responses, node, now, entry.getValue());
}
}
@@ -1251,6 +1260,14 @@ public class NetworkClient implements KafkaClient {
this.sendTimeMs = sendTimeMs;
}
+ public long timeElapsedSinceSendMs(long currentTimeMs) {
+ return Math.max(0, currentTimeMs - sendTimeMs);
+ }
+
+ public long timeElapsedSinceCreateMs(long currentTimeMs) {
+ return Math.max(0, currentTimeMs - createdTimeMs);
+ }
+
public ClientResponse completed(AbstractResponse response, long timeMs) {
return new ClientResponse(header, callback, destination, createdTimeMs, timeMs,
false, null, null, response);
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 2118861..6f16b34 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
@@ -925,8 +925,10 @@ public abstract class AbstractCoordinator implements Closeable {
// Disconnect from the coordinator to ensure that there are no in-flight requests remaining.
// Pending callbacks will be invoked with a DisconnectException on the next call to poll.
- if (!isDisconnected)
+ if (!isDisconnected) {
+ log.info("Requesting disconnect from last known coordinator {}", oldCoordinator);
client.disconnectAsync(oldCoordinator);
+ }
lastTimeOfConnectionMs = time.milliseconds();
} else {