You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by da...@apache.org on 2023/03/09 09:04:35 UTC
[kafka] branch trunk updated: KAFKA-14317: ProduceRequest timeouts are logged as network exceptions (#12813)
This is an automated email from the ASF dual-hosted git repository.
dajac 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 c5240c03908 KAFKA-14317: ProduceRequest timeouts are logged as network exceptions (#12813)
c5240c03908 is described below
commit c5240c0390892fe9ecbe5285185c370e7be8b2aa
Author: Kirk True <ki...@kirktrue.pro>
AuthorDate: Thu Mar 9 01:04:07 2023 -0800
KAFKA-14317: ProduceRequest timeouts are logged as network exceptions (#12813)
Minor changes to `Sender` and `NetworkClient` so that we can log timeouts during `ProduceRequest` with a more precise error message, denoting a timeout vs. "generic" network error.
Reviewers: Philip Nee <pn...@confluent.io>, Guozhang Wang <gu...@apache.org>, David Jacot <dj...@confluent.io>
---
.../org/apache/kafka/clients/ClientResponse.java | 47 +++++++++++++
.../org/apache/kafka/clients/NetworkClient.java | 65 +++++++++++++++---
.../kafka/clients/producer/internals/Sender.java | 8 ++-
.../apache/kafka/clients/NetworkClientTest.java | 77 +++++++++++++++-------
4 files changed, 164 insertions(+), 33 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/ClientResponse.java b/clients/src/main/java/org/apache/kafka/clients/ClientResponse.java
index 446bf44010f..2135dfa472b 100644
--- a/clients/src/main/java/org/apache/kafka/clients/ClientResponse.java
+++ b/clients/src/main/java/org/apache/kafka/clients/ClientResponse.java
@@ -33,6 +33,7 @@ public class ClientResponse {
private final long receivedTimeMs;
private final long latencyMs;
private final boolean disconnected;
+ private final boolean timedOut;
private final UnsupportedVersionException versionMismatch;
private final AuthenticationException authenticationException;
private final AbstractResponse responseBody;
@@ -40,10 +41,45 @@ public class ClientResponse {
/**
* @param requestHeader The header of the corresponding request
* @param callback The callback to be invoked
+ * @param destination The node the corresponding request was sent to
* @param createdTimeMs The unix timestamp when the corresponding request was created
+ * @param receivedTimeMs The unix timestamp when this response was received
+ * @param disconnected Whether the client disconnected before fully reading a response
+ * @param versionMismatch Whether there was a version mismatch that prevented sending the request.
+ * @param responseBody The response contents (or null) if we disconnected, no response was expected,
+ * or if there was a version mismatch.
+ */
+ public ClientResponse(RequestHeader requestHeader,
+ RequestCompletionHandler callback,
+ String destination,
+ long createdTimeMs,
+ long receivedTimeMs,
+ boolean disconnected,
+ UnsupportedVersionException versionMismatch,
+ AuthenticationException authenticationException,
+ AbstractResponse responseBody) {
+ this(requestHeader,
+ callback,
+ destination,
+ createdTimeMs,
+ receivedTimeMs,
+ disconnected,
+ false,
+ versionMismatch,
+ authenticationException,
+ responseBody);
+ }
+
+ /**
+ * @param requestHeader The header of the corresponding request
+ * @param callback The callback to be invoked
* @param destination The node the corresponding request was sent to
+ * @param createdTimeMs The unix timestamp when the corresponding request was created
* @param receivedTimeMs The unix timestamp when this response was received
* @param disconnected Whether the client disconnected before fully reading a response
+ * @param timedOut Whether the client was disconnected because of a timeout; when setting this
+ * to <code>true</code>, <code>disconnected</code> must be <code>true</code>
+ * or an {@link IllegalStateException} will be thrown
* @param versionMismatch Whether there was a version mismatch that prevented sending the request.
* @param responseBody The response contents (or null) if we disconnected, no response was expected,
* or if there was a version mismatch.
@@ -54,15 +90,20 @@ public class ClientResponse {
long createdTimeMs,
long receivedTimeMs,
boolean disconnected,
+ boolean timedOut,
UnsupportedVersionException versionMismatch,
AuthenticationException authenticationException,
AbstractResponse responseBody) {
+ if (!disconnected && timedOut)
+ throw new IllegalStateException("The client response can't be in the state of connected, yet timed out");
+
this.requestHeader = requestHeader;
this.callback = callback;
this.destination = destination;
this.receivedTimeMs = receivedTimeMs;
this.latencyMs = receivedTimeMs - createdTimeMs;
this.disconnected = disconnected;
+ this.timedOut = timedOut;
this.versionMismatch = versionMismatch;
this.authenticationException = authenticationException;
this.responseBody = responseBody;
@@ -76,6 +117,10 @@ public class ClientResponse {
return disconnected;
}
+ public boolean wasTimedOut() {
+ return timedOut;
+ }
+
public UnsupportedVersionException versionMismatch() {
return versionMismatch;
}
@@ -116,6 +161,8 @@ public class ClientResponse {
latencyMs +
", disconnected=" +
disconnected +
+ ", timedOut=" +
+ timedOut +
", requestHeader=" +
requestHeader +
", responseBody=" +
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 81463d508a6..433440ce933 100644
--- a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
@@ -324,11 +324,14 @@ public class NetworkClient implements KafkaClient {
log.info("Client requested disconnect from node {}", nodeId);
selector.close(nodeId);
long now = time.milliseconds();
- cancelInFlightRequests(nodeId, now, abortedSends);
+ cancelInFlightRequests(nodeId, now, abortedSends, false);
connectionStates.disconnected(nodeId, now);
}
- private void cancelInFlightRequests(String nodeId, long now, Collection<ClientResponse> responses) {
+ private void cancelInFlightRequests(String nodeId,
+ long now,
+ Collection<ClientResponse> responses,
+ boolean timedOut) {
Iterable<InFlightRequest> inFlightRequests = this.inFlightRequests.clearAll(nodeId);
for (InFlightRequest request : inFlightRequests) {
if (log.isDebugEnabled()) {
@@ -346,8 +349,16 @@ public class NetworkClient implements KafkaClient {
}
if (!request.isInternalRequest) {
- if (responses != null)
- responses.add(request.disconnected(now, null));
+ if (responses != null) {
+ ClientResponse clientResponse;
+
+ if (timedOut)
+ clientResponse = request.timedOut(now);
+ else
+ clientResponse = request.disconnected(now);
+
+ responses.add(clientResponse);
+ }
} else if (request.header.apiKey() == ApiKeys.METADATA) {
metadataUpdater.handleFailedRequest(now, Optional.empty());
}
@@ -366,7 +377,7 @@ public class NetworkClient implements KafkaClient {
log.info("Client requested connection close from node {}", nodeId);
selector.close(nodeId);
long now = time.milliseconds();
- cancelInFlightRequests(nodeId, now, null);
+ cancelInFlightRequests(nodeId, now, null, false);
connectionStates.remove(nodeId);
}
@@ -754,6 +765,34 @@ public class NetworkClient implements KafkaClient {
String nodeId,
long now,
ChannelState disconnectState) {
+ processDisconnection(responses, nodeId, now, disconnectState, false);
+ }
+
+ /**
+ * Post process disconnection of a node
+ *
+ * @param responses The list of responses to update
+ * @param nodeId Id of the node to be disconnected
+ * @param now The current time
+ */
+ private void processTimeoutDisconnection(List<ClientResponse> responses, String nodeId, long now) {
+ processDisconnection(responses, nodeId, now, ChannelState.LOCAL_CLOSE, true);
+ }
+
+ /**
+ * Post process disconnection of a node
+ *
+ * @param responses The list of responses to update
+ * @param nodeId Id of the node to be disconnected
+ * @param now The current time
+ * @param disconnectState The state of the disconnected channel
+ * @param timedOut {@code true} if the connection is disconnected because of a timeout (request or connection)
+ */
+ private void processDisconnection(List<ClientResponse> responses,
+ String nodeId,
+ long now,
+ ChannelState disconnectState,
+ boolean timedOut) {
connectionStates.disconnected(nodeId, now);
apiVersions.remove(nodeId);
nodesNeedingApiVersionsFetch.remove(nodeId);
@@ -778,7 +817,7 @@ public class NetworkClient implements KafkaClient {
break; // Disconnections in other states are logged at debug level in Selector
}
- cancelInFlightRequests(nodeId, now, responses);
+ cancelInFlightRequests(nodeId, now, responses, timedOut);
metadataUpdater.handleServerDisconnect(now, nodeId, Optional.ofNullable(disconnectState.exception()));
}
@@ -795,7 +834,7 @@ public class NetworkClient implements KafkaClient {
// close connection to the node
this.selector.close(nodeId);
log.info("Disconnecting from node {} due to request timeout.", nodeId);
- processDisconnection(responses, nodeId, now, ChannelState.LOCAL_CLOSE);
+ processTimeoutDisconnection(responses, nodeId, now);
}
}
@@ -821,7 +860,7 @@ public class NetworkClient implements KafkaClient {
"The timeout value is {} ms.",
nodeId,
connectionStates.connectionSetupTimeoutMs(nodeId));
- processDisconnection(responses, nodeId, now, ChannelState.LOCAL_CLOSE);
+ processTimeoutDisconnection(responses, nodeId, now);
}
}
@@ -1279,9 +1318,15 @@ public class NetworkClient implements KafkaClient {
false, null, null, response);
}
- public ClientResponse disconnected(long timeMs, AuthenticationException authenticationException) {
+ public ClientResponse timedOut(long timeMs) {
+ // A timed out request is considered disconnected as well
+ return new ClientResponse(header, callback, destination, createdTimeMs, timeMs,
+ true, true, null, null, null);
+ }
+
+ public ClientResponse disconnected(long timeMs) {
return new ClientResponse(header, callback, destination, createdTimeMs, timeMs,
- true, null, authenticationException, null);
+ true, null, null, null);
}
@Override
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
index 55eb6c7be2f..ac19c6252b7 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
@@ -549,7 +549,13 @@ public class Sender implements Runnable {
private void handleProduceResponse(ClientResponse response, Map<TopicPartition, ProducerBatch> batches, long now) {
RequestHeader requestHeader = response.requestHeader();
int correlationId = requestHeader.correlationId();
- if (response.wasDisconnected()) {
+ if (response.wasTimedOut()) {
+ log.trace("Cancelled request with header {} due to the last request to node {} timed out",
+ requestHeader, response.destination());
+ for (ProducerBatch batch : batches.values())
+ completeBatch(batch, new ProduceResponse.PartitionResponse(Errors.REQUEST_TIMED_OUT, String.format("Disconnected from node %s due to timeout", response.destination())),
+ correlationId, now);
+ } else if (response.wasDisconnected()) {
log.trace("Cancelled request with header {} due to node {} being disconnected",
requestHeader, response.destination());
for (ProducerBatch batch : batches.values())
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 ab7235e112e..810448efc04 100644
--- a/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java
@@ -451,41 +451,74 @@ public class NetworkClientTest {
@Test
public void testRequestTimeout() {
- awaitReady(client, node); // has to be before creating any request, as it may send ApiVersionsRequest and its response is mocked with correlation id 0
- ProduceRequest.Builder builder = ProduceRequest.forCurrentMagic(new ProduceRequestData()
- .setTopicData(new ProduceRequestData.TopicProduceDataCollection())
- .setAcks((short) 1)
- .setTimeoutMs(1000));
- TestCallbackHandler handler = new TestCallbackHandler();
- int requestTimeoutMs = defaultRequestTimeoutMs + 5000;
- ClientRequest request = client.newClientRequest(node.idString(), builder, time.milliseconds(), true,
- requestTimeoutMs, handler);
- assertEquals(requestTimeoutMs, request.requestTimeoutMs());
- testRequestTimeout(request);
+ testRequestTimeout(defaultRequestTimeoutMs + 5000);
}
@Test
public void testDefaultRequestTimeout() {
+ testRequestTimeout(defaultRequestTimeoutMs);
+ }
+
+ /**
+ * This is a helper method that will execute two produce calls. The first call is expected to work and the
+ * second produce call is intentionally made to emulate a request timeout. In the case that a timeout occurrs
+ * during a request, we want to ensure that we {@link Metadata#requestUpdate() request a metadata update} so that
+ * on a subsequent invocation of {@link NetworkClient#poll(long, long) poll}, the metadata request will be sent.
+ *
+ * <p/>
+ *
+ * The {@link MetadataUpdater} has a specific method to handle
+ * {@link NetworkClient.DefaultMetadataUpdater#handleServerDisconnect(long, String, Optional) server disconnects}
+ * which is where we {@link Metadata#requestUpdate() request a metadata update}. This test helper method ensures
+ * that is invoked by checking {@link Metadata#updateRequested()} after the simulated timeout.
+ *
+ * @param requestTimeoutMs Timeout in ms
+ */
+ private void testRequestTimeout(int requestTimeoutMs) {
+ Metadata metadata = new Metadata(50, 5000, new LogContext(), new ClusterResourceListeners());
+ MetadataResponse metadataResponse = RequestTestUtils.metadataUpdateWith(2, Collections.emptyMap());
+ metadata.updateWithCurrentRequestVersion(metadataResponse, false, time.milliseconds());
+
+ NetworkClient client = createNetworkClientWithNoVersionDiscovery(metadata);
+
+ // Send first produce without any timeout.
+ ClientResponse clientResponse = produce(client, requestTimeoutMs, false);
+ assertEquals(node.idString(), clientResponse.destination());
+ assertFalse(clientResponse.wasDisconnected(), "Expected response to succeed and not disconnect");
+ assertFalse(clientResponse.wasTimedOut(), "Expected response to succeed and not time out");
+ assertFalse(metadata.updateRequested(), "Expected NetworkClient to not need to update metadata");
+
+ // Send second request, but emulate a timeout.
+ clientResponse = produce(client, requestTimeoutMs, true);
+ assertEquals(node.idString(), clientResponse.destination());
+ assertTrue(clientResponse.wasDisconnected(), "Expected response to fail due to disconnection");
+ assertTrue(clientResponse.wasTimedOut(), "Expected response to fail due to timeout");
+ assertTrue(metadata.updateRequested(), "Expected NetworkClient to have called requestUpdate on metadata on timeout");
+ }
+
+ private ClientResponse produce(NetworkClient client, int requestTimeoutMs, boolean shouldEmulateTimeout) {
awaitReady(client, node); // has to be before creating any request, as it may send ApiVersionsRequest and its response is mocked with correlation id 0
ProduceRequest.Builder builder = ProduceRequest.forCurrentMagic(new ProduceRequestData()
.setTopicData(new ProduceRequestData.TopicProduceDataCollection())
.setAcks((short) 1)
.setTimeoutMs(1000));
- ClientRequest request = client.newClientRequest(node.idString(), builder, time.milliseconds(), true);
- assertEquals(defaultRequestTimeoutMs, request.requestTimeoutMs());
- testRequestTimeout(request);
- }
-
- private void testRequestTimeout(ClientRequest request) {
+ TestCallbackHandler handler = new TestCallbackHandler();
+ ClientRequest request = client.newClientRequest(node.idString(), builder, time.milliseconds(), true,
+ requestTimeoutMs, handler);
client.send(request, time.milliseconds());
- time.sleep(request.requestTimeoutMs() + 1);
- List<ClientResponse> responses = client.poll(0, time.milliseconds());
+ if (shouldEmulateTimeout) {
+ // For a delay of slightly more than our timeout threshold to emulate the request timing out.
+ time.sleep(requestTimeoutMs + 1);
+ } else {
+ ProduceResponse produceResponse = new ProduceResponse(new ProduceResponseData());
+ ByteBuffer buffer = RequestTestUtils.serializeResponseWithHeader(produceResponse, PRODUCE.latestVersion(), request.correlationId());
+ selector.completeReceive(new NetworkReceive(node.idString(), buffer));
+ }
+ List<ClientResponse> responses = client.poll(0, time.milliseconds());
assertEquals(1, responses.size());
- ClientResponse clientResponse = responses.get(0);
- assertEquals(node.idString(), clientResponse.destination());
- assertTrue(clientResponse.wasDisconnected(), "Expected response to fail due to disconnection");
+ return responses.get(0);
}
@Test