[ https://issues.apache.org/jira/browse/KAFKA-6299?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16469537#comment-16469537 ]
ASF GitHub Bot commented on KAFKA-6299: --------------------------------------- hachikuji closed pull request #4295: KAFKA-6299. Fix AdminClient error handling when metadata changes URL: https://github.com/apache/kafka/pull/4295 This is a PR merged from a forked repository. As GitHub hides the original diff on merge, it is displayed below for the sake of provenance: As this is a foreign pull request (from a fork), the diff is supplied below (as it won't show otherwise due to GitHub magic): diff --git a/clients/src/main/java/org/apache/kafka/clients/MetadataUpdater.java b/clients/src/main/java/org/apache/kafka/clients/MetadataUpdater.java index 126728342d4..09ed995d14c 100644 --- a/clients/src/main/java/org/apache/kafka/clients/MetadataUpdater.java +++ b/clients/src/main/java/org/apache/kafka/clients/MetadataUpdater.java @@ -29,7 +29,7 @@ * <p> * This class is not thread-safe! */ -interface MetadataUpdater { +public interface MetadataUpdater { /** * Gets the current cluster info without blocking. 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 d8c0bad8066..70e9fbd6fbb 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 @@ -22,11 +22,11 @@ import org.apache.kafka.clients.ClientResponse; import org.apache.kafka.clients.ClientUtils; import org.apache.kafka.clients.KafkaClient; -import org.apache.kafka.clients.Metadata; import org.apache.kafka.clients.NetworkClient; import org.apache.kafka.clients.admin.DeleteAclsResult.FilterResult; import org.apache.kafka.clients.admin.DeleteAclsResult.FilterResults; import org.apache.kafka.clients.admin.DescribeReplicaLogDirsResult.ReplicaLogDirInfo; +import org.apache.kafka.clients.admin.internal.AdminMetadataManager; import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.clients.consumer.internals.ConsumerProtocol; import org.apache.kafka.clients.consumer.internals.PartitionAssignor; @@ -44,7 +44,6 @@ import org.apache.kafka.common.config.ConfigResource; import org.apache.kafka.common.errors.ApiException; import org.apache.kafka.common.errors.AuthenticationException; -import org.apache.kafka.common.errors.BrokerNotAvailableException; import org.apache.kafka.common.errors.DisconnectException; import org.apache.kafka.common.errors.InvalidGroupIdException; import org.apache.kafka.common.errors.InvalidRequestException; @@ -186,9 +185,9 @@ private final Time time; /** - * The cluster metadata used by the KafkaClient. + * The cluster metadata manager used by the KafkaClient. */ - private final Metadata metadata; + private final AdminMetadataManager metadataManager; /** * The metrics for this KafkaAdminClient. @@ -327,8 +326,9 @@ static KafkaAdminClient createInternal(AdminClientConfig config, TimeoutProcesso try { // Since we only request node information, it's safe to pass true for allowAutoTopicCreation (and it // simplifies communication with older brokers) - Metadata metadata = new Metadata(config.getLong(AdminClientConfig.RETRY_BACKOFF_MS_CONFIG), - config.getLong(AdminClientConfig.METADATA_MAX_AGE_CONFIG), true); + AdminMetadataManager metadataManager = new AdminMetadataManager(logContext, time, + config.getLong(AdminClientConfig.RETRY_BACKOFF_MS_CONFIG), + config.getLong(AdminClientConfig.METADATA_MAX_AGE_CONFIG)); List<MetricsReporter> reporters = config.getConfiguredInstances(AdminClientConfig.METRIC_REPORTER_CLASSES_CONFIG, MetricsReporter.class); Map<String, String> metricTags = Collections.singletonMap("client-id", clientId); @@ -344,7 +344,7 @@ static KafkaAdminClient createInternal(AdminClientConfig config, TimeoutProcesso metrics, time, metricGrpPrefix, channelBuilder, logContext); networkClient = new NetworkClient( selector, - metadata, + metadataManager.updater(), clientId, 1, config.getLong(AdminClientConfig.RECONNECT_BACKOFF_MS_CONFIG), @@ -356,7 +356,7 @@ static KafkaAdminClient createInternal(AdminClientConfig config, TimeoutProcesso true, apiVersions, logContext); - return new KafkaAdminClient(config, clientId, time, metadata, metrics, networkClient, + return new KafkaAdminClient(config, clientId, time, metadataManager, metrics, networkClient, timeoutProcessorFactory, logContext); } catch (Throwable exc) { closeQuietly(metrics, "Metrics"); @@ -367,35 +367,39 @@ static KafkaAdminClient createInternal(AdminClientConfig config, TimeoutProcesso } } - static KafkaAdminClient createInternal(AdminClientConfig config, KafkaClient client, Metadata metadata, Time time) { + static KafkaAdminClient createInternal(AdminClientConfig config, KafkaClient client, Time time) { Metrics metrics = null; String clientId = generateClientId(config); try { metrics = new Metrics(new MetricConfig(), new LinkedList<MetricsReporter>(), time); - return new KafkaAdminClient(config, clientId, time, metadata, metrics, client, null, - createLogContext(clientId)); + LogContext logContext = createLogContext(clientId); + AdminMetadataManager metadataManager = new AdminMetadataManager(logContext, time, + config.getLong(AdminClientConfig.RETRY_BACKOFF_MS_CONFIG), + config.getLong(AdminClientConfig.METADATA_MAX_AGE_CONFIG)); + return new KafkaAdminClient(config, clientId, time, metadataManager, metrics, + client, null, logContext); } catch (Throwable exc) { closeQuietly(metrics, "Metrics"); throw new KafkaException("Failed create new KafkaAdminClient", exc); } } - private static LogContext createLogContext(String clientId) { + static LogContext createLogContext(String clientId) { return new LogContext("[AdminClient clientId=" + clientId + "] "); } - private KafkaAdminClient(AdminClientConfig config, String clientId, Time time, Metadata metadata, - Metrics metrics, KafkaClient client, TimeoutProcessorFactory timeoutProcessorFactory, - LogContext logContext) { + private KafkaAdminClient(AdminClientConfig config, String clientId, Time time, + AdminMetadataManager metadataManager, Metrics metrics, KafkaClient client, + TimeoutProcessorFactory timeoutProcessorFactory, LogContext logContext) { this.defaultTimeoutMs = config.getInt(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG); this.clientId = clientId; this.log = logContext.logger(KafkaAdminClient.class); this.time = time; - this.metadata = metadata; + this.metadataManager = metadataManager; List<InetSocketAddress> addresses = ClientUtils.parseAndValidateAddresses( config.getList(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG)); - this.metadata.update(Cluster.bootstrap(addresses), Collections.<String>emptySet(), time.milliseconds()); + metadataManager.update(Cluster.bootstrap(addresses), time.milliseconds(), null); this.metrics = metrics; this.client = client; this.runnable = new AdminClientRunnable(); @@ -462,6 +466,13 @@ public void close(long duration, TimeUnit unit) { Node provide(); } + private class MetadataUpdateNodeIdProvider implements NodeProvider { + @Override + public Node provide() { + return client.leastLoadedNode(time.milliseconds()); + } + } + private class ConstantNodeIdProvider implements NodeProvider { private final int nodeId; @@ -471,7 +482,16 @@ public void close(long duration, TimeUnit unit) { @Override public Node provide() { - return metadata.fetch().nodeById(nodeId); + if (metadataManager.isReady() && + (metadataManager.nodeById(nodeId) != null)) { + return metadataManager.nodeById(nodeId); + } + // If we can't find the node with the given constant ID, we schedule a + // metadata update and hope it appears. This behavior is useful for avoiding + // flaky behavior in tests when the cluster is starting up and not all nodes + // have appeared. + metadataManager.requestUpdate(); + return null; } } @@ -481,7 +501,12 @@ public Node provide() { private class ControllerNodeProvider implements NodeProvider { @Override public Node provide() { - return metadata.fetch().controller(); + if (metadataManager.isReady() && + (metadataManager.controller() != null)) { + return metadataManager.controller(); + } + metadataManager.requestUpdate(); + return null; } } @@ -491,23 +516,40 @@ public Node provide() { private class LeastLoadedNodeProvider implements NodeProvider { @Override public Node provide() { - return client.leastLoadedNode(time.milliseconds()); + if (metadataManager.isReady()) { + // This may return null if all nodes are busy. + // In that case, we will postpone node assignment. + return client.leastLoadedNode(time.milliseconds()); + } + metadataManager.requestUpdate(); + return null; } } abstract class Call { + private final boolean internal; private final String callName; private final long deadlineMs; private final NodeProvider nodeProvider; private int tries = 0; private boolean aborted = false; + private Node curNode = null; - Call(String callName, long deadlineMs, NodeProvider nodeProvider) { + Call(boolean internal, String callName, long deadlineMs, NodeProvider nodeProvider) { + this.internal = internal; this.callName = callName; this.deadlineMs = deadlineMs; this.nodeProvider = nodeProvider; } + Call(String callName, long deadlineMs, NodeProvider nodeProvider) { + this(false, callName, deadlineMs, nodeProvider); + } + + protected Node curNode() { + return curNode; + } + /** * Handle a failure. * @@ -615,6 +657,10 @@ boolean handleUnsupportedVersionException(UnsupportedVersionException exception) public String toString() { return "Call(callName=" + callName + ", deadlineMs=" + deadlineMs + ")"; } + + public boolean isInternal() { + return internal; + } } static class TimeoutProcessorFactory { @@ -698,43 +744,15 @@ int nextTimeoutMs() { private List<Call> newCalls = new LinkedList<>(); /** - * Check if the AdminClient metadata is ready. - * We need to know who the controller is, and have a non-empty view of the cluster. - * - * @param prevMetadataVersion The previous metadata version which wasn't usable. - * @return null if the metadata is usable; the current metadata - * version otherwise - */ - private Integer checkMetadataReady(Integer prevMetadataVersion) { - if (prevMetadataVersion != null) { - if (prevMetadataVersion == metadata.version()) - return prevMetadataVersion; - } - Cluster cluster = metadata.fetch(); - if (cluster.nodes().isEmpty()) { - log.trace("Metadata is not ready yet. No cluster nodes found."); - return metadata.requestUpdate(); - } - if (cluster.controller() == null) { - log.trace("Metadata is not ready yet. No controller found."); - return metadata.requestUpdate(); - } - if (prevMetadataVersion != null) { - log.trace("Metadata is now ready."); - } - return null; - } - - /** - * Time out the elements in the newCalls list which are expired. + * Time out the elements in the pendingCalls list which are expired. * * @param processor The timeout processor. */ - private synchronized void timeoutNewCalls(TimeoutProcessor processor) { - int numTimedOut = processor.handleTimeouts(newCalls, + private void timeoutPendingCalls(TimeoutProcessor processor, List<Call> pendingCalls) { + int numTimedOut = processor.handleTimeouts(pendingCalls, "Timed out waiting for a node assignment."); if (numTimedOut > 0) - log.debug("Timed out {} new calls.", numTimedOut); + log.debug("Timed out {} pending calls.", numTimedOut); } /** @@ -743,7 +761,7 @@ private synchronized void timeoutNewCalls(TimeoutProcessor processor) { * @param processor The timeout processor. * @param callsToSend A map of nodes to the calls they need to handle. */ - private void timeoutCallsToSend(TimeoutProcessor processor, Map<Node, List<Call>> callsToSend) { + private int timeoutCallsToSend(TimeoutProcessor processor, Map<Node, List<Call>> callsToSend) { int numTimedOut = 0; for (List<Call> callList : callsToSend.values()) { numTimedOut += processor.handleTimeouts(callList, @@ -751,48 +769,52 @@ private void timeoutCallsToSend(TimeoutProcessor processor, Map<Node, List<Call> } if (numTimedOut > 0) log.debug("Timed out {} call(s) with assigned nodes.", numTimedOut); + return numTimedOut; } /** - * Choose nodes for the calls in the callsToSend list. + * Drain all the calls from newCalls into pendingCalls. * * This function holds the lock for the minimum amount of time, to avoid blocking * users of AdminClient who will also take the lock to add new calls. - * - * @param now The current time in milliseconds. - * @param callsToSend A map of nodes to the calls they need to handle. - * */ - private void chooseNodesForNewCalls(long now, Map<Node, List<Call>> callsToSend) { - List<Call> newCallsToAdd = null; - synchronized (this) { - if (newCalls.isEmpty()) { - return; - } - newCallsToAdd = newCalls; - newCalls = new LinkedList<>(); - } - for (Call call : newCallsToAdd) { - chooseNodeForNewCall(now, callsToSend, call); + private synchronized void drainNewCalls(ArrayList<Call> pendingCalls) { + if (!newCalls.isEmpty()) { + pendingCalls.addAll(newCalls); + newCalls.clear(); } } /** - * Choose a node for a new call. + * Choose nodes for the calls in the pendingCalls list. * * @param now The current time in milliseconds. + * @param pendingIter An iterator yielding pending calls. * @param callsToSend A map of nodes to the calls they need to handle. - * @param call The call. + * */ - private void chooseNodeForNewCall(long now, Map<Node, List<Call>> callsToSend, Call call) { - Node node = call.nodeProvider.provide(); - if (node == null) { - call.fail(now, new BrokerNotAvailableException( - String.format("Error choosing node for %s: no node found.", call.callName))); - return; + private void chooseNodesForPendingCalls(long now, Iterator<Call> pendingIter, + Map<Node, List<Call>> callsToSend) { + while (pendingIter.hasNext()) { + Call call = pendingIter.next(); + 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); + pendingIter.remove(); + call.fail(now, t); + } + if (node != null) { + log.trace("Assigned {} to node {}", call, node); + pendingIter.remove(); + call.curNode = node; + getOrCreateListValue(callsToSend, node).add(call); + } else { + log.trace("Unable to assign {} to a node.", call); + } } - log.trace("Assigned {} to {}", call, node); - getOrCreateListValue(callsToSend, node).add(call); } /** @@ -880,37 +902,6 @@ private void timeoutCallsInFlight(TimeoutProcessor processor, Map<String, List<C log.debug("Timed out {} call(s) in flight.", numTimedOut); } - /** - * If an authentication exception is encountered with connection to any broker, - * fail all pending requests. - */ - private void handleAuthenticationException(long now, Map<Node, List<Call>> callsToSend) { - AuthenticationException authenticationException = metadata.getAndClearAuthenticationException(); - if (authenticationException == null) { - for (Node node : callsToSend.keySet()) { - authenticationException = client.authenticationException(node); - if (authenticationException != null) - break; - } - } - if (authenticationException != null) { - synchronized (this) { - failCalls(now, newCalls, authenticationException); - } - for (List<Call> calls : callsToSend.values()) { - failCalls(now, calls, authenticationException); - } - callsToSend.clear(); - } - } - - private void failCalls(long now, List<Call> calls, AuthenticationException authenticationException) { - for (Call call : calls) { - call.fail(now, authenticationException); - } - calls.clear(); - } - /** * Handle responses from the server. * @@ -952,9 +943,14 @@ private void handleResponses(long now, if (response.versionMismatch() != null) { call.fail(now, response.versionMismatch()); } else if (response.wasDisconnected()) { - call.fail(now, new DisconnectException(String.format( - "Cancelled %s request with correlation id %s due to node %s being disconnected", - call.callName, correlationId, response.destination()))); + AuthenticationException authException = client.authenticationException(call.curNode()); + if (authException != null) { + call.fail(now, authException); + } else { + call.fail(now, new DisconnectException(String.format( + "Cancelled %s request with correlation id %s due to node %s being disconnected", + call.callName, correlationId, response.destination()))); + } } else { try { call.handleResponse(response.responseBody()); @@ -970,13 +966,41 @@ private void handleResponses(long now, } } - private synchronized boolean threadShouldExit(long now, long curHardShutdownTimeMs, + private boolean hasActiveExternalCalls(Collection<Call> calls) { + for (Call call : calls) { + if (!call.isInternal()) { + return true; + } + } + return false; + } + + /** + * Return true if there are currently active external calls. + */ + private boolean hasActiveExternalCalls(List<Call> pendingCalls, Map<Node, List<Call>> callsToSend, Map<Integer, Call> correlationIdToCalls) { - if (newCalls.isEmpty() && callsToSend.isEmpty() && correlationIdToCalls.isEmpty()) { + if (hasActiveExternalCalls(pendingCalls)) { + return true; + } + for (List<Call> callList : callsToSend.values()) { + if (hasActiveExternalCalls(callList)) { + return true; + } + } + if (hasActiveExternalCalls(correlationIdToCalls.values())) { + return true; + } + return false; + } + + private boolean threadShouldExit(long now, long curHardShutdownTimeMs, List<Call> pendingCalls, + Map<Node, List<Call>> callsToSend, Map<Integer, Call> correlationIdToCalls) { + if (!hasActiveExternalCalls(pendingCalls, callsToSend, correlationIdToCalls)) { log.trace("All work has been completed, and the I/O thread is now exiting."); return true; } - if (now > curHardShutdownTimeMs) { + if (now >= curHardShutdownTimeMs) { log.info("Forcing a hard I/O thread shutdown. Requests in progress will be aborted."); return true; } @@ -986,38 +1010,46 @@ private synchronized boolean threadShouldExit(long now, long curHardShutdownTime @Override public void run() { - /* + /** + * Calls which have not yet been assigned to a node. + * Only accessed from this thread. + */ + ArrayList<Call> pendingCalls = new ArrayList<>(); + + /** * Maps nodes to calls that we want to send. + * Only accessed from this thread. */ Map<Node, List<Call>> callsToSend = new HashMap<>(); - /* + /** * Maps node ID strings to calls that have been sent. + * Only accessed from this thread. */ Map<String, List<Call>> callsInFlight = new HashMap<>(); - /* + /** * Maps correlation IDs to calls that have been sent. + * Only accessed from this thread. */ Map<Integer, Call> correlationIdToCalls = new HashMap<>(); - /* - * The previous metadata version which wasn't usable, or null if there is none. - */ - Integer prevMetadataVersion = null; - long now = time.milliseconds(); log.trace("Thread starting"); while (true) { + // Copy newCalls into pendingCalls. + drainNewCalls(pendingCalls); + // Check if the AdminClient thread should shut down. long curHardShutdownTimeMs = hardShutdownTimeMs.get(); if ((curHardShutdownTimeMs != INVALID_SHUTDOWN_TIME) && - threadShouldExit(now, curHardShutdownTimeMs, callsToSend, correlationIdToCalls)) + threadShouldExit(now, curHardShutdownTimeMs, pendingCalls, + callsToSend, correlationIdToCalls)) break; // Handle timeouts. TimeoutProcessor timeoutProcessor = timeoutProcessorFactory.create(now); - timeoutNewCalls(timeoutProcessor); + timeoutPendingCalls(timeoutProcessor, pendingCalls); timeoutCallsToSend(timeoutProcessor, callsToSend); timeoutCallsInFlight(timeoutProcessor, callsInFlight); @@ -1026,12 +1058,22 @@ public void run() { pollTimeout = Math.min(pollTimeout, curHardShutdownTimeMs - now); } - // Handle new calls and metadata update requests. - prevMetadataVersion = checkMetadataReady(prevMetadataVersion); - if (prevMetadataVersion == null) { - chooseNodesForNewCalls(now, callsToSend); - pollTimeout = Math.min(pollTimeout, - sendEligibleCalls(now, callsToSend, correlationIdToCalls, callsInFlight)); + // Choose nodes for our pending calls. + chooseNodesForPendingCalls(now, pendingCalls.iterator(), callsToSend); + 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), + callsToSend); + } + pollTimeout = Math.min(pollTimeout, + sendEligibleCalls(now, callsToSend, correlationIdToCalls, callsInFlight)); + if (metadataFetchDelayMs > 0) { + pollTimeout = Math.min(pollTimeout, metadataFetchDelayMs); } // Wait for network responses. @@ -1041,7 +1083,6 @@ public void run() { // Update the current time and handle the latest responses. now = time.milliseconds(); - handleAuthenticationException(now, callsToSend); handleResponses(now, responses, callsInFlight, correlationIdToCalls); } int numTimedOut = 0; @@ -1051,10 +1092,13 @@ public void run() { "The AdminClient thread has exited."); newCalls = null; } + numTimedOut += timeoutProcessor.handleTimeouts(pendingCalls, + "The AdminClient thread has exited."); + numTimedOut += timeoutCallsToSend(timeoutProcessor, callsToSend); numTimedOut += timeoutProcessor.handleTimeouts(correlationIdToCalls.values(), "The AdminClient thread has exited."); if (numTimedOut > 0) { - log.debug("Timed out {} remaining operations.", numTimedOut); + log.debug("Timed out {} remaining operation(s).", numTimedOut); } closeQuietly(client, "KafkaClient"); closeQuietly(metrics, "Metrics"); @@ -1106,6 +1150,40 @@ void call(Call call, long now) { enqueue(call, now); } } + + /** + * Create a new metadata call. + */ + private Call makeMetadataCall(long now) { + return new Call(true, "fetchMetadata", calcDeadlineMs(now, defaultTimeoutMs), + new MetadataUpdateNodeIdProvider()) { + @Override + public AbstractRequest.Builder createRequest(int timeoutMs) { + // Since this only requests node information, it's safe to pass true + // for allowAutoTopicCreation (and it simplifies communication with + // older brokers) + return new MetadataRequest.Builder(Collections.<String>emptyList(), true); + } + + @Override + public void handleResponse(AbstractResponse abstractResponse) { + MetadataResponse response = (MetadataResponse) abstractResponse; + metadataManager.update(response.cluster(), time.milliseconds(), null); + } + + @Override + public void handleFailure(Throwable e) { + if (e instanceof AuthenticationException) { + log.info("Unable to fetch cluster metadata from node {} because of " + + "authentication error", curNode(), e); + metadataManager.update(Cluster.empty(), time.milliseconds(), (AuthenticationException) e); + } else { + log.info("Unable to fetch cluster metadata from node {}", + curNode(), e); + } + } + }; + } } /** @@ -1149,6 +1227,14 @@ public CreateTopicsResult createTopics(final Collection<NewTopic> newTopics, @Override public void handleResponse(AbstractResponse abstractResponse) { CreateTopicsResponse response = (CreateTopicsResponse) abstractResponse; + // Check for controller change + for (ApiError error : response.errors().values()) { + if (error.error() == Errors.NOT_CONTROLLER) { + metadataManager.clearController(); + metadataManager.requestUpdate(); + throw error.exception(); + } + } // Handle server responses for particular topics. for (Map.Entry<String, ApiError> entry : response.errors().entrySet()) { KafkaFutureImpl<Void> future = topicFutures.get(entry.getKey()); @@ -1212,6 +1298,14 @@ public DeleteTopicsResult deleteTopics(Collection<String> topicNames, @Override void handleResponse(AbstractResponse abstractResponse) { DeleteTopicsResponse response = (DeleteTopicsResponse) abstractResponse; + // Check for controller change + for (Errors error : response.errors().values()) { + if (error == Errors.NOT_CONTROLLER) { + metadataManager.clearController(); + metadataManager.requestUpdate(); + throw error.exception(); + } + } // Handle server responses for particular topics. for (Map.Entry<String, Errors> entry : response.errors().entrySet()) { KafkaFutureImpl<Void> future = topicFutures.get(entry.getKey()); @@ -1982,6 +2076,14 @@ public CreatePartitionsResult createPartitions(Map<String, NewPartitions> newPar @Override public void handleResponse(AbstractResponse abstractResponse) { CreatePartitionsResponse response = (CreatePartitionsResponse) abstractResponse; + // Check for controller change + for (ApiError error : response.errors().values()) { + if (error.error() == Errors.NOT_CONTROLLER) { + metadataManager.clearController(); + metadataManager.requestUpdate(); + throw error.exception(); + } + } for (Map.Entry<String, ApiError> result : response.errors().entrySet()) { KafkaFutureImpl<Void> future = futures.get(result.getKey()); if (result.getValue().isSuccess()) { diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internal/AdminMetadataManager.java b/clients/src/main/java/org/apache/kafka/clients/admin/internal/AdminMetadataManager.java new file mode 100644 index 00000000000..63e7fc8fa0a --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/admin/internal/AdminMetadataManager.java @@ -0,0 +1,247 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.kafka.clients.admin.internal; + +import org.apache.kafka.clients.MetadataUpdater; +import org.apache.kafka.common.Cluster; +import org.apache.kafka.common.Node; +import org.apache.kafka.common.PartitionInfo; +import org.apache.kafka.common.errors.AuthenticationException; +import org.apache.kafka.common.requests.MetadataResponse; +import org.apache.kafka.common.requests.RequestHeader; +import org.apache.kafka.common.utils.LogContext; +import org.apache.kafka.common.utils.Time; +import org.slf4j.Logger; + +import java.util.Collections; +import java.util.List; + +/** + * Manages the metadata for KafkaAdminClient. + * + * This class is not thread-safe. It is only accessed from the AdminClient + * service thread (which also uses the NetworkClient). + */ +public class AdminMetadataManager { + private Logger log; + + /** + * The timer. + */ + private final Time time; + + /** + * The minimum amount of time that we should wait between subsequent + * retries, when fetching metadata. + */ + private final long refreshBackoffMs; + + /** + * The minimum amount of time that we should wait before triggering an + * automatic metadata refresh. + */ + private final long metadataExpireMs; + + /** + * Used to update the NetworkClient metadata. + */ + private final AdminMetadataUpdater updater; + + /** + * The current metadata state. + */ + private State state = State.QUIESCENT; + + /** + * The time in wall-clock milliseconds when we last updated the metadata. + */ + private long lastMetadataUpdateMs = 0; + + /** + * The time in wall-clock milliseconds when we last attempted to fetch new + * metadata. + */ + private long lastMetadataFetchAttemptMs = 0; + + /** + * The current cluster information. + */ + private Cluster cluster = Cluster.empty(); + + /** + * If we got an authorization exception when we last attempted to fetch + * metadata, this is it; null, otherwise. + */ + private AuthenticationException authException = null; + + public class AdminMetadataUpdater implements MetadataUpdater { + @Override + public List<Node> fetchNodes() { + return cluster.nodes(); + } + + @Override + public boolean isUpdateDue(long now) { + return false; + } + + @Override + public long maybeUpdate(long now) { + return Long.MAX_VALUE; + } + + @Override + public void handleDisconnection(String destination) { + // Do nothing + } + + @Override + public void handleAuthenticationFailure(AuthenticationException e) { + log.info("AdminMetadataManager got AuthenticationException", e); + update(Cluster.empty(), time.milliseconds(), e); + } + + @Override + public void handleCompletedMetadataResponse(RequestHeader requestHeader, long now, MetadataResponse metadataResponse) { + // Do nothing + } + + @Override + public void requestUpdate() { + // Do nothing + } + } + + /** + * The current AdminMetadataManager state. + */ + enum State { + QUIESCENT, + UPDATE_REQUESTED, + UPDATE_PENDING; + } + + public AdminMetadataManager(LogContext logContext, Time time, long refreshBackoffMs, + long metadataExpireMs) { + this.log = logContext.logger(AdminMetadataManager.class); + this.time = time; + this.refreshBackoffMs = refreshBackoffMs; + this.metadataExpireMs = metadataExpireMs; + this.updater = new AdminMetadataUpdater(); + } + + public AdminMetadataUpdater updater() { + return updater; + } + + public boolean isReady() { + if (authException != null) { + log.debug("Metadata is not usable: failed to get metadata.", authException); + throw authException; + } + if (cluster.nodes().isEmpty()) { + log.trace("Metadata is not ready: bootstrap nodes have not been " + + "initialized yet."); + return false; + } + if (cluster.isBootstrapConfigured()) { + log.trace("Metadata is not ready: we have not fetched metadata from " + + "the bootstrap nodes yet."); + return false; + } + log.trace("Metadata is ready to use."); + return true; + } + + public Node controller() { + return cluster.controller(); + } + + public Node nodeById(int nodeId) { + return cluster.nodeById(nodeId); + } + + public void requestUpdate() { + if (state == State.QUIESCENT) { + state = State.UPDATE_REQUESTED; + log.debug("Requesting metadata update."); + } + } + + public void clearController() { + if (cluster.controller() != null) { + log.trace("Clearing cached controller node {}.", cluster.controller()); + this.cluster = new Cluster(cluster.clusterResource().clusterId(), + cluster.nodes(), + Collections.<PartitionInfo>emptySet(), + Collections.<String>emptySet(), + Collections.<String>emptySet(), + null); + } + } + + /** + * Determine if the AdminClient should fetch new metadata. + */ + public long metadataFetchDelayMs(long now) { + switch (state) { + case QUIESCENT: + // Calculate the time remaining until the next periodic update. + // We want to avoid making many metadata requests in a short amount of time, + // so there is a metadata refresh backoff period. + long timeSinceUpdate = now - lastMetadataUpdateMs; + long timeRemainingUntilUpdate = metadataExpireMs - timeSinceUpdate; + long timeSinceAttempt = now - lastMetadataFetchAttemptMs; + long timeRemainingUntilAttempt = refreshBackoffMs - timeSinceAttempt; + return Math.max(Math.max(0L, timeRemainingUntilUpdate), timeRemainingUntilAttempt); + case UPDATE_REQUESTED: + // An update has been explicitly requested. Do it as soon as possible. + return 0; + default: + // An update is already pending, so we don't need to initiate another one. + return Long.MAX_VALUE; + } + } + + /** + * Transition into the UPDATE_PENDING state. Updates lastMetadataFetchAttemptMs. + */ + public void transitionToUpdatePending(long now) { + this.state = State.UPDATE_PENDING; + this.lastMetadataFetchAttemptMs = now; + } + + /** + * Receive new metadata, and transition into the QUIESCENT state. + * Updates lastMetadataUpdateMs, cluster, and authException. + */ + public void update(Cluster cluster, long now, AuthenticationException authException) { + if (cluster.isBootstrapConfigured()) { + log.debug("Setting bootstrap cluster metadata {}.", cluster); + } else { + log.debug("Received cluster metadata {}{}.", + cluster, authException == null ? "" : " with authentication exception."); + } + this.state = State.QUIESCENT; + this.lastMetadataUpdateMs = now; + this.authException = authException; + if (!cluster.nodes().isEmpty()) { + this.cluster = cluster; + } + } +} diff --git a/clients/src/main/java/org/apache/kafka/common/Cluster.java b/clients/src/main/java/org/apache/kafka/common/Cluster.java index 0c59f33824d..ccbaa306d48 100644 --- a/clients/src/main/java/org/apache/kafka/common/Cluster.java +++ b/clients/src/main/java/org/apache/kafka/common/Cluster.java @@ -274,7 +274,8 @@ public Node controller() { @Override public String toString() { - return "Cluster(id = " + clusterResource.clusterId() + ", nodes = " + this.nodes + ", partitions = " + this.partitionsByTopicPartition.values() + ")"; + return "Cluster(id = " + clusterResource.clusterId() + ", nodes = " + this.nodes + + ", partitions = " + this.partitionsByTopicPartition.values() + ", controller = " + controller + ")"; } } 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 37b43e5e36e..7a8ba1c6b29 100644 --- a/clients/src/test/java/org/apache/kafka/clients/MockClient.java +++ b/clients/src/test/java/org/apache/kafka/clients/MockClient.java @@ -81,6 +81,7 @@ public FutureResponse(Node node, private Node node = null; private final Set<String> ready = new HashSet<>(); private final Map<Node, Long> blackedOut = new HashMap<>(); + private final Map<Node, Long> pendingAuthenticationErrors = new HashMap<>(); private final Map<Node, AuthenticationException> authenticationErrors = new HashMap<>(); // Use concurrent queue for requests so that requests may be queried from a different thread private final Queue<ClientRequest> requests = new ConcurrentLinkedDeque<>(); @@ -128,11 +129,16 @@ public void blackout(Node node, long duration) { } public void authenticationFailed(Node node, long duration) { + pendingAuthenticationErrors.remove(node); authenticationErrors.put(node, (AuthenticationException) Errors.SASL_AUTHENTICATION_FAILED.exception()); disconnect(node.idString()); blackout(node, duration); } + 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); @@ -174,6 +180,26 @@ public void disconnect(String node) { @Override public void send(ClientRequest request, long now) { + // Check if the request is directed to a node with a pending authentication error. + for (Iterator<Map.Entry<Node, Long>> authErrorIter = + pendingAuthenticationErrors.entrySet().iterator(); authErrorIter.hasNext(); ) { + Map.Entry<Node, Long> entry = authErrorIter.next(); + Node node = entry.getKey(); + long blackoutMs = entry.getValue(); + if (node.idString().equals(request.destination())) { + authErrorIter.remove(); + // Set up a disconnected ClientResponse and create an authentication error + // for the affected node. + authenticationFailed(node, blackoutMs); + AbstractRequest.Builder<?> builder = request.requestBuilder(); + 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); + responses.add(resp); + return; + } + } Iterator<FutureResponse> iterator = futureResponses.iterator(); while (iterator.hasNext()) { FutureResponse futureResp = iterator.next(); 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 10281fb6ffa..f862c140c0d 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 @@ -16,11 +16,12 @@ */ package org.apache.kafka.clients.admin; -import org.apache.kafka.clients.Metadata; import org.apache.kafka.clients.MockClient; import org.apache.kafka.common.Cluster; +import org.apache.kafka.common.requests.MetadataResponse; import org.apache.kafka.common.utils.Time; +import java.util.Collections; import java.util.HashMap; import java.util.Map; @@ -50,13 +51,25 @@ public AdminClientUnitTestEnv(Time time, Cluster cluster, String...vals) { } public AdminClientUnitTestEnv(Time time, Cluster cluster, Map<String, Object> config) { + this(newMockClient(time, cluster), time, cluster, config); + } + + 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.<MetadataResponse.TopicMetadata>emptyList())); + return mockClient; + } + + public AdminClientUnitTestEnv(MockClient mockClient, Time time, Cluster cluster, + Map<String, Object> config) { this.time = time; this.cluster = cluster; AdminClientConfig adminClientConfig = new AdminClientConfig(config); - Metadata metadata = new Metadata(adminClientConfig.getLong(AdminClientConfig.RETRY_BACKOFF_MS_CONFIG), - adminClientConfig.getLong(AdminClientConfig.METADATA_MAX_AGE_CONFIG), false); - this.mockClient = new MockClient(time, metadata); - this.adminClient = KafkaAdminClient.createInternal(adminClientConfig, mockClient, metadata, time); + this.mockClient = mockClient; + this.adminClient = KafkaAdminClient.createInternal(adminClientConfig, mockClient, time); } public Time time() { 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 0debed3dead..cdd9a282a3d 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 @@ -16,6 +16,7 @@ */ package org.apache.kafka.clients.admin; +import org.apache.kafka.clients.MockClient; import org.apache.kafka.clients.NodeApiVersions; import org.apache.kafka.clients.admin.DeleteAclsResult.FilterResults; import org.apache.kafka.clients.consumer.OffsetAndMetadata; @@ -39,6 +40,7 @@ import org.apache.kafka.common.errors.LeaderNotAvailableException; import org.apache.kafka.common.errors.NotLeaderForPartitionException; import org.apache.kafka.common.errors.OffsetOutOfRangeException; +import org.apache.kafka.common.errors.SaslAuthenticationException; import org.apache.kafka.common.errors.SecurityDisabledException; import org.apache.kafka.common.errors.TimeoutException; import org.apache.kafka.common.errors.UnknownTopicOrPartitionException; @@ -65,6 +67,8 @@ import org.apache.kafka.common.resource.ResourceFilter; import org.apache.kafka.common.resource.ResourceType; import org.apache.kafka.common.utils.MockTime; +import org.apache.kafka.common.utils.Time; +import org.apache.kafka.common.utils.Utils; import org.apache.kafka.test.TestCondition; import org.apache.kafka.test.TestUtils; import org.junit.Ignore; @@ -86,6 +90,7 @@ import java.util.Set; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; import static java.util.Arrays.asList; import static org.apache.kafka.common.requests.ResourceType.BROKER; @@ -169,15 +174,18 @@ public void testGenerateClientId() { KafkaAdminClient.generateClientId(newConfMap(AdminClientConfig.CLIENT_ID_CONFIG, "myCustomId"))); } - private static AdminClientUnitTestEnv mockClientEnv(String... configVals) { + private static Cluster mockCluster(int controllerIndex) { HashMap<Integer, Node> nodes = new HashMap<>(); nodes.put(0, new Node(0, "localhost", 8121)); nodes.put(1, new Node(1, "localhost", 8122)); nodes.put(2, new Node(2, "localhost", 8123)); - Cluster cluster = new Cluster("mockClusterId", nodes.values(), + return new Cluster("mockClusterId", nodes.values(), Collections.<PartitionInfo>emptySet(), Collections.<String>emptySet(), - Collections.<String>emptySet(), nodes.get(0)); - return new AdminClientUnitTestEnv(cluster, configVals); + Collections.<String>emptySet(), nodes.get(controllerIndex)); + } + + private static AdminClientUnitTestEnv mockClientEnv(String... configVals) { + return new AdminClientUnitTestEnv(mockCluster(0), configVals); } @Test @@ -204,7 +212,13 @@ private static void assertFutureError(Future<?> future, Class<? extends Throwabl */ @Test public void testTimeoutWithoutMetadata() throws Exception { - try (AdminClientUnitTestEnv env = mockClientEnv(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "10")) { + Cluster cluster = mockCluster(0); + MockClient mockClient = new MockClient(Time.SYSTEM); + try (final AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(mockClient, + Time.SYSTEM, + cluster, + newStrMap(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:8121", + AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "10"))) { env.kafkaClient().setNodeApiVersions(NodeApiVersions.create()); env.kafkaClient().setNode(new Node(0, "localhost", 8121)); env.kafkaClient().prepareResponse(new CreateTopicsResponse(Collections.singletonMap("myTopic", new ApiError(Errors.NONE, "")))); @@ -215,6 +229,30 @@ public void testTimeoutWithoutMetadata() throws Exception { } } + /** + * Test that we propagate exceptions encountered when fetching metadata. + */ + @Test + public void testPropagatedMetadataFetchException() throws Exception { + Cluster cluster = mockCluster(0); + MockClient mockClient = new MockClient(Time.SYSTEM); + mockClient.createPendingAuthenticationError(cluster.nodeById(0), + TimeUnit.DAYS.toMillis(1)); + try (final AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(mockClient, + Time.SYSTEM, + cluster, + newStrMap(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:8121", + AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "10"))) { + env.kafkaClient().setNodeApiVersions(NodeApiVersions.create()); + env.kafkaClient().setNode(env.cluster().nodeById(0)); + env.kafkaClient().prepareResponse(new CreateTopicsResponse(Collections.singletonMap("myTopic", new ApiError(Errors.NONE, "")))); + KafkaFuture<Void> future = env.adminClient().createTopics( + Collections.singleton(new NewTopic("myTopic", Collections.singletonMap(Integer.valueOf(0), asList(new Integer[]{0, 1, 2})))), + new CreateTopicsOptions().timeoutMs(1000)).all(); + assertFutureError(future, SaslAuthenticationException.class); + } + } + @Test public void testCreateTopics() throws Exception { try (AdminClientUnitTestEnv env = mockClientEnv()) { @@ -229,6 +267,30 @@ public void testCreateTopics() throws Exception { } } + @Test + public void testCreateTopicsHandleNotControllerException() throws Exception { + try (AdminClientUnitTestEnv env = mockClientEnv()) { + env.kafkaClient().setNodeApiVersions(NodeApiVersions.create()); + env.kafkaClient().prepareMetadataUpdate(mockCluster(0), Collections.<String>emptySet()); + env.kafkaClient().prepareMetadataUpdate(mockCluster(1), Collections.<String>emptySet()); + env.kafkaClient().setNode(env.cluster().nodeById(0)); + env.kafkaClient().prepareResponseFrom(new CreateTopicsResponse( + Collections.singletonMap("myTopic", new ApiError(Errors.NOT_CONTROLLER, ""))), + env.cluster().nodeById(0)); + env.kafkaClient().prepareResponse(new MetadataResponse(env.cluster().nodes(), + env.cluster().clusterResource().clusterId(), + 1, + Collections.<MetadataResponse.TopicMetadata>emptyList())); + env.kafkaClient().prepareResponseFrom(new CreateTopicsResponse( + Collections.singletonMap("myTopic", new ApiError(Errors.NONE, ""))), + env.cluster().nodeById(1)); + KafkaFuture<Void> future = env.adminClient().createTopics( + Collections.singleton(new NewTopic("myTopic", Collections.singletonMap(Integer.valueOf(0), asList(new Integer[]{0, 1, 2})))), + new CreateTopicsOptions().timeoutMs(10000)).all(); + future.get(); + } + } + @Test public void testInvalidTopicNames() throws Exception { try (AdminClientUnitTestEnv env = mockClientEnv()) { @@ -266,33 +328,31 @@ public void testInvalidTopicNames() throws Exception { } @Test - public void testAdminClientApisWithinBlackoutPeriodAfterAuthenticationFailure() throws Exception { - AdminClientUnitTestEnv env = mockClientEnv(); - Node node = env.cluster().controller(); - env.kafkaClient().setNodeApiVersions(NodeApiVersions.create()); - env.kafkaClient().setNode(node); - env.kafkaClient().authenticationFailed(node, 300); - - callAdminClientApisAndExpectAnAuthenticationError(env); - - // wait less than the blackout period, the connection should fail and the authentication error should remain - env.time().sleep(30); - assertTrue(env.kafkaClient().connectionFailed(node)); - callAdminClientApisAndExpectAnAuthenticationError(env); - - env.close(); + public void testAdminClientApisAuthenticationFailure() throws Exception { + Cluster cluster = mockCluster(0); + MockClient mockClient = new MockClient(Time.SYSTEM); + mockClient.createPendingAuthenticationError(cluster.nodeById(0), + TimeUnit.DAYS.toMillis(1)); + try (final AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(mockClient, + Time.SYSTEM, + cluster, + newStrMap(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:8121", + AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "1000"))) { + env.kafkaClient().setNodeApiVersions(NodeApiVersions.create()); + env.kafkaClient().setNode(env.cluster().controller()); + callAdminClientApisAndExpectAnAuthenticationError(env); + } } private void callAdminClientApisAndExpectAnAuthenticationError(AdminClientUnitTestEnv env) throws InterruptedException { - env.kafkaClient().prepareMetadataUpdate(env.cluster(), Collections.<String>emptySet()); - try { env.adminClient().createTopics( Collections.singleton(new NewTopic("myTopic", Collections.singletonMap(Integer.valueOf(0), asList(new Integer[]{0, 1, 2})))), new CreateTopicsOptions().timeoutMs(10000)).all().get(); fail("Expected an authentication error."); } catch (ExecutionException e) { - assertTrue("Expected only an authentication error.", e.getCause() instanceof AuthenticationException); + assertTrue("Expected an authentication error, but got " + Utils.stackTrace(e), + e.getCause() instanceof AuthenticationException); } try { @@ -302,35 +362,40 @@ private void callAdminClientApisAndExpectAnAuthenticationError(AdminClientUnitTe env.adminClient().createPartitions(counts).all().get(); fail("Expected an authentication error."); } catch (ExecutionException e) { - assertTrue("Expected only an authentication error.", e.getCause() instanceof AuthenticationException); + assertTrue("Expected an authentication error, but got " + Utils.stackTrace(e), + e.getCause() instanceof AuthenticationException); } try { env.adminClient().createAcls(asList(ACL1, ACL2)).all().get(); fail("Expected an authentication error."); } catch (ExecutionException e) { - assertTrue("Expected only an authentication error.", e.getCause() instanceof AuthenticationException); + assertTrue("Expected an authentication error, but got " + Utils.stackTrace(e), + e.getCause() instanceof AuthenticationException); } try { env.adminClient().describeAcls(FILTER1).values().get(); fail("Expected an authentication error."); } catch (ExecutionException e) { - assertTrue("Expected only an authentication error.", e.getCause() instanceof AuthenticationException); + assertTrue("Expected an authentication error, but got " + Utils.stackTrace(e), + e.getCause() instanceof AuthenticationException); } try { env.adminClient().deleteAcls(asList(FILTER1, FILTER2)).all().get(); fail("Expected an authentication error."); } catch (ExecutionException e) { - assertTrue("Expected only an authentication error.", e.getCause() instanceof AuthenticationException); + assertTrue("Expected an authentication error, but got " + Utils.stackTrace(e), + e.getCause() instanceof AuthenticationException); } try { env.adminClient().describeConfigs(Collections.singleton(new ConfigResource(ConfigResource.Type.BROKER, "0"))).all().get(); fail("Expected an authentication error."); } catch (ExecutionException e) { - assertTrue("Expected only an authentication error.", e.getCause() instanceof AuthenticationException); + assertTrue("Expected an authentication error, but got " + Utils.stackTrace(e), + e.getCause() instanceof AuthenticationException); } } @@ -918,7 +983,7 @@ public FailureInjectingTimeoutProcessor(long now) { } boolean callHasExpired(KafkaAdminClient.Call call) { - if (shouldInjectFailure()) { + if ((!call.isInternal()) && shouldInjectFailure()) { log.debug("Injecting timeout for {}.", call); return true; } else { diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/util/TopicAdminTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/util/TopicAdminTest.java index cda68795689..5b1e1553559 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/util/TopicAdminTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/util/TopicAdminTest.java @@ -52,7 +52,6 @@ public void returnNullWithApiVersionMismatch() { try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(cluster)) { env.kafkaClient().setNode(cluster.controller()); env.kafkaClient().setNodeApiVersions(NodeApiVersions.create()); - env.kafkaClient().prepareMetadataUpdate(env.cluster(), Collections.<String>emptySet()); env.kafkaClient().prepareResponse(createTopicResponseWithUnsupportedVersion(newTopic)); TopicAdmin admin = new TopicAdmin(null, env.adminClient()); boolean created = admin.createTopic(newTopic); @@ -65,7 +64,7 @@ public void returnNullWithClusterAuthorizationFailure() { final NewTopic newTopic = TopicAdmin.defineTopic("myTopic").partitions(1).compacted().build(); Cluster cluster = createCluster(1); try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(cluster)) { - env.kafkaClient().prepareMetadataUpdate(env.cluster(), Collections.<String>emptySet()); + env.kafkaClient().setNode(cluster.nodes().iterator().next()); env.kafkaClient().prepareResponse(createTopicResponseWithClusterAuthorizationException(newTopic)); TopicAdmin admin = new TopicAdmin(null, env.adminClient()); boolean created = admin.createTopic(newTopic); ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Fix AdminClient error handling when metadata changes > ---------------------------------------------------- > > Key: KAFKA-6299 > URL: https://issues.apache.org/jira/browse/KAFKA-6299 > Project: Kafka > Issue Type: Bug > Reporter: Colin P. McCabe > Assignee: Colin P. McCabe > Priority: Major > Fix For: 2.0.0 > > > * AdminClient should only call Metadata#requestUpdate when needed. > * AdminClient should retry requests for which the controller has changed. > * Fix an issue where AdminClient requests might not get a security exception, > even when a metadata fetch fails with an authorization exception. > * Fix a possible issue where AdminClient might leak a socket after the > timeout expires on a hard close, if a very narrow race condition is hit -- This message was sent by Atlassian JIRA (v7.6.3#76005)