[LI-HOTFIX] revert sticky metadata fetch hotfix

clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java

@@ -147,9 +147,6 @@ public class CommonClientConfigs {
                                                            + "consumer's session stays active and to facilitate rebalancing when new consumers join or leave the group. "
                                                            + "The value must be set lower than <code>session.timeout.ms</code>, but typically should be set no higher "
                                                            + "than 1/3 of that value. It can be adjusted even lower to control the expected time for normal rebalances.";
-    public static final String ENABLE_STICKY_METADATA_FETCH_CONFIG = "enable.sticky.metadata.fetch";
-    public static final String ENABLE_STICKY_METADATA_FETCH_DOC = "Fetch metadata from the least loaded broker if false. Otherwise fetch metadata "
-                                                         + "from the same broker until it is disconnected.";
 
     /**
      * Postprocess the configuration so that exponential backoff is disabled when reconnect backoff

clients/src/main/java/org/apache/kafka/clients/NetworkClient.java

@@ -115,8 +115,6 @@ private enum State {
 
     private final Time time;
 
-    private boolean enableStickyMetadataFetch = true;
-
     /**
      * True if we should send an ApiVersionRequest when first connecting to a broker.
      */
@@ -275,10 +273,6 @@ private NetworkClient(MetadataUpdater metadataUpdater,
         this.state = new AtomicReference<>(State.ACTIVE);
     }
 
-    public void setEnableStickyMetadataFetch(boolean enableStickyMetadataFetch) {
-        this.enableStickyMetadataFetch = enableStickyMetadataFetch;
-    }
-
     /**
      * Begin connecting to the given node, return true if we are already connected and ready to send to that node.
      *
@@ -577,12 +571,6 @@ public List<ClientResponse> poll(long timeout, long now) {
         handleTimedOutRequests(responses, updatedNow);
         completeResponses(responses);
 
-        // We changed the metadataUpdater.maybeUpdate() such that it will keep sending MetadataRequest
-        // to the same broker instead choosing the least loaded node. If we don't try to send metadata here, it is possible that
-        // another request is sent to the broker before the next networkClient.poll(). This can cause starvation
-        // for the MetadataRequest and consumer's metadata may be stale for a long time.
-        metadataUpdater.maybeUpdate(updatedNow);
-
         return responses;
     }
 
@@ -994,16 +982,13 @@ class DefaultMetadataUpdater implements MetadataUpdater {
 
         /* the current cluster metadata */
         private final Metadata metadata;
-        // Consumer needs to keep fetching metadata from the same node until that node goes down
-        private Node nodeToFetchMetadata;
 
         // Defined if there is a request in progress, null otherwise
         private Integer inProgressRequestVersion;
 
         DefaultMetadataUpdater(Metadata metadata) {
             this.metadata = metadata;
             this.inProgressRequestVersion = null;
-            this.nodeToFetchMetadata = null;
         }
 
         @Override
@@ -1033,15 +1018,13 @@ public long maybeUpdate(long now) {
 
             // Beware that the behavior of this method and the computation of timeouts for poll() are
             // highly dependent on the behavior of leastLoadedNode.
-            if (!enableStickyMetadataFetch || nodeToFetchMetadata == null || !connectionStates.isReady(nodeToFetchMetadata.idString(), now))
-                nodeToFetchMetadata = leastLoadedNode(now);
-
-            if (nodeToFetchMetadata == null) {
+            Node node = leastLoadedNode(now);
+            if (node == null) {
                 log.debug("Give up sending metadata request since no node is available");
                 return reconnectBackoffMs;
             }
 
-            return maybeUpdate(now, nodeToFetchMetadata);
+            return maybeUpdate(now, node);
         }
 
         @Override

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

@@ -524,11 +524,6 @@ public class ConsumerConfig extends AbstractConfig {
                                         CommonClientConfigs.DEFAULT_SECURITY_PROTOCOL,
                                         Importance.MEDIUM,
                                         CommonClientConfigs.SECURITY_PROTOCOL_DOC)
-                                .define(CommonClientConfigs.ENABLE_STICKY_METADATA_FETCH_CONFIG,
-                                        Type.BOOLEAN,
-                                        true,
-                                        Importance.MEDIUM,
-                                        CommonClientConfigs.ENABLE_STICKY_METADATA_FETCH_DOC)
                                 .withClientSslSupport()
                                 .withClientSaslSupport();
     }

clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java

@@ -762,7 +762,6 @@ else if (enableAutoCommit)
                     apiVersions,
                     throttleTimeSensor,
                     logContext);
-            netClient.setEnableStickyMetadataFetch(config.getBoolean(CommonClientConfigs.ENABLE_STICKY_METADATA_FETCH_CONFIG));
             this.client = new ConsumerNetworkClient(
                     logContext,
                     netClient,

clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java

@@ -373,11 +373,6 @@ public class ProducerConfig extends AbstractConfig {
                                         null,
                                         Importance.LOW,
                                         SECURITY_PROVIDERS_DOC)
-                                .define(CommonClientConfigs.ENABLE_STICKY_METADATA_FETCH_CONFIG,
-                                        Type.BOOLEAN,
-                                        false,
-                                        Importance.MEDIUM,
-                                        CommonClientConfigs.ENABLE_STICKY_METADATA_FETCH_DOC)
                                 .withClientSslSupport()
                                 .withClientSaslSupport()
                                 .define(ENABLE_IDEMPOTENCE_CONFIG,