[LI-HOTFIX] Resolve the bootstrap server when cluster metadata hasn't refreshed for a long time

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.

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;

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<TopicPartition, Integer> 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) {

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).

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<ClientResponse> abortedSends = new LinkedList<>();
 
+    private final List<String> bootstrapServers = new LinkedList<>();
+
     private final Sensor throttleTimeSensor;
 
     private final AtomicReference<State> 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<String> 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<String> 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<String> 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<String> 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<ClientResponse> poll(long timeout, long now) {
         long updatedNow = this.time.milliseconds();
         List<ClientResponse> 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<InetSocketAddress> addresses = ClientUtils.parseAndValidateAddresses(
+                this.bootstrapServers,
+                this.clientDnsLookup.toString());
+            List<Node> 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();

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) {

clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminMetadataManager.java

@@ -90,6 +90,11 @@ public List<Node> fetchNodes() {
             return cluster.nodes();
         }
 
+        @Override
+        public boolean isUpdateClusterMetadataDue(long now) {
+            return false;
+        }
+
         @Override
         public boolean isUpdateDue(long now) {
             return false;