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 0ff30e93515eb..446bf44010fc5 100644 --- a/clients/src/main/java/org/apache/kafka/clients/ClientResponse.java +++ b/clients/src/main/java/org/apache/kafka/clients/ClientResponse.java @@ -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; @@ -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; /** @@ -53,6 +55,7 @@ public ClientResponse(RequestHeader requestHeader, long receivedTimeMs, boolean disconnected, UnsupportedVersionException versionMismatch, + AuthenticationException authenticationException, AbstractResponse responseBody) { this.requestHeader = requestHeader; this.callback = callback; @@ -61,6 +64,7 @@ public ClientResponse(RequestHeader requestHeader, this.latencyMs = receivedTimeMs - createdTimeMs; this.disconnected = disconnected; this.versionMismatch = versionMismatch; + this.authenticationException = authenticationException; this.responseBody = responseBody; } @@ -76,6 +80,10 @@ public UnsupportedVersionException versionMismatch() { return versionMismatch; } + public AuthenticationException authenticationException() { + return authenticationException; + } + public RequestHeader requestHeader() { return requestHeader; } 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 26fa467f81a50..7c87277eb8f0a 100644 --- a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java +++ b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java @@ -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); @@ -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); } } @@ -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 responses, String nodeId, long now, ChannelState disconnectState) { + private void processDisconnection(List responses, + String nodeId, + long now, + ChannelState disconnectState) { connectionStates.disconnected(nodeId, now); apiVersions.remove(nodeId); nodesNeedingApiVersionsFetch.remove(nodeId); @@ -641,7 +644,7 @@ private void processDisconnection(List 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); } @@ -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 diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java index 6b02e11e9bbc2..450de06fcd104 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java @@ -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; @@ -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 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 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; } /** @@ -992,26 +1004,25 @@ private void handleResponses(long now, List 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 pendingCallsToSend = new ArrayList<>(); + private void unassignUnsentCalls(Predicate shouldUnassign) { for (Iterator>> iter = callsToSend.entrySet().iterator(); iter.hasNext(); ) { Map.Entry> entry = iter.next(); - if (!disconnectedOnly || client.connectionFailed(entry.getKey())) { - for (Call call : entry.getValue()) { - pendingCallsToSend.add(call); - } + Node node = entry.getKey(); + List 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 calls) { @@ -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()) + pollTimeout = Math.min(pollTimeout, retryBackoffMs); + // Wait for network responses. log.trace("Entering KafkaClient#poll(timeout={})", pollTimeout); List 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; @@ -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 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 b718d63c54241..a9e167a9a2e7f 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 @@ -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)); } } } @@ -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()); @@ -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> unsent; 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 7c2fc0bbdb70d..2a1e21375b96d 100644 --- a/clients/src/test/java/org/apache/kafka/clients/MockClient.java +++ b/clients/src/test/java/org/apache/kafka/clients/MockClient.java @@ -78,7 +78,12 @@ public FutureResponse(Node node, private Cluster cluster; private Node node = null; private final Set ready = new HashSet<>(); - private final Map blackedOut = new HashMap<>(); + + // Nodes awaiting reconnect backoff, will not be chosen by leastLoadedNode + private final TransientSet blackedOut; + // Nodes which will always fail to connect, but can be chosen by leastLoadedNode + private final TransientSet unreachable; + private final Map pendingAuthenticationErrors = new HashMap<>(); private final Map authenticationErrors = new HashMap<>(); // Use concurrent queue for requests so that requests may be queried from a different thread @@ -98,6 +103,8 @@ public MockClient(Time time, Metadata metadata) { this.time = time; this.metadata = metadata; this.unavailableTopics = Collections.emptySet(); + this.blackedOut = new TransientSet<>(time); + this.unreachable = new TransientSet<>(time); } @Override @@ -107,19 +114,21 @@ public boolean isReady(Node node, long now) { @Override public boolean ready(Node node, long now) { - if (isBlackedOut(node)) + if (blackedOut.contains(node, now)) + return false; + + if (unreachable.contains(node, now)) { + blackout(node, 100); 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; + return blackedOut.expirationDelayMs(node, now); } @Override @@ -127,37 +136,29 @@ public long pollDelayMs(Node node, long now) { return connectionDelay(node, now); } - public void blackout(Node node, long duration) { - blackedOut.put(node, time.milliseconds() + duration); + public void blackout(Node node, long durationMs) { + blackedOut.add(node, durationMs); + } + + public void setUnreachable(Node node, long durationMs) { + disconnect(node.idString()); + unreachable.add(node, durationMs); } - public void authenticationFailed(Node node, long duration) { + public void authenticationFailed(Node node, long blackoutMs) { pendingAuthenticationErrors.remove(node); authenticationErrors.put(node, (AuthenticationException) Errors.SASL_AUTHENTICATION_FAILED.exception()); disconnect(node.idString()); - blackout(node, duration); + blackout(node, blackoutMs); } public void createPendingAuthenticationError(Node node, long blackoutMs) { pendingAuthenticationErrors.put(node, blackoutMs); } - private boolean isBlackedOut(Node node) { - if (blackedOut.containsKey(node)) { - long expiration = blackedOut.get(node); - if (time.milliseconds() > expiration) { - blackedOut.remove(node); - return false; - } else { - return true; - } - } - return false; - } - @Override public boolean connectionFailed(Node node) { - return isBlackedOut(node); + return blackedOut.contains(node); } @Override @@ -174,7 +175,7 @@ public void disconnect(String node) { if (request.destination().equals(node)) { short version = request.requestBuilder().latestAllowedVersion(); responses.add(new ClientResponse(request.makeHeader(version), request.callback(), request.destination(), - request.createdTimeMs(), now, true, null, null)); + request.createdTimeMs(), now, true, null, null, null)); iter.remove(); } } @@ -198,7 +199,8 @@ public void send(ClientRequest request, long now) { short version = nodeApiVersions.latestUsableVersion(request.apiKey(), builder.oldestAllowedVersion(), builder.latestAllowedVersion()); ClientResponse resp = new ClientResponse(request.makeHeader(version), request.callback(), request.destination(), - request.createdTimeMs(), time.milliseconds(), true, null, null); + request.createdTimeMs(), time.milliseconds(), true, null, + new AuthenticationException("Authentication failed"), null); responses.add(resp); return; } @@ -223,7 +225,7 @@ public void send(ClientRequest request, long now) { ClientResponse resp = new ClientResponse(request.makeHeader(version), request.callback(), request.destination(), request.createdTimeMs(), time.milliseconds(), futureResp.disconnected, - unsupportedVersionException, futureResp.responseBody); + unsupportedVersionException, null, futureResp.responseBody); responses.add(resp); iterator.remove(); return; @@ -320,7 +322,7 @@ public void respondToRequest(ClientRequest clientRequest, AbstractResponse respo requests.remove(clientRequest); short version = clientRequest.requestBuilder().latestAllowedVersion(); responses.add(new ClientResponse(clientRequest.makeHeader(version), clientRequest.callback(), clientRequest.destination(), - clientRequest.createdTimeMs(), time.milliseconds(), false, null, response)); + clientRequest.createdTimeMs(), time.milliseconds(), false, null, null, response)); } @@ -328,7 +330,7 @@ public void respond(AbstractResponse response, boolean disconnected) { ClientRequest request = requests.remove(); short version = request.requestBuilder().latestAllowedVersion(); responses.add(new ClientResponse(request.makeHeader(version), request.callback(), request.destination(), - request.createdTimeMs(), time.milliseconds(), disconnected, null, response)); + request.createdTimeMs(), time.milliseconds(), disconnected, null, null, response)); } public void respondFrom(AbstractResponse response, Node node) { @@ -343,7 +345,7 @@ public void respondFrom(AbstractResponse response, Node node, boolean disconnect iterator.remove(); short version = request.requestBuilder().latestAllowedVersion(); responses.add(new ClientResponse(request.makeHeader(version), request.callback(), request.destination(), - request.createdTimeMs(), time.milliseconds(), disconnected, null, response)); + request.createdTimeMs(), time.milliseconds(), disconnected, null, null, response)); return; } } @@ -420,6 +422,7 @@ public boolean conditionMet() { public void reset() { ready.clear(); blackedOut.clear(); + unreachable.clear(); requests.clear(); responses.clear(); futureResponses.clear(); @@ -511,6 +514,9 @@ public void close(String node) { @Override public Node leastLoadedNode(long now) { + // Consistent with NetworkClient, we do not return nodes awaiting reconnect backoff + if (blackedOut.contains(node, now)) + return null; return this.node; } @@ -539,4 +545,43 @@ private static class MetadataUpdate { } } + private static class TransientSet { + // The elements in the set mapped to their expiration timestamps + private final Map elements = new HashMap<>(); + private final Time time; + + private TransientSet(Time time) { + this.time = time; + } + + boolean contains(T element) { + return contains(element, time.milliseconds()); + } + + boolean contains(T element, long now) { + return expirationDelayMs(element, now) > 0; + } + + void add(T element, long durationMs) { + elements.put(element, time.milliseconds() + durationMs); + } + + long expirationDelayMs(T element, long now) { + Long expirationTimeMs = elements.get(element); + if (expirationTimeMs == null) { + return 0; + } else if (now > expirationTimeMs) { + elements.remove(element); + return 0; + } else { + return expirationTimeMs - now; + } + } + + void clear() { + elements.clear(); + } + + } + } diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/AdminClientUnitTestEnv.java b/clients/src/test/java/org/apache/kafka/clients/admin/AdminClientUnitTestEnv.java index f862c140c0ddf..3cd807d93dca4 100644 --- a/clients/src/test/java/org/apache/kafka/clients/admin/AdminClientUnitTestEnv.java +++ b/clients/src/test/java/org/apache/kafka/clients/admin/AdminClientUnitTestEnv.java @@ -54,12 +54,16 @@ public AdminClientUnitTestEnv(Time time, Cluster cluster, Map co this(newMockClient(time, cluster), time, cluster, config); } + public AdminClientUnitTestEnv(MockClient mockClient, Time time, Cluster cluster) { + this(mockClient, time, cluster, newStrMap()); + } + private static MockClient newMockClient(Time time, Cluster cluster) { MockClient mockClient = new MockClient(time); mockClient.prepareResponse(new MetadataResponse(cluster.nodes(), cluster.clusterResource().clusterId(), - cluster.controller().id(), - Collections.emptyList())); + cluster.controller() == null ? MetadataResponse.NO_CONTROLLER_ID : cluster.controller().id(), + Collections.emptyList())); return mockClient; } diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java index 0bc786be59e6a..c4af2c7bcd6ac 100644 --- a/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java @@ -183,8 +183,8 @@ private static Cluster mockCluster(int controllerIndex) { nodes.put(1, new Node(1, "localhost", 8122)); nodes.put(2, new Node(2, "localhost", 8123)); return new Cluster("mockClusterId", nodes.values(), - Collections.emptySet(), Collections.emptySet(), - Collections.emptySet(), nodes.get(controllerIndex)); + Collections.emptySet(), Collections.emptySet(), + Collections.emptySet(), nodes.get(controllerIndex)); } private static AdminClientUnitTestEnv mockClientEnv(String... configVals) { @@ -192,7 +192,7 @@ private static AdminClientUnitTestEnv mockClientEnv(String... configVals) { } @Test - public void testCloseAdminClient() throws Exception { + public void testCloseAdminClient() { try (AdminClientUnitTestEnv env = mockClientEnv()) { } } @@ -232,6 +232,60 @@ public void testTimeoutWithoutMetadata() throws Exception { } } + @Test + public void testConnectionFailureOnMetadataUpdate() throws Exception { + // This tests the scenario in which we successfully connect to the bootstrap server, but + // the server disconnects before sending the full response + + Cluster cluster = Cluster.bootstrap(Collections.singletonList(new InetSocketAddress("localhost", 8121))); + MockClient mockClient = new MockClient(Time.SYSTEM); + mockClient.setNodeApiVersions(NodeApiVersions.create()); + mockClient.setNode(cluster.nodes().get(0)); + + try (final AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(mockClient, Time.SYSTEM, cluster)) { + Cluster discoveredCluster = mockCluster(0); + env.kafkaClient().prepareResponse(request -> request instanceof MetadataRequest, null, true); + env.kafkaClient().prepareResponse(body -> body instanceof MetadataRequest, + new MetadataResponse(discoveredCluster.nodes(), discoveredCluster.clusterResource().clusterId(), + 1, Collections.emptyList())); + env.kafkaClient().prepareResponse(body -> body instanceof CreateTopicsRequest, + new CreateTopicsResponse(Collections.singletonMap("myTopic", new ApiError(Errors.NONE, "")))); + + KafkaFuture future = env.adminClient().createTopics( + Collections.singleton(new NewTopic("myTopic", Collections.singletonMap(0, asList(0, 1, 2)))), + new CreateTopicsOptions().timeoutMs(10000)).all(); + + future.get(); + } + } + + @Test + public void testUnreachableBootstrapServer() throws Exception { + // This tests the scenario in which the bootstrap server is unreachable for a short while, + // which prevents AdminClient from being able to send the initial metadata request + + Cluster cluster = Cluster.bootstrap(Collections.singletonList(new InetSocketAddress("localhost", 8121))); + MockClient mockClient = new MockClient(Time.SYSTEM); + mockClient.setNodeApiVersions(NodeApiVersions.create()); + mockClient.setNode(cluster.nodes().get(0)); + mockClient.setUnreachable(cluster.nodes().get(0), 200); + + try (final AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(mockClient, Time.SYSTEM, cluster)) { + Cluster discoveredCluster = mockCluster(0); + env.kafkaClient().prepareResponse(body -> body instanceof MetadataRequest, + new MetadataResponse(discoveredCluster.nodes(), discoveredCluster.clusterResource().clusterId(), + 1, Collections.emptyList())); + env.kafkaClient().prepareResponse(body -> body instanceof CreateTopicsRequest, + new CreateTopicsResponse(Collections.singletonMap("myTopic", new ApiError(Errors.NONE, "")))); + + KafkaFuture future = env.adminClient().createTopics( + Collections.singleton(new NewTopic("myTopic", Collections.singletonMap(0, asList(0, 1, 2)))), + new CreateTopicsOptions().timeoutMs(10000)).all(); + + future.get(); + } + } + /** * Test that we propagate exceptions encountered when fetching metadata. */ diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java index 4886c6b218e57..4be688422ce77 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java @@ -333,7 +333,6 @@ private KafkaConsumer newConsumer(Properties props) { return new KafkaConsumer<>(props, new ByteArrayDeserializer(), new ByteArrayDeserializer()); } - @Test public void verifyHeartbeatSent() throws Exception { Time time = new MockTime(); @@ -1215,7 +1214,7 @@ public void testOffsetOfPausedPartitions() { Set partitions = Utils.mkSet(tp0, tp1); consumer.assign(partitions); // verify consumer's assignment - assertTrue(consumer.assignment().equals(partitions)); + assertEquals(partitions, consumer.assignment()); consumer.pause(partitions); consumer.seekToEnd(partitions); @@ -1491,8 +1490,52 @@ public boolean conditionMet() { } } - @Test - public void testConsumerWithinBlackoutPeriodAfterAuthenticationFailure() { + @Test(expected = AuthenticationException.class) + public void testPartitionsForAuthenticationFailure() { + final KafkaConsumer consumer = consumerWithPendingAuthentication(); + consumer.partitionsFor("some other topic"); + } + + @Test(expected = AuthenticationException.class) + public void testBeginningOffsetsAuthenticationFailure() { + final KafkaConsumer consumer = consumerWithPendingAuthentication(); + consumer.beginningOffsets(Collections.singleton(tp0)); + } + + @Test(expected = AuthenticationException.class) + public void testEndOffsetsAuthenticationFailure() { + final KafkaConsumer consumer = consumerWithPendingAuthentication(); + consumer.endOffsets(Collections.singleton(tp0)); + } + + @Test(expected = AuthenticationException.class) + public void testPollAuthenticationFailure() { + final KafkaConsumer consumer = consumerWithPendingAuthentication(); + consumer.subscribe(singleton(topic)); + consumer.poll(Duration.ZERO); + } + + @Test(expected = AuthenticationException.class) + public void testOffsetsForTimesAuthenticationFailure() { + final KafkaConsumer consumer = consumerWithPendingAuthentication(); + consumer.offsetsForTimes(singletonMap(tp0, 0L)); + } + + @Test(expected = AuthenticationException.class) + public void testCommitSyncAuthenticationFailure() { + final KafkaConsumer consumer = consumerWithPendingAuthentication(); + Map offsets = new HashMap<>(); + offsets.put(tp0, new OffsetAndMetadata(10L)); + consumer.commitSync(offsets); + } + + @Test(expected = AuthenticationException.class) + public void testCommittedAuthenticationFaiure() { + final KafkaConsumer consumer = consumerWithPendingAuthentication(); + consumer.committed(tp0); + } + + private KafkaConsumer consumerWithPendingAuthentication() { Time time = new MockTime(); Map tpCounts = new HashMap<>(); tpCounts.put(topic, 1); @@ -1500,70 +1543,14 @@ public void testConsumerWithinBlackoutPeriodAfterAuthenticationFailure() { Node node = cluster.nodes().get(0); Metadata metadata = createMetadata(); - metadata.update(cluster, Collections.emptySet(), time.milliseconds()); + metadata.update(cluster, Collections.emptySet(), time.milliseconds()); MockClient client = new MockClient(time, metadata); client.setNode(node); - client.authenticationFailed(node, 300); PartitionAssignor assignor = new RangeAssignor(); - final KafkaConsumer consumer = newConsumer(time, client, metadata, assignor, true); - consumer.subscribe(Collections.singleton(topic)); - callConsumerApisAndExpectAnAuthenticationError(consumer, tp0); - - time.sleep(30); // wait less than the blackout period - assertTrue(client.connectionFailed(node)); - callConsumerApisAndExpectAnAuthenticationError(consumer, tp0); - - client.requests().clear(); - consumer.close(0, TimeUnit.MILLISECONDS); - } - - private void callConsumerApisAndExpectAnAuthenticationError(KafkaConsumer consumer, TopicPartition partition) { - try { - consumer.partitionsFor("some other topic"); - fail("Expected an authentication error!"); - } catch (AuthenticationException e) { - // OK - } - - try { - consumer.beginningOffsets(Collections.singleton(partition)); - fail("Expected an authentication error!"); - } catch (AuthenticationException e) { - // OK - } - - try { - consumer.endOffsets(Collections.singleton(partition)); - fail("Expected an authentication error!"); - } catch (AuthenticationException e) { - // OK - } - - try { - consumer.poll(Duration.ZERO); - fail("Expected an authentication error!"); - } catch (AuthenticationException e) { - // OK - } - - Map offset = new HashMap<>(); - offset.put(partition, new OffsetAndMetadata(10L)); - - try { - consumer.commitSync(offset); - fail("Expected an authentication error!"); - } catch (AuthenticationException e) { - // OK - } - - try { - consumer.committed(partition); - fail("Expected an authentication error!"); - } catch (AuthenticationException e) { - // OK - } + client.createPendingAuthenticationError(node, 0); + return newConsumer(time, client, metadata, assignor, false); } private ConsumerRebalanceListener getConsumerRebalanceListener(final KafkaConsumer consumer) { @@ -1724,12 +1711,11 @@ private FetchResponse fetchResponse(Map( + Errors.NONE, 0, FetchResponse.INVALID_LAST_STABLE_OFFSET, + 0L, null, records)); } - return new FetchResponse(Errors.NONE, tpResponses, 0, INVALID_SESSION_ID); + return new FetchResponse<>(Errors.NONE, tpResponses, 0, INVALID_SESSION_ID); } private FetchResponse fetchResponse(TopicPartition partition, long fetchOffset, int count) { diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java index 135763d3f4169..32aae442388c5 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java @@ -521,20 +521,10 @@ public void testWakeupInOnJoinComplete() throws Exception { } @Test - public void testEnsureCoordinatorReadyWithinBlackoutPeriodAfterAuthenticationFailure() { + public void testAuthenticationErrorInEnsureCoordinatorReady() { setupCoordinator(RETRY_BACKOFF_MS); - mockClient.authenticationFailed(node, 300); - - try { - coordinator.ensureCoordinatorReady(Long.MAX_VALUE); - fail("Expected an authentication error."); - } catch (AuthenticationException e) { - // OK - } - - mockTime.sleep(30); // wait less than the blackout period - assertTrue(mockClient.connectionFailed(node)); + mockClient.createPendingAuthenticationError(node, 300); try { coordinator.ensureCoordinatorReady(Long.MAX_VALUE); diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java index 18288735805e3..32da34a166181 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java @@ -1572,18 +1572,8 @@ public void testNoCoordinatorDiscoveryIfPartitionAwaitingReset() { } @Test - public void testEnsureActiveGroupWithinBlackoutPeriodAfterAuthenticationFailure() { - client.authenticationFailed(node, 300); - - try { - coordinator.ensureActiveGroup(); - fail("Expected an authentication error."); - } catch (AuthenticationException e) { - // OK - } - - time.sleep(30); // wait less than the blackout period - assertTrue(client.connectionFailed(node)); + public void testAuthenticationFailureInEnsureActiveGroup() { + client.createPendingAuthenticationError(node, 300); try { coordinator.ensureActiveGroup(); diff --git a/core/src/main/scala/kafka/common/InterBrokerSendThread.scala b/core/src/main/scala/kafka/common/InterBrokerSendThread.scala index c65e557297405..60635f17073d2 100644 --- a/core/src/main/scala/kafka/common/InterBrokerSendThread.scala +++ b/core/src/main/scala/kafka/common/InterBrokerSendThread.scala @@ -22,6 +22,7 @@ import java.util.Map.Entry import kafka.utils.ShutdownableThread import org.apache.kafka.clients.{ClientRequest, ClientResponse, NetworkClient, RequestCompletionHandler} import org.apache.kafka.common.Node +import org.apache.kafka.common.errors.AuthenticationException import org.apache.kafka.common.internals.FatalExitError import org.apache.kafka.common.requests.AbstractRequest import org.apache.kafka.common.utils.Time @@ -106,9 +107,10 @@ abstract class InterBrokerSendThread(name: String, if (!requests.isEmpty && networkClient.connectionFailed(node)) { iterator.remove() for (request <- requests.asScala) { - if (networkClient.authenticationException(node) != null) + val authenticationException = networkClient.authenticationException(node) + if (authenticationException != null) error(s"Failed to send the following request due to authentication error: $request") - completeWithDisconnect(request, now) + completeWithDisconnect(request, now, authenticationException) } } } @@ -119,15 +121,17 @@ abstract class InterBrokerSendThread(name: String, val expiredRequests = unsentRequests.removeExpiredRequests(now, unsentExpiryMs) for (request <- expiredRequests.asScala) { debug(s"Failed to send the following request after $unsentExpiryMs ms: $request") - completeWithDisconnect(request, now) + completeWithDisconnect(request, now, null) } } - def completeWithDisconnect(request: ClientRequest, now: Long): Unit = { + def completeWithDisconnect(request: ClientRequest, + now: Long, + authenticationException: AuthenticationException): Unit = { val handler = request.callback handler.onComplete(new ClientResponse(request.makeHeader(request.requestBuilder().latestAllowedVersion()), handler, request.destination, now /* createdTimeMs */ , now /* receivedTimeMs */ , true /* disconnected */ , - null /* versionMismatch */ , null /* responseBody */)) + null /* versionMismatch */ , authenticationException, null)) } def wakeup(): Unit = networkClient.wakeup() diff --git a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorConcurrencyTest.scala b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorConcurrencyTest.scala index 046741afa1e54..6168077248ba4 100644 --- a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorConcurrencyTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorConcurrencyTest.scala @@ -181,7 +181,7 @@ class TransactionCoordinatorConcurrencyTest extends AbstractCoordinatorConcurren val request = requestAndHandler.request.asInstanceOf[WriteTxnMarkersRequest.Builder].build() val response = createResponse(request) requestAndHandler.handler.onComplete(new ClientResponse(new RequestHeader(ApiKeys.PRODUCE, 0, "client", 1), - null, null, 0, 0, false, null, response)) + null, null, 0, 0, false, null, null, response)) } } } diff --git a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMarkerChannelManagerTest.scala b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMarkerChannelManagerTest.scala index a039c53b3de40..454b3617328e6 100644 --- a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMarkerChannelManagerTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMarkerChannelManagerTest.scala @@ -302,7 +302,7 @@ class TransactionMarkerChannelManagerTest { val response = new WriteTxnMarkersResponse(createPidErrorMap(Errors.NONE)) for (requestAndHandler <- requestAndHandlers) { requestAndHandler.handler.onComplete(new ClientResponse(new RequestHeader(ApiKeys.PRODUCE, 0, "client", 1), - null, null, 0, 0, false, null, response)) + null, null, 0, 0, false, null, null, response)) } EasyMock.verify(txnStateManager) @@ -351,7 +351,7 @@ class TransactionMarkerChannelManagerTest { val response = new WriteTxnMarkersResponse(createPidErrorMap(Errors.NONE)) for (requestAndHandler <- requestAndHandlers) { requestAndHandler.handler.onComplete(new ClientResponse(new RequestHeader(ApiKeys.PRODUCE, 0, "client", 1), - null, null, 0, 0, false, null, response)) + null, null, 0, 0, false, null, null, response)) } EasyMock.verify(txnStateManager) @@ -406,7 +406,7 @@ class TransactionMarkerChannelManagerTest { val response = new WriteTxnMarkersResponse(createPidErrorMap(Errors.NONE)) for (requestAndHandler <- requestAndHandlers) { requestAndHandler.handler.onComplete(new ClientResponse(new RequestHeader(ApiKeys.PRODUCE, 0, "client", 1), - null, null, 0, 0, false, null, response)) + null, null, 0, 0, false, null, null, response)) } // call this again so that append log will be retried diff --git a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMarkerRequestCompletionHandlerTest.scala b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMarkerRequestCompletionHandlerTest.scala index 41ec15969b170..3ca6c1b00b495 100644 --- a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMarkerRequestCompletionHandlerTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMarkerRequestCompletionHandlerTest.scala @@ -72,7 +72,7 @@ class TransactionMarkerRequestCompletionHandlerTest { EasyMock.replay(markerChannelManager) handler.onComplete(new ClientResponse(new RequestHeader(ApiKeys.PRODUCE, 0, "client", 1), - null, null, 0, 0, true, null, null)) + null, null, 0, 0, true, null, null, null)) EasyMock.verify(markerChannelManager) } @@ -86,7 +86,7 @@ class TransactionMarkerRequestCompletionHandlerTest { try { handler.onComplete(new ClientResponse(new RequestHeader(ApiKeys.PRODUCE, 0, "client", 1), - null, null, 0, 0, false, null, response)) + null, null, 0, 0, false, null, null, response)) fail("should have thrown illegal argument exception") } catch { case _: IllegalStateException => // ok @@ -204,7 +204,7 @@ class TransactionMarkerRequestCompletionHandlerTest { val response = new WriteTxnMarkersResponse(createProducerIdErrorMap(error)) handler.onComplete(new ClientResponse(new RequestHeader(ApiKeys.PRODUCE, 0, "client", 1), - null, null, 0, 0, false, null, response)) + null, null, 0, 0, false, null, null, response)) assertEquals(txnMetadata.topicPartitions, mutable.Set[TopicPartition](topicPartition)) EasyMock.verify(markerChannelManager) @@ -216,7 +216,7 @@ class TransactionMarkerRequestCompletionHandlerTest { val response = new WriteTxnMarkersResponse(createProducerIdErrorMap(error)) try { handler.onComplete(new ClientResponse(new RequestHeader(ApiKeys.PRODUCE, 0, "client", 1), - null, null, 0, 0, false, null, response)) + null, null, 0, 0, false, null, null, response)) fail("should have thrown illegal state exception") } catch { case _: IllegalStateException => // ok @@ -237,7 +237,7 @@ class TransactionMarkerRequestCompletionHandlerTest { val response = new WriteTxnMarkersResponse(createProducerIdErrorMap(error)) handler.onComplete(new ClientResponse(new RequestHeader(ApiKeys.PRODUCE, 0, "client", 1), - null, null, 0, 0, false, null, response)) + null, null, 0, 0, false, null, null, response)) assertTrue(txnMetadata.topicPartitions.isEmpty) assertTrue(completed) @@ -257,7 +257,7 @@ class TransactionMarkerRequestCompletionHandlerTest { val response = new WriteTxnMarkersResponse(createProducerIdErrorMap(error)) handler.onComplete(new ClientResponse(new RequestHeader(ApiKeys.PRODUCE, 0, "client", 1), - null, null, 0, 0, false, null, response)) + null, null, 0, 0, false, null, null, response)) assertTrue(removed) }