You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/03/08 22:36:19 UTC
[GitHub] [kafka] cmccabe opened a new pull request #10281: KAFKA-12432: Fix AdminClient timeout handling in the presence of bad brokers
cmccabe opened a new pull request #10281:
URL: https://github.com/apache/kafka/pull/10281
Previously, we relied solely on the broker to enforce
request.timeout.ms. If the broker failed to enforce this timeout, the
admin client would hang until the call's API timeout hit, which would
return an error to the caller.
With this change, the Admin Client will time out the call after
request.timeout.ms even if the broker does not. This is useful because
if the broker is hanging for some reason, we don't want to use up all of
our default.api.timeout.ms. Since we would still prefer for the broker
to initiate the timeout, we arrange for the client request deadline to be
a few milliseconds after the broker request deadline.
Another issue that this PR fixes is that previously, the admin client
would not initiate a disconnect from a broker which was able to accept
connections, but which didn't allow us to send any requests.
With this PR, we handle this case by closing the connection and
putting the calls back into the list of calls which are pending a node
assignment.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10281: KAFKA-12432: AdminClient should time out nodes that are never ready
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10281:
URL: https://github.com/apache/kafka/pull/10281#discussion_r595632280
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -1089,29 +1106,61 @@ private long sendEligibleCalls(long now) {
continue;
}
Node node = entry.getKey();
+ if (!callsInFlight.getOrDefault(node.idString(), Collections.emptyList()).isEmpty()) {
+ log.trace("Still waiting for other calls to finish on node {}.", node);
+ nodeReadyDeadlines.remove(node);
+ continue;
+ }
if (!client.ready(node, now)) {
+ Long deadline = nodeReadyDeadlines.get(node);
+ if (deadline != null) {
+ if (now >= deadline) {
+ log.info("Disconnecting from {} and revoking {} node assignment(s) " +
+ "because the node is taking too long to become ready.",
+ node.idString(), calls.size());
+ transitionToPendingAndClearList(calls);
+ client.disconnect(node.idString());
+ nodeReadyDeadlines.remove(node);
+ iter.remove();
+ continue;
+ }
+ pollTimeout = Math.min(pollTimeout, deadline - now);
+ } else {
+ nodeReadyDeadlines.put(node, now + requestTimeoutMs);
+ }
long nodeTimeout = client.pollDelayMs(node, now);
pollTimeout = Math.min(pollTimeout, nodeTimeout);
log.trace("Client is not ready to send to {}. Must delay {} ms", node, nodeTimeout);
continue;
}
- Call call = calls.remove(0);
- int requestTimeoutMs = Math.min(KafkaAdminClient.this.requestTimeoutMs,
+ int remainingRequestTime;
+ Long deadlineMs = nodeReadyDeadlines.remove(node);
Review comment:
I added a comment
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on pull request #10281: KAFKA-12432: AdminClient should time out nodes that are never ready
Posted by GitBox <gi...@apache.org>.
cmccabe commented on pull request #10281:
URL: https://github.com/apache/kafka/pull/10281#issuecomment-805938039
For some reason the branch for this PR just stopped changing even when I pushed to it. Maybe it's a github bug?
I opened a different PR, https://github.com/apache/kafka/pull/10395 , that has the changes we discussed here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10281: KAFKA-12432: AdminClient should time out nodes that are never ready
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10281:
URL: https://github.com/apache/kafka/pull/10281#discussion_r592865485
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -1136,17 +1185,12 @@ private void timeoutCallsInFlight(TimeoutProcessor processor) {
// only one we need to check the timeout for.
Call call = contexts.get(0);
if (processor.callHasExpired(call)) {
- if (call.aborted) {
- log.warn("Aborted call {} is still in callsInFlight.", call);
- } else {
- log.debug("Closing connection to {} due to timeout while awaiting {}", nodeId, call);
- call.aborted = true;
- client.disconnect(nodeId);
- numTimedOut++;
- // We don't remove anything from the callsInFlight data structure. Because the connection
- // has been closed, the calls should be returned by the next client#poll(),
- // and handled at that point.
- }
+ log.debug("Disconnecting from {} due to timeout while awaiting {}", nodeId, call);
Review comment:
I wonder if we can raise this to info? This is a really important event to see in the logs, and we are always left guessing about it when the user does not have debug logging enabled. The frequency at the rate of the request timeout should keep it from being too spammy.
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -1089,29 +1106,61 @@ private long sendEligibleCalls(long now) {
continue;
}
Node node = entry.getKey();
+ if (!callsInFlight.getOrDefault(node.idString(), Collections.emptyList()).isEmpty()) {
+ log.trace("Still waiting for other calls to finish on node {}.", node);
+ nodeReadyDeadlines.remove(node);
+ continue;
Review comment:
Maybe I am missing something here. Why do we need to continue? Do we not support multiple in-flight requests? If not, then why does `callsInFlight` map to a list?
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -1089,29 +1106,61 @@ private long sendEligibleCalls(long now) {
continue;
}
Node node = entry.getKey();
+ if (!callsInFlight.getOrDefault(node.idString(), Collections.emptyList()).isEmpty()) {
+ log.trace("Still waiting for other calls to finish on node {}.", node);
+ nodeReadyDeadlines.remove(node);
+ continue;
+ }
if (!client.ready(node, now)) {
+ Long deadline = nodeReadyDeadlines.get(node);
+ if (deadline != null) {
+ if (now >= deadline) {
+ log.info("Disconnecting from {} and revoking {} node assignment(s) " +
+ "because the node is taking too long to become ready.",
+ node.idString(), calls.size());
+ transitionToPendingAndClearList(calls);
+ client.disconnect(node.idString());
+ nodeReadyDeadlines.remove(node);
+ iter.remove();
+ continue;
+ }
+ pollTimeout = Math.min(pollTimeout, deadline - now);
+ } else {
+ nodeReadyDeadlines.put(node, now + requestTimeoutMs);
Review comment:
It seems like it would be quite intuitive to use `socket.connection.setup.timeout.ms`. Is it because of the complexity of `socket.connection.setup.timeout.max.ms` that we don't do this? Basically we don't know how long the network client itself is planning to wait for the connection to be established.
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -1089,29 +1106,61 @@ private long sendEligibleCalls(long now) {
continue;
}
Node node = entry.getKey();
+ if (!callsInFlight.getOrDefault(node.idString(), Collections.emptyList()).isEmpty()) {
+ log.trace("Still waiting for other calls to finish on node {}.", node);
+ nodeReadyDeadlines.remove(node);
+ continue;
+ }
if (!client.ready(node, now)) {
+ Long deadline = nodeReadyDeadlines.get(node);
+ if (deadline != null) {
+ if (now >= deadline) {
+ log.info("Disconnecting from {} and revoking {} node assignment(s) " +
+ "because the node is taking too long to become ready.",
+ node.idString(), calls.size());
+ transitionToPendingAndClearList(calls);
+ client.disconnect(node.idString());
+ nodeReadyDeadlines.remove(node);
+ iter.remove();
+ continue;
+ }
+ pollTimeout = Math.min(pollTimeout, deadline - now);
+ } else {
+ nodeReadyDeadlines.put(node, now + requestTimeoutMs);
+ }
long nodeTimeout = client.pollDelayMs(node, now);
pollTimeout = Math.min(pollTimeout, nodeTimeout);
log.trace("Client is not ready to send to {}. Must delay {} ms", node, nodeTimeout);
continue;
}
- Call call = calls.remove(0);
- int requestTimeoutMs = Math.min(KafkaAdminClient.this.requestTimeoutMs,
+ int remainingRequestTime;
+ Long deadlineMs = nodeReadyDeadlines.remove(node);
Review comment:
So I guess the intent here is to count the "ready time" toward the request timeout. I think that's reasonable. It means that request timeout represents the maximum time we will commit to trying to get a response from a particular node before retrying. It might be worth a comment to explain this?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10281: KAFKA-12432: AdminClient should time out nodes that are never ready
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10281:
URL: https://github.com/apache/kafka/pull/10281#discussion_r597950036
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -1089,29 +1106,61 @@ private long sendEligibleCalls(long now) {
continue;
}
Node node = entry.getKey();
+ if (!callsInFlight.getOrDefault(node.idString(), Collections.emptyList()).isEmpty()) {
+ log.trace("Still waiting for other calls to finish on node {}.", node);
+ nodeReadyDeadlines.remove(node);
+ continue;
Review comment:
ok. I changed it to just `Map<String, Call>`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10281: KAFKA-12432: AdminClient should time out nodes that are never ready
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10281:
URL: https://github.com/apache/kafka/pull/10281#discussion_r596376827
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -1089,29 +1106,63 @@ private long sendEligibleCalls(long now) {
continue;
}
Node node = entry.getKey();
+ if (!callsInFlight.getOrDefault(node.idString(), Collections.emptyList()).isEmpty()) {
+ log.trace("Still waiting for other calls to finish on node {}.", node);
+ nodeReadyDeadlines.remove(node);
+ continue;
+ }
if (!client.ready(node, now)) {
+ Long deadline = nodeReadyDeadlines.get(node);
+ if (deadline != null) {
+ if (now >= deadline) {
+ log.info("Disconnecting from {} and revoking {} node assignment(s) " +
+ "because the node is taking too long to become ready.",
+ node.idString(), calls.size());
+ transitionToPendingAndClearList(calls);
+ client.disconnect(node.idString());
+ nodeReadyDeadlines.remove(node);
+ iter.remove();
+ continue;
+ }
+ pollTimeout = Math.min(pollTimeout, deadline - now);
Review comment:
Might be a good idea to use `Math.max(0, deadline - now)` since system time is non-monotonic.
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -1367,15 +1414,15 @@ private void processRequests() {
void enqueue(Call call, long now) {
if (call.tries > maxRetries) {
log.debug("Max retries {} for {} reached", maxRetries, call);
- call.fail(time.milliseconds(), new TimeoutException());
+ call.handleTimeoutFailure(time.milliseconds(), new TimeoutException());
Review comment:
While we're here, can we set the exception message to indicate that max retries had been reached?
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -1089,29 +1106,61 @@ private long sendEligibleCalls(long now) {
continue;
}
Node node = entry.getKey();
+ if (!callsInFlight.getOrDefault(node.idString(), Collections.emptyList()).isEmpty()) {
+ log.trace("Still waiting for other calls to finish on node {}.", node);
+ nodeReadyDeadlines.remove(node);
+ continue;
Review comment:
Ok. I think we might as well simplify the type since we already have logic which assumes one inflight request. It doesn't feel like we're buying ourselves much in terms of future-proofing.
##########
File path: clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
##########
@@ -5307,6 +5308,71 @@ public void testUnregisterBrokerTimeoutMaxWait() {
}
}
+ /**
+ * Test that if the client can obtain a node assignment, but can't send to the given
+ * node, it will disconnect and try a different node.
+ */
+ @Test
+ public void testClientSideTimeoutAfterFailureToSend() throws Exception {
+ Cluster cluster = mockCluster(3, 0);
+ CompletableFuture<String> disconnectFuture = new CompletableFuture<>();
+ MockTime time = new MockTime();
+ try (final AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(time, cluster,
+ newStrMap(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "1",
+ AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "100000",
+ AdminClientConfig.RETRY_BACKOFF_MS_CONFIG, "1"))) {
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+ for (Node node : cluster.nodes()) {
+ env.kafkaClient().delayReady(node, 100);
+ }
+ env.kafkaClient().setDisconnectFuture(disconnectFuture);
+ final ListTopicsResult result = env.adminClient().listTopics();
+ env.kafkaClient().prepareResponse(prepareMetadataResponse(cluster, Errors.NONE));
+ log.debug("Advancing clock by 25 ms to trigger client-side disconnect.");
+ time.sleep(25);
+ disconnectFuture.get();
+ log.debug("Enabling nodes to send requests again.");
+ for (Node node : cluster.nodes()) {
+ env.kafkaClient().delayReady(node, 0);
+ }
+ time.sleep(5);
+ log.info("Waiting for result.");
+ assertEquals(0, result.listings().get().size());
+ }
+ }
+
+ /**
+ * Test that if the client can send to a node, but doesn't receive a response, it will
+ * disconnect and try a different node.
+ */
+ @Test
+ public void testClientSideTimeoutAfterFailureToReceiveResponse() throws Exception {
+ Cluster cluster = mockCluster(3, 0);
+ CompletableFuture<String> disconnectFuture = new CompletableFuture<>();
+ MockTime time = new MockTime();
+ try (final AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(time, cluster,
+ newStrMap(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "1",
+ AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "100000",
+ AdminClientConfig.RETRY_BACKOFF_MS_CONFIG, "1"))) {
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+ env.kafkaClient().setDisconnectFuture(disconnectFuture);
+ final ListTopicsResult result = env.adminClient().listTopics();
+ while (true) {
+ time.sleep(1);
+ try {
+ disconnectFuture.get(1, TimeUnit.MICROSECONDS);
Review comment:
Might be simpler to check `isDone`.
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -1089,29 +1106,63 @@ private long sendEligibleCalls(long now) {
continue;
}
Node node = entry.getKey();
+ if (!callsInFlight.getOrDefault(node.idString(), Collections.emptyList()).isEmpty()) {
+ log.trace("Still waiting for other calls to finish on node {}.", node);
+ nodeReadyDeadlines.remove(node);
+ continue;
+ }
if (!client.ready(node, now)) {
+ Long deadline = nodeReadyDeadlines.get(node);
+ if (deadline != null) {
+ if (now >= deadline) {
+ log.info("Disconnecting from {} and revoking {} node assignment(s) " +
+ "because the node is taking too long to become ready.",
+ node.idString(), calls.size());
+ transitionToPendingAndClearList(calls);
+ client.disconnect(node.idString());
+ nodeReadyDeadlines.remove(node);
+ iter.remove();
+ continue;
+ }
+ pollTimeout = Math.min(pollTimeout, deadline - now);
+ } else {
+ nodeReadyDeadlines.put(node, now + requestTimeoutMs);
+ }
long nodeTimeout = client.pollDelayMs(node, now);
pollTimeout = Math.min(pollTimeout, nodeTimeout);
log.trace("Client is not ready to send to {}. Must delay {} ms", node, nodeTimeout);
continue;
}
- Call call = calls.remove(0);
- int requestTimeoutMs = Math.min(KafkaAdminClient.this.requestTimeoutMs,
+ // Subtract the time we spent waiting for the node to become ready from
+ // the total request time.
+ int remainingRequestTime;
+ Long deadlineMs = nodeReadyDeadlines.remove(node);
+ if (deadlineMs == null) {
+ remainingRequestTime = requestTimeoutMs;
+ } else {
+ remainingRequestTime = calcTimeoutMsRemainingAsInt(now, deadlineMs);
+ }
+ while (!calls.isEmpty()) {
+ Call call = calls.remove(0);
+ int timeoutMs = Math.min(remainingRequestTime,
calcTimeoutMsRemainingAsInt(now, call.deadlineMs));
- AbstractRequest.Builder<?> requestBuilder;
- try {
- requestBuilder = call.createRequest(requestTimeoutMs);
- } catch (Throwable throwable) {
- call.fail(now, new KafkaException(String.format(
- "Internal error sending %s to %s.", call.callName, node)));
- continue;
+ AbstractRequest.Builder<?> requestBuilder;
+ try {
+ requestBuilder = call.createRequest(timeoutMs);
+ } catch (Throwable throwable) {
+ call.fail(now, new KafkaException(String.format(
Review comment:
I noticed that we lose the throwable that we caught here. Can we add it as a cause?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10281: KAFKA-12432: AdminClient should time out nodes that are never ready
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10281:
URL: https://github.com/apache/kafka/pull/10281#discussion_r595631316
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -1136,17 +1185,12 @@ private void timeoutCallsInFlight(TimeoutProcessor processor) {
// only one we need to check the timeout for.
Call call = contexts.get(0);
if (processor.callHasExpired(call)) {
- if (call.aborted) {
- log.warn("Aborted call {} is still in callsInFlight.", call);
- } else {
- log.debug("Closing connection to {} due to timeout while awaiting {}", nodeId, call);
- call.aborted = true;
- client.disconnect(nodeId);
- numTimedOut++;
- // We don't remove anything from the callsInFlight data structure. Because the connection
- // has been closed, the calls should be returned by the next client#poll(),
- // and handled at that point.
- }
+ log.debug("Disconnecting from {} due to timeout while awaiting {}", nodeId, call);
Review comment:
OK, let's raise it to `INFO`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10281: KAFKA-12432: AdminClient should time out nodes that are never ready
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10281:
URL: https://github.com/apache/kafka/pull/10281#discussion_r597950877
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -1089,29 +1106,63 @@ private long sendEligibleCalls(long now) {
continue;
}
Node node = entry.getKey();
+ if (!callsInFlight.getOrDefault(node.idString(), Collections.emptyList()).isEmpty()) {
+ log.trace("Still waiting for other calls to finish on node {}.", node);
+ nodeReadyDeadlines.remove(node);
+ continue;
+ }
if (!client.ready(node, now)) {
+ Long deadline = nodeReadyDeadlines.get(node);
+ if (deadline != null) {
+ if (now >= deadline) {
+ log.info("Disconnecting from {} and revoking {} node assignment(s) " +
+ "because the node is taking too long to become ready.",
+ node.idString(), calls.size());
+ transitionToPendingAndClearList(calls);
+ client.disconnect(node.idString());
+ nodeReadyDeadlines.remove(node);
+ iter.remove();
+ continue;
+ }
+ pollTimeout = Math.min(pollTimeout, deadline - now);
Review comment:
Let's just do that in the call to `NetworkClient#poll`. In general I agree that monotonic time would have been better (maybe this needs a rewrite...)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10281: KAFKA-12432: AdminClient should time out nodes that are never ready
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10281:
URL: https://github.com/apache/kafka/pull/10281#discussion_r595633992
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -1089,29 +1106,61 @@ private long sendEligibleCalls(long now) {
continue;
}
Node node = entry.getKey();
+ if (!callsInFlight.getOrDefault(node.idString(), Collections.emptyList()).isEmpty()) {
+ log.trace("Still waiting for other calls to finish on node {}.", node);
+ nodeReadyDeadlines.remove(node);
+ continue;
Review comment:
Oh, and on the question of why inflight requests has a list, it was done for future-proofing
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe merged pull request #10281: KAFKA-12432: AdminClient should time out nodes that are never ready
Posted by GitBox <gi...@apache.org>.
cmccabe merged pull request #10281:
URL: https://github.com/apache/kafka/pull/10281
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on pull request #10281: KAFKA-12432: Fix AdminClient timeout handling in the presence of bad brokers
Posted by GitBox <gi...@apache.org>.
hachikuji commented on pull request #10281:
URL: https://github.com/apache/kafka/pull/10281#issuecomment-794717132
> Previously, we relied solely on the broker to enforce
request.timeout.ms. If the broker failed to enforce this timeout, the
admin client would hang until the call's API timeout hit, which would
return an error to the caller.
I don't think this is accurate. The `NetworkClient` enforces the request timeout, which is passed through the `ClientRequest` object. If the response is not received before the timeout expires, then `NetworkClient` will disconnect.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10281: KAFKA-12432: AdminClient should time out nodes that are never ready
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10281:
URL: https://github.com/apache/kafka/pull/10281#discussion_r595633668
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -1089,29 +1106,61 @@ private long sendEligibleCalls(long now) {
continue;
}
Node node = entry.getKey();
+ if (!callsInFlight.getOrDefault(node.idString(), Collections.emptyList()).isEmpty()) {
+ log.trace("Still waiting for other calls to finish on node {}.", node);
+ nodeReadyDeadlines.remove(node);
+ continue;
Review comment:
Notice that we set `maxInFlightRequestsPerConnection` to 1 when constructing the `NetworkClient`. We don't support sending multiple requests to a single node on a single connection in `AdminClient`. I think we could add this support, but we'd have to check how the server handled it since we've never done it before. Maybe there should be a JIRA.
Also, if we do choose to add this support for multiple outstanding requests per node per socket we'd need some way to distinguish between "waiting for a chance to use this connection" from "waiting for this connection to be opened" in NetworkClient. Currently ready just returns a boolean, which isn't enough information to distinguish these two cases. We could probably add a new method that returned an enum or something.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10281: KAFKA-12432: AdminClient should time out nodes that are never ready
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10281:
URL: https://github.com/apache/kafka/pull/10281#discussion_r595631870
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -1089,29 +1106,61 @@ private long sendEligibleCalls(long now) {
continue;
}
Node node = entry.getKey();
+ if (!callsInFlight.getOrDefault(node.idString(), Collections.emptyList()).isEmpty()) {
+ log.trace("Still waiting for other calls to finish on node {}.", node);
+ nodeReadyDeadlines.remove(node);
+ continue;
+ }
if (!client.ready(node, now)) {
+ Long deadline = nodeReadyDeadlines.get(node);
+ if (deadline != null) {
+ if (now >= deadline) {
+ log.info("Disconnecting from {} and revoking {} node assignment(s) " +
+ "because the node is taking too long to become ready.",
+ node.idString(), calls.size());
+ transitionToPendingAndClearList(calls);
+ client.disconnect(node.idString());
+ nodeReadyDeadlines.remove(node);
+ iter.remove();
+ continue;
+ }
+ pollTimeout = Math.min(pollTimeout, deadline - now);
+ } else {
+ nodeReadyDeadlines.put(node, now + requestTimeoutMs);
Review comment:
The complexity of the min / max issue is one thing. Another thing is that we don't know when the connection has been established, and when it has not. NetworkClient doesn't expose this information. `NetworkClient#ready` may return false for a variety of reasons, many of which are not indicative of connection establishment.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10281: KAFKA-12432: AdminClient should time out nodes that are never ready
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10281:
URL: https://github.com/apache/kafka/pull/10281#discussion_r597952673
##########
File path: clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
##########
@@ -5307,6 +5308,71 @@ public void testUnregisterBrokerTimeoutMaxWait() {
}
}
+ /**
+ * Test that if the client can obtain a node assignment, but can't send to the given
+ * node, it will disconnect and try a different node.
+ */
+ @Test
+ public void testClientSideTimeoutAfterFailureToSend() throws Exception {
+ Cluster cluster = mockCluster(3, 0);
+ CompletableFuture<String> disconnectFuture = new CompletableFuture<>();
+ MockTime time = new MockTime();
+ try (final AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(time, cluster,
+ newStrMap(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "1",
+ AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "100000",
+ AdminClientConfig.RETRY_BACKOFF_MS_CONFIG, "1"))) {
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+ for (Node node : cluster.nodes()) {
+ env.kafkaClient().delayReady(node, 100);
+ }
+ env.kafkaClient().setDisconnectFuture(disconnectFuture);
+ final ListTopicsResult result = env.adminClient().listTopics();
+ env.kafkaClient().prepareResponse(prepareMetadataResponse(cluster, Errors.NONE));
+ log.debug("Advancing clock by 25 ms to trigger client-side disconnect.");
+ time.sleep(25);
+ disconnectFuture.get();
+ log.debug("Enabling nodes to send requests again.");
+ for (Node node : cluster.nodes()) {
+ env.kafkaClient().delayReady(node, 0);
+ }
+ time.sleep(5);
+ log.info("Waiting for result.");
+ assertEquals(0, result.listings().get().size());
+ }
+ }
+
+ /**
+ * Test that if the client can send to a node, but doesn't receive a response, it will
+ * disconnect and try a different node.
+ */
+ @Test
+ public void testClientSideTimeoutAfterFailureToReceiveResponse() throws Exception {
+ Cluster cluster = mockCluster(3, 0);
+ CompletableFuture<String> disconnectFuture = new CompletableFuture<>();
+ MockTime time = new MockTime();
+ try (final AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(time, cluster,
+ newStrMap(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "1",
+ AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "100000",
+ AdminClientConfig.RETRY_BACKOFF_MS_CONFIG, "1"))) {
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+ env.kafkaClient().setDisconnectFuture(disconnectFuture);
+ final ListTopicsResult result = env.adminClient().listTopics();
+ while (true) {
+ time.sleep(1);
+ try {
+ disconnectFuture.get(1, TimeUnit.MICROSECONDS);
Review comment:
In this case, `isDone` is not supposed to be true until we advance the clock (so that a new node can be tried, after the previous one disconnected)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe removed a comment on pull request #10281: KAFKA-12432: AdminClient should time out nodes that are never ready
Posted by GitBox <gi...@apache.org>.
cmccabe removed a comment on pull request #10281:
URL: https://github.com/apache/kafka/pull/10281#issuecomment-805938039
For some reason the branch for this PR just stopped changing even when I pushed to it. Maybe it's a github bug?
I opened a different PR, https://github.com/apache/kafka/pull/10395 , that has the changes we discussed here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on pull request #10281: KAFKA-12432: AdminClient should time out nodes that are never ready
Posted by GitBox <gi...@apache.org>.
cmccabe commented on pull request #10281:
URL: https://github.com/apache/kafka/pull/10281#issuecomment-796936178
> I don't think this is accurate. The NetworkClient enforces the request timeout, which is passed through the ClientRequest object. If the response is not received before the timeout expires, then NetworkClient will disconnect.
Thanks for the correction. This part of the change description was inaccurate-- I have removed it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org