diff --git a/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java b/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java index 266acd587f37c..2c98ab910d316 100644 --- a/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java +++ b/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java @@ -158,6 +158,12 @@ public class CommonClientConfigs { + "of 3 lowest-expected-latency nodes)"; public static final String DEFAULT_LEAST_LOADED_NODE_ALGORITHM = LeastLoadedNodeAlgorithm.VANILLA.name(); + public static final String LI_CLIENT_CLUSTER_METADATA_EXPIRE_TIME_MS_CONFIG = "li.client.cluster.metadata.expire.time.ms"; + public static final String LI_CLIENT_CLUSTER_METADATA_EXPIRE_TIME_MS_DOC = "The configuration controls the max time the client cluster metadata can remain idle/unchanged before " + + "trying to resolve the bootstrap servers from given url again; different from the other metadata.max.age.ms config, " + + "which will try to refresh metadata by choosing from existing resolved node set, this config will force resolving " + + "the bootstrap url again to get new node set and use the new node set to send update metadata request"; + /** * Postprocess the configuration so that exponential backoff is disabled when reconnect backoff * is explicitly configured but the maximum reconnect backoff is not explicitly configured. diff --git a/clients/src/main/java/org/apache/kafka/clients/ManualMetadataUpdater.java b/clients/src/main/java/org/apache/kafka/clients/ManualMetadataUpdater.java index 4868fc38aeeb9..2ba6cb0492a3c 100644 --- a/clients/src/main/java/org/apache/kafka/clients/ManualMetadataUpdater.java +++ b/clients/src/main/java/org/apache/kafka/clients/ManualMetadataUpdater.java @@ -60,6 +60,11 @@ public boolean isUpdateDue(long now) { return false; } + @Override + public boolean isUpdateClusterMetadataDue(long now) { + return false; + } + @Override public long maybeUpdate(long now) { return Long.MAX_VALUE; diff --git a/clients/src/main/java/org/apache/kafka/clients/Metadata.java b/clients/src/main/java/org/apache/kafka/clients/Metadata.java index 47e4141c7da28..f191659bae746 100644 --- a/clients/src/main/java/org/apache/kafka/clients/Metadata.java +++ b/clients/src/main/java/org/apache/kafka/clients/Metadata.java @@ -76,6 +76,10 @@ public class Metadata implements Closeable { private boolean isClosed; private final Map lastSeenLeaderEpochs; + private final long maxClusterMetadataExpireTimeMs; + private int nodesTriedSinceLastSuccessfulRefresh; + private boolean forceClusterMetadataUpdateFromBootstrap; + /** * Create a new Metadata instance * @@ -89,6 +93,14 @@ public Metadata(long refreshBackoffMs, long metadataExpireMs, LogContext logContext, ClusterResourceListeners clusterResourceListeners) { + this(refreshBackoffMs, metadataExpireMs, logContext, clusterResourceListeners, Long.MAX_VALUE); + } + + public Metadata(long refreshBackoffMs, + long metadataExpireMs, + LogContext logContext, + ClusterResourceListeners clusterResourceListeners, + long metadataClusterMetadataExpireTimeMs) { this.log = logContext.logger(Metadata.class); this.refreshBackoffMs = refreshBackoffMs; this.metadataExpireMs = metadataExpireMs; @@ -102,6 +114,9 @@ public Metadata(long refreshBackoffMs, this.lastSeenLeaderEpochs = new HashMap<>(); this.invalidTopics = Collections.emptySet(); this.unauthorizedTopics = Collections.emptySet(); + this.maxClusterMetadataExpireTimeMs = metadataClusterMetadataExpireTimeMs; + this.nodesTriedSinceLastSuccessfulRefresh = 0; + this.forceClusterMetadataUpdateFromBootstrap = false; } /** @@ -111,6 +126,26 @@ public synchronized Cluster fetch() { return cache.cluster(); } + /** + * Increment the nodesTriedSinceLastSuccessfulRefresh + */ + public synchronized void incrementNodesTriedSinceLastSuccessfulRefresh() { + this.nodesTriedSinceLastSuccessfulRefresh++; + } + + /** + * Whether the client should update the cluster metadata by resolving the bootstrap server again + * @param nowMs + * @return true if client hasn't refreshed cluster metadata for maxClusterMetadataExpireTimeMs and + * has tried connecting to at least one node in current node set; or forceClusterMetadataUpdateFromBootstrap + * has been set by receiving stale metadata from a different cluster + */ + public synchronized boolean shouldUpdateClusterMetadataFromBootstrap(long nowMs) { + return (this.nodesTriedSinceLastSuccessfulRefresh >= 1 && + this.lastSuccessfulRefreshMs + this.maxClusterMetadataExpireTimeMs <= nowMs) || + this.forceClusterMetadataUpdateFromBootstrap; + } + /** * Return the next time when the current cluster info can be updated (i.e., backoff time has elapsed). * @@ -146,6 +181,15 @@ public synchronized int requestUpdate() { return this.updateVersion; } + /** + * Request an update of the current cluster metadata info by resolving the bootstrap server and randomly pick + * a node from the resolved node set. This happens when client receives stale metadata response from brokers in + * a different cluster and need to refresh the cluster metadata without waiting for maxClusterMetadataExpireTimeMs + */ + public synchronized void requestClusterMetadataUpdateFromBootstrap() { + this.forceClusterMetadataUpdateFromBootstrap = true; + } + /** * Request an update for the partition metadata iff the given leader epoch is at newer than the last seen leader epoch */ @@ -236,7 +280,18 @@ public synchronized void update(int requestVersion, MetadataResponse response, l if (isClosed()) throw new IllegalStateException("Update requested after metadata close"); - validateCluster(response.clusterId()); + if (!validateCluster(response.clusterId())) { + //if validateCluster fails, do not update metadataCache with the wrong cluster information, + //just return and wait for next update + // + //here we don't blacklist this node from the cluster's + //node set since we don't have enough information from the response to map to the actual node, + //and since there are usually hours to days interval before we put a removed broker to a different + //cluster, clients should either find another node in cached node set or resolved bootstrap server + //again and find a new node to send update metadata request, it should be ok to not blacklist this node + requestClusterMetadataUpdateFromBootstrap(); + return; + } if (requestVersion == this.requestVersion) this.needUpdate = false; @@ -246,6 +301,8 @@ public synchronized void update(int requestVersion, MetadataResponse response, l this.lastRefreshMs = now; this.lastSuccessfulRefreshMs = now; this.updateVersion += 1; + this.nodesTriedSinceLastSuccessfulRefresh = 0; + this.forceClusterMetadataUpdateFromBootstrap = false; String previousClusterId = cache.cluster().clusterResource().clusterId(); @@ -265,8 +322,9 @@ public synchronized void update(int requestVersion, MetadataResponse response, l log.debug("Updated cluster metadata updateVersion {} to {}", this.updateVersion, this.cache); } - private void validateCluster(String newClusterId) { + private boolean validateCluster(String newClusterId) { String previousClusterId = this.cache.cluster().clusterResource().clusterId(); + boolean validateResult = true; if (previousClusterId != null && newClusterId != null && !previousClusterId.equals(newClusterId)) { // kafka cluster id is unique. @@ -288,10 +346,10 @@ private void validateCluster(String newClusterId) { log.error("Received metadata from a different cluster {}, current cluster {} has no valid brokers anymore," + "please reboot the producer/consumer", newClusterId, previousClusterId); - throw new StaleClusterMetadataException( - "Trying to access a different cluster " + newClusterId + ", previous connected cluster " + previousClusterId); - + validateResult = false; } + + return validateResult; } private void maybeSetMetadataError(Cluster cluster) { 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 b91a0d6a0ca0d..2fffad0859076 100644 --- a/clients/src/main/java/org/apache/kafka/clients/MetadataUpdater.java +++ b/clients/src/main/java/org/apache/kafka/clients/MetadataUpdater.java @@ -45,6 +45,12 @@ public interface MetadataUpdater extends Closeable { */ boolean isUpdateDue(long now); + /** + * Returns true if the cluster metadata hasn't refreshed for li.client.cluster.metadata.expire.time.ms + * and has tried at least one node in the cached metadata node set + */ + boolean isUpdateClusterMetadataDue(long now); + /** * Starts a cluster metadata update if needed and possible. Returns the time until the metadata update (which would * be 0 if an update has been started as a result of this call). 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 873a5d990ce30..3c86c6f000e79 100644 --- a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java +++ b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java @@ -16,6 +16,7 @@ */ package org.apache.kafka.clients; +import java.util.Collections; import java.util.PriorityQueue; import org.apache.kafka.common.Cluster; import org.apache.kafka.common.KafkaException; @@ -135,6 +136,8 @@ private enum State { private final List abortedSends = new LinkedList<>(); + private final List bootstrapServers = new LinkedList<>(); + private final Sensor throttleTimeSensor; private final AtomicReference state; @@ -172,7 +175,8 @@ public NetworkClient(Selectable selector, null, logContext, null, - LeastLoadedNodeAlgorithm.VANILLA); + LeastLoadedNodeAlgorithm.VANILLA, + Collections.emptyList()); } public NetworkClient(Selectable selector, @@ -207,7 +211,8 @@ public NetworkClient(Selectable selector, throttleTimeSensor, logContext, null, - LeastLoadedNodeAlgorithm.VANILLA); + LeastLoadedNodeAlgorithm.VANILLA, + Collections.emptyList()); } public NetworkClient(Selectable selector, @@ -225,7 +230,8 @@ public NetworkClient(Selectable selector, ApiVersions apiVersions, Sensor throttleTimeSensor, LogContext logContext, - LeastLoadedNodeAlgorithm leastLoadedNodeAlgorithm) { + LeastLoadedNodeAlgorithm leastLoadedNodeAlgorithm, + List bootstrapServers) { this(null, metadata, selector, @@ -243,7 +249,8 @@ public NetworkClient(Selectable selector, throttleTimeSensor, logContext, null, - leastLoadedNodeAlgorithm); + leastLoadedNodeAlgorithm, + bootstrapServers); } public NetworkClient(Selectable selector, @@ -279,7 +286,8 @@ public NetworkClient(Selectable selector, throttleTimeSensor, logContext, clientSoftwareNameAndCommit, - LeastLoadedNodeAlgorithm.VANILLA); + LeastLoadedNodeAlgorithm.VANILLA, + Collections.emptyList()); } public NetworkClient(Selectable selector, @@ -298,7 +306,8 @@ public NetworkClient(Selectable selector, Sensor throttleTimeSensor, LogContext logContext, String clientSoftwareNameAndCommit, - LeastLoadedNodeAlgorithm leastLoadedNodeAlgorithm) { + LeastLoadedNodeAlgorithm leastLoadedNodeAlgorithm, + List bootstrapServerUrls) { this(null, metadata, selector, @@ -316,7 +325,8 @@ public NetworkClient(Selectable selector, throttleTimeSensor, logContext, clientSoftwareNameAndCommit, - leastLoadedNodeAlgorithm); + leastLoadedNodeAlgorithm, + bootstrapServerUrls); } public NetworkClient(Selectable selector, @@ -350,7 +360,8 @@ public NetworkClient(Selectable selector, null, logContext, null, - LeastLoadedNodeAlgorithm.VANILLA); + LeastLoadedNodeAlgorithm.VANILLA, + Collections.emptyList()); } public NetworkClient(Selectable selector, @@ -367,7 +378,8 @@ public NetworkClient(Selectable selector, boolean discoverBrokerVersions, ApiVersions apiVersions, LogContext logContext, - LeastLoadedNodeAlgorithm leastLoadedNodeAlgorithm) { + LeastLoadedNodeAlgorithm leastLoadedNodeAlgorithm, + List bootstrapServerUrls) { this(metadataUpdater, null, selector, @@ -385,7 +397,8 @@ public NetworkClient(Selectable selector, null, logContext, null, - leastLoadedNodeAlgorithm); + leastLoadedNodeAlgorithm, + bootstrapServerUrls); } private NetworkClient(MetadataUpdater metadataUpdater, @@ -405,7 +418,8 @@ private NetworkClient(MetadataUpdater metadataUpdater, Sensor throttleTimeSensor, LogContext logContext, String clientSoftwareNameAndCommit, - LeastLoadedNodeAlgorithm leastLoadedNodeAlgorithm) { + LeastLoadedNodeAlgorithm leastLoadedNodeAlgorithm, + List bootstrapServersConfig) { /* It would be better if we could pass `DefaultMetadataUpdater` from the public constructor, but it's not * possible because `DefaultMetadataUpdater` is an inner class and it can only be instantiated after the * super constructor is invoked. @@ -440,6 +454,7 @@ private NetworkClient(MetadataUpdater metadataUpdater, throw new IllegalArgumentException("must specify leastLoadedNodeAlgorithm"); } this.leastLoadedNodeAlgorithm = leastLoadedNodeAlgorithm; + this.bootstrapServers.addAll(bootstrapServersConfig); } public void setEnableClientResponseWithFinalize(boolean enableClientResponseWithFinalize) { @@ -736,16 +751,7 @@ public List poll(long timeout, long now) { long updatedNow = this.time.milliseconds(); List responses = new ArrayList<>(); handleCompletedSends(responses, updatedNow); - - try { - handleCompletedReceives(responses, updatedNow); - } catch (StaleClusterMetadataException e) { - // upon stale metadata exception from a different cluster, close the network client - // the producer/consumer will hit closedSelector exception and close - log.error("Received stale metadata from a different cluster, close the network client now"); - this.close(); - } - + handleCompletedReceives(responses, updatedNow); handleDisconnections(responses, updatedNow); handleConnections(); handleInitiateApiVersionRequests(updatedNow); @@ -855,6 +861,27 @@ public Node leastLoadedNode(long now) { if (algo == null) { throw new IllegalStateException("leastLoadedNodeAlgorithm cannot be null"); } + + if (this.metadataUpdater.isUpdateClusterMetadataDue(now)) { + //update cluster metadata due, resolve bootstrap server and randomly pick up + //one node from the resolved node set as least loaded node + List addresses = ClientUtils.parseAndValidateAddresses( + this.bootstrapServers, + this.clientDnsLookup.toString()); + List newNodes = new ArrayList<>(); + + int nodeId = -1; + for (InetSocketAddress address : addresses) { + newNodes.add(new Node(nodeId--, address.getHostString(), address.getPort())); + } + + int offset = this.randOffset.nextInt(newNodes.size()); + Node node = newNodes.get(offset); + log.info("Resolved bootstrap server again, randomly picked node {} as least loaded node from the resolved node set", node); + + return node; + } + switch (algo) { case VANILLA: return vanillaLeastLoadedNode(now, nodes); @@ -1371,6 +1398,11 @@ public boolean isUpdateDue(long now) { return !hasFetchInProgress() && this.metadata.timeToNextUpdate(now) == 0; } + @Override + public boolean isUpdateClusterMetadataDue(long now) { + return this.metadata.shouldUpdateClusterMetadataFromBootstrap(now); + } + private boolean hasFetchInProgress() { return inProgressRequestVersion != null; } @@ -1487,6 +1519,7 @@ private boolean isAnyNodeConnecting() { */ private long maybeUpdate(long now, Node node) { String nodeConnectionId = node.idString(); + this.metadata.incrementNodesTriedSinceLastSuccessfulRefresh(); if (canSendRequest(nodeConnectionId, now)) { Metadata.MetadataRequestAndVersion requestAndVersion = metadata.newMetadataRequestAndVersion(); 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 7473a4a8aa757..5ff8b04826bc4 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 @@ -465,7 +465,8 @@ static KafkaAdminClient createInternal(AdminClientConfig config, TimeoutProcesso true, apiVersions, logContext, - leastLoadedNodeAlgorithm); + leastLoadedNodeAlgorithm, + Collections.emptyList()); return new KafkaAdminClient(config, clientId, time, metadataManager, metrics, networkClient, timeoutProcessorFactory, logContext); } catch (Throwable exc) { diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminMetadataManager.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminMetadataManager.java index 1de680fae1dc0..1f1b12aa403d7 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminMetadataManager.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminMetadataManager.java @@ -90,6 +90,11 @@ public List fetchNodes() { return cluster.nodes(); } + @Override + public boolean isUpdateClusterMetadataDue(long now) { + return false; + } + @Override public boolean isUpdateDue(long now) { return false; diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java index 15e089efb1b1f..bac8f80f12e23 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java @@ -315,6 +315,8 @@ public class ConsumerConfig extends AbstractConfig { public static final String ENABLE_CLIENT_RESPONSE_LEAK_CHECK = "linkedin.enable.client.resonse.leakcheck"; public static final String ENABLE_CLIENT_RESPONSE_LEAK_CHECK_DOC = "Use ClientResponse with finalize method to check the release of NetworkReceive buffer."; + public static final String LI_CLIENT_CLUSTER_METADATA_EXPIRE_TIME_MS_CONFIG = CommonClientConfigs.LI_CLIENT_CLUSTER_METADATA_EXPIRE_TIME_MS_CONFIG; + static { CONFIG = new ConfigDef().define(BOOTSTRAP_SERVERS_CONFIG, Type.LIST, @@ -358,6 +360,12 @@ public class ConsumerConfig extends AbstractConfig { atLeast(0), Importance.LOW, CommonClientConfigs.METADATA_MAX_AGE_DOC) + .define(LI_CLIENT_CLUSTER_METADATA_EXPIRE_TIME_MS_CONFIG, + Type.LONG, + 60 * 60 * 1000, + atLeast(0), + Importance.LOW, + CommonClientConfigs.LI_CLIENT_CLUSTER_METADATA_EXPIRE_TIME_MS_DOC) .define(ENABLE_AUTO_COMMIT_CONFIG, Type.BOOLEAN, true, diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java index fb9847da8932d..f47b58f451619 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java @@ -735,7 +735,8 @@ else if (enableAutoCommit) config.getLong(ConsumerConfig.METADATA_MAX_AGE_CONFIG), !config.getBoolean(ConsumerConfig.EXCLUDE_INTERNAL_TOPICS_CONFIG), config.getBoolean(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG), - subscriptions, logContext, clusterResourceListeners, metrics); + subscriptions, logContext, clusterResourceListeners, metrics, + config.getLong(ConsumerConfig.LI_CLIENT_CLUSTER_METADATA_EXPIRE_TIME_MS_CONFIG)); List addresses = ClientUtils.parseAndValidateAddresses( config.getList(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG), config.getString(ConsumerConfig.CLIENT_DNS_LOOKUP_CONFIG)); this.metadata.bootstrap(addresses); @@ -776,7 +777,8 @@ else if (enableAutoCommit) throttleTimeSensor, logContext, config.getString(ConsumerConfig.LI_CLIENT_SOFTWARE_NAME_AND_COMMIT_CONFIG), - leastLoadedNodeAlgorithm); + leastLoadedNodeAlgorithm, + config.getList(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)); netClient.setEnableClientResponseWithFinalize(config.getBoolean(ConsumerConfig.ENABLE_CLIENT_RESPONSE_LEAK_CHECK)); this.client = new ConsumerNetworkClient( diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerMetadata.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerMetadata.java index dd07d7eac611e..e5177576b06b4 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerMetadata.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerMetadata.java @@ -46,7 +46,20 @@ public ConsumerMetadata(long refreshBackoffMs, LogContext logContext, ClusterResourceListeners clusterResourceListeners, Metrics metrics) { - super(refreshBackoffMs, metadataExpireMs, logContext, clusterResourceListeners); + this(refreshBackoffMs, metadataExpireMs, includeInternalTopics, allowAutoTopicCreation, subscription, + logContext, clusterResourceListeners, metrics, Long.MAX_VALUE); + } + + public ConsumerMetadata(long refreshBackoffMs, + long metadataExpireMs, + boolean includeInternalTopics, + boolean allowAutoTopicCreation, + SubscriptionState subscription, + LogContext logContext, + ClusterResourceListeners clusterResourceListeners, + Metrics metrics, + long clusterMetadataExpireMs) { + super(refreshBackoffMs, metadataExpireMs, logContext, clusterResourceListeners, clusterMetadataExpireMs); this.includeInternalTopics = includeInternalTopics; this.allowAutoTopicCreation = allowAutoTopicCreation; this.subscription = subscription; diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java index 67adc032c5781..c0426d01f69c4 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java @@ -431,7 +431,8 @@ public KafkaProducer(Properties properties, Serializer keySerializer, Seriali Time.SYSTEM, config.getLong(ProducerConfig.METADATA_TOPIC_EXPIRY_MS_CONFIG), config.getBoolean(ProducerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG), - metrics); + metrics, + config.getLong(ProducerConfig.LI_CLIENT_CLUSTER_METADATA_EXPIRE_TIME_MS_CONFIG)); this.metadata.bootstrap(addresses); } this.errors = this.metrics.sensor("errors"); @@ -488,7 +489,8 @@ Sender newSender(LogContext logContext, KafkaClient kafkaClient, ProducerMetadat throttleTimeSensor, logContext, producerConfig.getString(ProducerConfig.LI_CLIENT_SOFTWARE_NAME_AND_COMMIT_CONFIG), - leastLoadedNodeAlgorithm); + leastLoadedNodeAlgorithm, + producerConfig.getList(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG)); int retries = configureRetries(producerConfig, transactionManager != null, log); short acks = configureAcks(producerConfig, transactionManager != null, log); return new Sender(logContext, diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java b/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java index 546c6b1352f7a..c36ccf038d4f2 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java @@ -273,6 +273,8 @@ public class ProducerConfig extends AbstractConfig { public static final String LEAST_LOADED_NODE_ALGORITHM_DOC = CommonClientConfigs.LEAST_LOADED_NODE_ALGORITHM_DOC; public static final String DEFAULT_LEAST_LOADED_NODE_ALGORITHM = CommonClientConfigs.DEFAULT_LEAST_LOADED_NODE_ALGORITHM; + public static final String LI_CLIENT_CLUSTER_METADATA_EXPIRE_TIME_MS_CONFIG = CommonClientConfigs.LI_CLIENT_CLUSTER_METADATA_EXPIRE_TIME_MS_CONFIG; + static { CONFIG = new ConfigDef().define(BOOTSTRAP_SERVERS_CONFIG, Type.LIST, Collections.emptyList(), new ConfigDef.NonNullValidator(), Importance.HIGH, CommonClientConfigs.BOOTSTRAP_SERVERS_DOC) .define(CLIENT_DNS_LOOKUP_CONFIG, @@ -321,6 +323,12 @@ public class ProducerConfig extends AbstractConfig { Importance.MEDIUM, REQUEST_TIMEOUT_MS_DOC) .define(METADATA_MAX_AGE_CONFIG, Type.LONG, 5 * 60 * 1000, atLeast(0), Importance.LOW, METADATA_MAX_AGE_DOC) + .define(LI_CLIENT_CLUSTER_METADATA_EXPIRE_TIME_MS_CONFIG, + Type.LONG, + 60 * 60 * 1000, + atLeast(0), + Importance.LOW, + CommonClientConfigs.LI_CLIENT_CLUSTER_METADATA_EXPIRE_TIME_MS_DOC) .define(METRICS_SAMPLE_WINDOW_MS_CONFIG, Type.LONG, 30000, diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerMetadata.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerMetadata.java index e21aae5fac7c2..1af770be3dda2 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerMetadata.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerMetadata.java @@ -68,7 +68,20 @@ public ProducerMetadata(long refreshBackoffMs, long topicExpiryMs, boolean allowAutoTopicCreation, Metrics metrics) { - super(refreshBackoffMs, metadataExpireMs, logContext, clusterResourceListeners); + this(refreshBackoffMs, metadataExpireMs, logContext, clusterResourceListeners, time, + topicExpiryMs, allowAutoTopicCreation, metrics, Long.MAX_VALUE); + } + + public ProducerMetadata(long refreshBackoffMs, + long metadataExpireMs, + LogContext logContext, + ClusterResourceListeners clusterResourceListeners, + Time time, + long topicExpiryMs, + boolean allowAutoTopicCreation, + Metrics metrics, + long clusterMetadataExpireMs) { + super(refreshBackoffMs, metadataExpireMs, logContext, clusterResourceListeners, clusterMetadataExpireMs); this.log = logContext.logger(ProducerMetadata.class); this.time = time; this.topicExpiryMs = topicExpiryMs; diff --git a/clients/src/test/java/org/apache/kafka/clients/MetadataTest.java b/clients/src/test/java/org/apache/kafka/clients/MetadataTest.java index f690dc685b27b..64b68be3d9669 100644 --- a/clients/src/test/java/org/apache/kafka/clients/MetadataTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/MetadataTest.java @@ -19,6 +19,7 @@ import org.apache.kafka.common.Cluster; import org.apache.kafka.common.Node; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.config.ConfigException; import org.apache.kafka.common.errors.InvalidTopicException; import org.apache.kafka.common.errors.TopicAuthorizationException; import org.apache.kafka.common.internals.ClusterResourceListeners; @@ -37,6 +38,7 @@ import org.apache.kafka.common.utils.MockTime; import org.apache.kafka.common.utils.Time; import org.apache.kafka.test.MockClusterResourceListener; +import org.apache.kafka.test.MockSelector; import org.apache.kafka.test.TestUtils; import org.junit.Test; @@ -59,11 +61,24 @@ public class MetadataTest { + protected final int defaultRequestTimeoutMs = 1000; + protected final MockTime time = new MockTime(); + protected final MockSelector selector = new MockSelector(time); + protected final Node node = TestUtils.singletonCluster().nodes().iterator().next(); + private long refreshBackoffMs = 100; private long metadataExpireMs = 1000; private Metadata metadata = new Metadata(refreshBackoffMs, metadataExpireMs, new LogContext(), new ClusterResourceListeners()); + private Metadata clusterMetadata = new Metadata(refreshBackoffMs, metadataExpireMs, new LogContext(), + new ClusterResourceListeners(), 0); + + private NetworkClient clusterClient = new NetworkClient(selector, clusterMetadata, "mock-cluster-md", Integer.MAX_VALUE, + 0, 0, 64 * 1024, 64 * 1024, + defaultRequestTimeoutMs, ClientDnsLookup.DEFAULT, time, true, new ApiVersions(), null, new LogContext(), + LeastLoadedNodeAlgorithm.VANILLA, Collections.singletonList("some.invalid.hostname.foo.bar.local:9999")); + private static MetadataResponse emptyMetadataResponse() { return MetadataResponse.prepareResponse( Collections.emptyList(), @@ -78,6 +93,23 @@ public void testMetadataUpdateAfterClose() { metadata.update(emptyMetadataResponse(), 1000); } + @Test(expected = ConfigException.class) + public void testResolveBootstrapAfterClusterMetadataTimeout() { + List addresses = ClientUtils.parseAndValidateAddresses(Collections.singletonList("example.com:10000")); + //bootstrap the metadata cache with some valid nodes + clusterMetadata.bootstrap(addresses); + + //first time call leastLoadedNode on the created NetworkClient should pass since nodesTriedSinceLastSuccessfulRefresh is 0 + clusterClient.leastLoadedNode(time.milliseconds()); + + //simulate metadata refresh attempt by incrementing the nodesTriedSinceLastSuccessfulRefresh by 1 + clusterMetadata.incrementNodesTriedSinceLastSuccessfulRefresh(); + + //now second call to leastLoadedNode on the NetworkClient should fail since we passed an invalid bootstrap server to it + //it should throw a ConfigException of no resolvable bootstrap server in provided urls + clusterClient.leastLoadedNode(time.milliseconds()); + } + private static void checkTimeToNextUpdate(long refreshBackoffMs, long metadataExpireMs) { long now = 10000; diff --git a/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java b/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java index 597d5beef380c..17def86d032f0 100644 --- a/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java @@ -77,10 +77,12 @@ public class NetworkClientTest { protected final long reconnectBackoffMaxMsTest = 10 * 10000; private final TestMetadataUpdater metadataUpdater = new TestMetadataUpdater(Collections.singletonList(node)); + private final TestClusterMetadataUpdater clusterMetadataUpdater = new TestClusterMetadataUpdater(Collections.singletonList(node)); private final NetworkClient client = createNetworkClient(reconnectBackoffMaxMsTest); private final NetworkClient clientWithNoExponentialBackoff = createNetworkClient(reconnectBackoffMsTest); private final NetworkClient clientWithStaticNodes = createNetworkClientWithStaticNodes(); private final NetworkClient clientWithNoVersionDiscovery = createNetworkClientWithNoVersionDiscovery(); + private final NetworkClient clusterClient = createClusterNetworkClient(); private NetworkClient createNetworkClient(long reconnectBackoffMaxMs) { return new NetworkClient(selector, metadataUpdater, "mock", Integer.MAX_VALUE, @@ -107,6 +109,13 @@ private NetworkClient createNetworkClientWithNoVersionDiscovery() { ClientDnsLookup.DEFAULT, time, false, new ApiVersions(), new LogContext()); } + private NetworkClient createClusterNetworkClient() { + return new NetworkClient(selector, clusterMetadataUpdater, "mock-cluster-md", Integer.MAX_VALUE, + 0, 0, 64 * 1024, 64 * 1024, + defaultRequestTimeoutMs, ClientDnsLookup.DEFAULT, time, true, new ApiVersions(), new LogContext(), + LeastLoadedNodeAlgorithm.VANILLA, Collections.singletonList("example.com:10000")); + } + @Before public void setup() { selector.reset(); @@ -650,6 +659,13 @@ private void sendThrottledProduceResponse(int correlationId, int throttleMs) { resp); } + @Test + public void testResolveBootstrapInLeastLoadedNode() { + clusterClient.ready(node, time.milliseconds()); + assertFalse(clusterClient.isReady(node, time.milliseconds())); + assertNotEquals(node, clusterClient.leastLoadedNode(time.milliseconds())); + } + @Test public void testLeastLoadedNode() { client.ready(node, time.milliseconds()); @@ -1049,4 +1065,38 @@ public KafkaException getAndClearFailure() { return failure; } } + + private static class TestClusterMetadataUpdater extends ManualMetadataUpdater { + KafkaException failure; + + public TestClusterMetadataUpdater(List nodes) { + super(nodes); + } + + @Override + public boolean isUpdateClusterMetadataDue(long now) { + return true; + } + + @Override + public void handleServerDisconnect(long now, String destinationId, Optional maybeAuthException) { + maybeAuthException.ifPresent(exception -> { + failure = exception; + }); + super.handleServerDisconnect(now, destinationId, maybeAuthException); + } + + @Override + public void handleFailedRequest(long now, Optional maybeFatalException) { + maybeFatalException.ifPresent(exception -> { + failure = exception; + }); + } + + public KafkaException getAndClearFailure() { + KafkaException failure = this.failure; + this.failure = null; + return failure; + } + } }