Skip to content

MINOR: Fix bug in AdminClient node reassignment following connection failure #5112

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rajinisivaram merged 6 commits into from
Jun 4, 2018
Merged

MINOR: Fix bug in AdminClient node reassignment following connection failure #5112

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
4a75c3e
MINOR: Fix bug in AdminClient connection failure detection
hachikuji May 31, 2018
c93ce30
Small cleanups
hachikuji Jun 1, 2018
e849eea
A bit of cleanup
hachikuji Jun 1, 2018
34afa9b
Fix retry backoff timeout check
hachikuji Jun 1, 2018
171be20
Fix inconsistent leastLoadedNode behavior in MockClient
hachikuji Jun 2, 2018
8873984
Fix compilation errors
hachikuji Jun 2, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions clients/src/main/java/org/apache/kafka/clients/ClientResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
*/
package org.apache.kafka.clients;

import org.apache.kafka.common.errors.AuthenticationException;
import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.requests.AbstractResponse;
import org.apache.kafka.common.requests.RequestHeader;
Expand All @@ -33,6 +34,7 @@ public class ClientResponse {
private final long latencyMs;
private final boolean disconnected;
private final UnsupportedVersionException versionMismatch;
private final AuthenticationException authenticationException;
private final AbstractResponse responseBody;

/**
Expand All @@ -53,6 +55,7 @@ public ClientResponse(RequestHeader requestHeader,
long receivedTimeMs,
boolean disconnected,
UnsupportedVersionException versionMismatch,
AuthenticationException authenticationException,
AbstractResponse responseBody) {
this.requestHeader = requestHeader;
this.callback = callback;
Expand All @@ -61,6 +64,7 @@ public ClientResponse(RequestHeader requestHeader,
this.latencyMs = receivedTimeMs - createdTimeMs;
this.disconnected = disconnected;
this.versionMismatch = versionMismatch;
this.authenticationException = authenticationException;
this.responseBody = responseBody;
}

Expand All @@ -76,6 +80,10 @@ public UnsupportedVersionException versionMismatch() {
return versionMismatch;
}

public AuthenticationException authenticationException() {
return authenticationException;
}

public RequestHeader requestHeader() {
return requestHeader;
}
Expand Down
23 changes: 14 additions & 9 deletions clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -260,7 +260,7 @@ public void disconnect(String nodeId) {
requestTypes.add(request.header.apiKey());
abortedSends.add(new ClientResponse(request.header,
request.callback, request.destination, request.createdTimeMs, now,
true, null, null));
true, null, null, null));
}
}
connectionStates.disconnected(nodeId, now);
Expand Down Expand Up @@ -413,14 +413,14 @@ private void doSend(ClientRequest clientRequest, boolean isInternalRequest, long
// The call to build may also throw UnsupportedVersionException, if there are essential
// fields that cannot be represented in the chosen version.
doSend(clientRequest, isInternalRequest, now, builder.build(version));
} catch (UnsupportedVersionException e) {
} catch (UnsupportedVersionException unsupportedVersionException) {
// If the version is not supported, skip sending the request over the wire.
// Instead, simply add it to the local queue of aborted requests.
log.debug("Version mismatch when attempting to send {} with correlation id {} to {}", builder,
clientRequest.correlationId(), clientRequest.destination(), e);
clientRequest.correlationId(), clientRequest.destination(), unsupportedVersionException);
ClientResponse clientResponse = new ClientResponse(clientRequest.makeHeader(builder.latestAllowedVersion()),
clientRequest.callback(), clientRequest.destination(), now, now,
false, e, null);
false, unsupportedVersionException, null, null);
abortedSends.add(clientResponse);
}
}
Expand Down Expand Up @@ -615,7 +615,10 @@ private static Struct parseStructMaybeUpdateThrottleTimeMetrics(ByteBuffer respo
* @param nodeId Id of the node to be disconnected
* @param now The current time
*/
private void processDisconnection(List<ClientResponse> responses, String nodeId, long now, ChannelState disconnectState) {
private void processDisconnection(List<ClientResponse> responses,
String nodeId,
long now,
ChannelState disconnectState) {
connectionStates.disconnected(nodeId, now);
apiVersions.remove(nodeId);
nodesNeedingApiVersionsFetch.remove(nodeId);
Expand All @@ -641,7 +644,7 @@ private void processDisconnection(List<ClientResponse> responses, String nodeId,
log.trace("Cancelled request {} {} with correlation id {} due to node {} being disconnected",
request.header.apiKey(), request.request, request.header.correlationId(), nodeId);
if (!request.isInternalRequest)
responses.add(request.disconnected(now));
responses.add(request.disconnected(now, disconnectState.exception()));
else if (request.header.apiKey() == ApiKeys.METADATA)
metadataUpdater.handleDisconnection(request.destination);
}
Expand Down Expand Up @@ -1034,11 +1037,13 @@ public InFlightRequest(RequestHeader header,
}

public ClientResponse completed(AbstractResponse response, long timeMs) {
return new ClientResponse(header, callback, destination, createdTimeMs, timeMs, false, null, response);
return new ClientResponse(header, callback, destination, createdTimeMs, timeMs,
false, null, null, response);
}

public ClientResponse disconnected(long timeMs) {
return new ClientResponse(header, callback, destination, createdTimeMs, timeMs, true, null, null);
public ClientResponse disconnected(long timeMs, AuthenticationException authenticationException) {
return new ClientResponse(header, callback, destination, createdTimeMs, timeMs,
true, null, authenticationException, null);
}

@Override
Expand Down
88 changes: 55 additions & 33 deletions clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -138,6 +138,7 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Predicate;

import static org.apache.kafka.common.utils.Utils.closeQuietly;

Expand Down Expand Up @@ -820,39 +821,50 @@ private synchronized void drainNewCalls() {
* Choose nodes for the calls in the pendingCalls list.
*
* @param now The current time in milliseconds.
* @param pendingIter An iterator yielding pending calls.
* @return The minimum time until a call is ready to be retried if any of the pending
* calls are backing off after a failure
*/
private long chooseNodesForPendingCalls(long now, Iterator<Call> pendingIter) {
private long maybeDrainPendingCalls(long now) {
long pollTimeout = Long.MAX_VALUE;
log.trace("Trying to choose nodes for {} at {}", pendingIter, now);
log.trace("Trying to choose nodes for {} at {}", pendingCalls, now);

Iterator<Call> pendingIter = pendingCalls.iterator();
while (pendingIter.hasNext()) {
Call call = pendingIter.next();

// If the call is being retried, await the proper backoff before finding the node
if (now < call.nextAllowedTryMs) {
pollTimeout = Math.min(pollTimeout, call.nextAllowedTryMs - now);
continue;
}

Node node = null;
try {
node = call.nodeProvider.provide();
} catch (Throwable t) {
// Handle authentication errors while choosing nodes.
log.debug("Unable to choose node for {}", call, t);
} else if (maybeDrainPendingCall(call, now)) {
pendingIter.remove();
call.fail(now, t);
}
}
return pollTimeout;
}

/**
* Check whether a pending call can be assigned a node. Return true if the pending call was either
* transferred to the callsToSend collection or if the call was failed. Return false if it
* should remain pending.
*/
private boolean maybeDrainPendingCall(Call call, long now) {
try {
Node node = call.nodeProvider.provide();
if (node != null) {
log.trace("Assigned {} to node {}", call, node);
pendingIter.remove();
call.curNode = node;
getOrCreateListValue(callsToSend, node).add(call);
return true;
} else {
log.trace("Unable to assign {} to a node.", call);
return false;
}
} catch (Throwable t) {
// Handle authentication errors while choosing nodes.
log.debug("Unable to choose node for {}", call, t);
call.fail(now, t);
return true;
}
return pollTimeout;
}

/**
Expand Down Expand Up @@ -992,26 +1004,25 @@ private void handleResponses(long now, List<ClientResponse> responses) {
}

/**
* Reassign calls that have not yet been sent. When metadata is refreshed,
* all unsent calls are reassigned to handle controller change and node changes.
* When a node is disconnected, all calls assigned to the node are reassigned.
* Unassign calls that have not yet been sent based on some predicate. For example, this
* is used to reassign the calls that have been assigned to a disconnected node.
*
* @param now The current time in milliseconds
* @param disconnectedOnly Reassign only calls to nodes that were disconnected
* in the last poll
* @param shouldUnassign Condition for reassignment. If the predicate is true, then the calls will
* be put back in the pendingCalls collection and they will be reassigned
*/
private void reassignUnsentCalls(long now, boolean disconnectedOnly) {
ArrayList<Call> pendingCallsToSend = new ArrayList<>();
private void unassignUnsentCalls(Predicate<Node> shouldUnassign) {
for (Iterator<Map.Entry<Node, List<Call>>> iter = callsToSend.entrySet().iterator(); iter.hasNext(); ) {
Map.Entry<Node, List<Call>> entry = iter.next();
if (!disconnectedOnly || client.connectionFailed(entry.getKey())) {
for (Call call : entry.getValue()) {
pendingCallsToSend.add(call);
}
Node node = entry.getKey();
List<Call> awaitingCalls = entry.getValue();

if (awaitingCalls.isEmpty()) {
iter.remove();
} else if (shouldUnassign.test(node)) {
pendingCalls.addAll(awaitingCalls);
iter.remove();
}
}
chooseNodesForPendingCalls(now, pendingCallsToSend.iterator());
}

private boolean hasActiveExternalCalls(Collection<Call> calls) {
Expand Down Expand Up @@ -1076,29 +1087,37 @@ public void run() {
}

// Choose nodes for our pending calls.
pollTimeout = Math.min(pollTimeout, chooseNodesForPendingCalls(now, pendingCalls.iterator()));
pollTimeout = Math.min(pollTimeout, maybeDrainPendingCalls(now));
long metadataFetchDelayMs = metadataManager.metadataFetchDelayMs(now);
if (metadataFetchDelayMs == 0) {
metadataManager.transitionToUpdatePending(now);
Call metadataCall = makeMetadataCall(now);
// Create a new metadata fetch call and add it to the end of pendingCalls.
// Assign a node for just the new call (we handled the other pending nodes above).
pendingCalls.add(metadataCall);
chooseNodesForPendingCalls(now, pendingCalls.listIterator(pendingCalls.size() - 1));

if (!maybeDrainPendingCall(metadataCall, now))
pendingCalls.add(metadataCall);
}
pollTimeout = Math.min(pollTimeout, sendEligibleCalls(now));

if (metadataFetchDelayMs > 0) {
pollTimeout = Math.min(pollTimeout, metadataFetchDelayMs);
}

// Ensure that we use a small poll timeout if there are pending calls which need to be sent
if (!pendingCalls.isEmpty())

@lindong28 lindong28 Jun 2, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not sure this optimization is useful. If there is pending calls, it means these calls can not find nodes to send to and we need to wait for the metadata update before trying again. We already have the logic for reducing pollTimeout based on metadataFetchDelayMs. And client.poll() will return immediately after metadata response is received even if we use large pollTimeout. Did I miss something here?

@hachikuji hachikuji Jun 2, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's less of an optimization and more of a safety net. The intent was to ensure we are not stuck in poll() with a long timeout while we have pending requests waiting to be sent. It may be unnecessary if we're convinced that the poll timeout is computed correctly.

@lindong28 lindong28 Jun 3, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see. It is reasonable to have a safety net.

pollTimeout = Math.min(pollTimeout, retryBackoffMs);

// Wait for network responses.
log.trace("Entering KafkaClient#poll(timeout={})", pollTimeout);
List<ClientResponse> responses = client.poll(pollTimeout, now);
log.trace("KafkaClient#poll retrieved {} response(s)", responses.size());

// unassign calls to disconnected nodes
unassignUnsentCalls(client::connectionFailed);

// Update the current time and handle the latest responses.
now = time.milliseconds();
reassignUnsentCalls(now, true); // reassign calls to disconnected nodes
handleResponses(now, responses);
}
int numTimedOut = 0;
Expand Down Expand Up @@ -1184,7 +1203,10 @@ public void handleResponse(AbstractResponse abstractResponse) {
MetadataResponse response = (MetadataResponse) abstractResponse;
long now = time.milliseconds();
metadataManager.update(response.cluster(), now);
reassignUnsentCalls(now, false);

// Unassign all unsent requests after a metadata refresh to allow for a new
// destination to be selected from the new metadata
unassignUnsentCalls(node -> true);
}

@Override
Expand Down
11 changes: 5 additions & 6 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -413,12 +413,9 @@ private void checkDisconnects(long now) {
for (ClientRequest request : requests) {
RequestFutureCompletionHandler handler = (RequestFutureCompletionHandler) request.callback();
AuthenticationException authenticationException = client.authenticationException(node);
if (authenticationException != null)
handler.onFailure(authenticationException);
else
handler.onComplete(new ClientResponse(request.makeHeader(request.requestBuilder().latestAllowedVersion()),
handler.onComplete(new ClientResponse(request.makeHeader(request.requestBuilder().latestAllowedVersion()),
request.callback(), request.destination(), request.createdTimeMs(), now, true,
null, null));
null, authenticationException, null));
}
}
}
Expand Down Expand Up @@ -571,6 +568,8 @@ private RequestFutureCompletionHandler() {
public void fireCompletion() {
if (e != null) {
future.raise(e);
} else if (response.authenticationException() != null) {
future.raise(response.authenticationException());
} else if (response.wasDisconnected()) {
log.debug("Cancelled request with header {} due to node {} being disconnected",
response.requestHeader(), response.destination());
Expand Down Expand Up @@ -611,7 +610,7 @@ public interface PollCondition {
}

/*
* A threadsafe helper class to hold requests per node that have not been sent yet
* A thread-safe helper class to hold requests per node that have not been sent yet
*/
private final static class UnsentRequests {
private final ConcurrentMap<Node, ConcurrentLinkedQueue<ClientRequest>> unsent;
Expand Down
Loading