MINOR: Document connections.max.idle.ms dependency on max.poll.interval.ms for Classic consumer (#22752)

In the Classic consumer, `connections.max.idle.ms` can cause the
coordinator connection to be reaped during a long rebalance if
`max.poll.interval.ms` is higher than it. When this happens, in-flight
`JoinGroup` requests are cancelled and the group cycles through
connection reaping, which prolongs or destabilizes the rebalance —
particularly in large consumer groups where the coordinator waits longer
for all members to rejoin.

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>

Author: frankvicky

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

@@ -196,13 +196,19 @@ public class CommonClientConfigs {
                                                           + "For consumers using a non-null <code>group.instance.id</code> which reach this timeout, partitions will not be immediately reassigned. "
                                                           + "Instead, the consumer will stop sending heartbeats and partitions will be reassigned "
                                                           + "after expiration of the session timeout (defined by the client config <code>session.timeout.ms</code> if using the Classic rebalance protocol, or by the broker config <code>group.consumer.session.timeout.ms</code> if using the Consumer protocol). "
-                                                          + "This mirrors the behavior of a static consumer which has shutdown.";
+                                                          + "This mirrors the behavior of a static consumer which has shutdown. "
+                                                          + "In the Classic consumer, <code>connections.max.idle.ms</code> should be set to a value greater than or equal to this timeout "
+                                                          + "to avoid closing the connection to the group coordinator during an ongoing rebalance. "
+                                                          + "This is particularly important in large groups, where a rebalance may take longer to complete "
+                                                          + "while the coordinator waits for all members to rejoin.";
 
     public static final String REBALANCE_TIMEOUT_MS_CONFIG = "rebalance.timeout.ms";
     public static final String REBALANCE_TIMEOUT_MS_DOC = "The maximum allowed time for each worker to join the group "
                                                           + "once a rebalance has begun. This is basically a limit on the amount of time needed for all tasks to "
                                                           + "flush any pending data and commit offsets. If the timeout is exceeded, then the worker will be removed "
-                                                          + "from the group, which will cause offset commit failures.";
+                                                          + "from the group, which will cause offset commit failures. "
+                                                          + "<code>connections.max.idle.ms</code> should be set to a value greater than or equal to this timeout "
+                                                          + "to avoid closing the connection to the group coordinator during an ongoing rebalance.";
 
     public static final String SESSION_TIMEOUT_MS_CONFIG = "session.timeout.ms";
     public static final String SESSION_TIMEOUT_MS_DOC = "The timeout used to detect client failures when using "

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

@@ -38,6 +38,9 @@
 import org.apache.kafka.common.serialization.Deserializer;
 import org.apache.kafka.common.utils.Utils;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -60,6 +63,7 @@
  */
 @InterfaceAudience.Public
 public class ConsumerConfig extends AbstractConfig {
+    private static final Logger log = LoggerFactory.getLogger(ConsumerConfig.class);
     private static final ConfigDef CONFIG;
 
     // a list contains all the assignor names that only assign subscribed topics to consumer. Should be updated when new assignor added.
@@ -724,9 +728,26 @@ protected Map<String, Object> postProcessParsedConfig(final Map<String, Object>
         maybeOverrideClientId(refinedConfigs);
         maybeOverrideEnableAutoCommit(refinedConfigs);
         checkUnsupportedConfigsPostProcess();
+        warnIfConnectionsMaxIdleMsLowerThanMaxPollIntervalMs();
         return refinedConfigs;
     }
 
+    private void warnIfConnectionsMaxIdleMsLowerThanMaxPollIntervalMs() {
+        String groupProtocol = getString(GROUP_PROTOCOL_CONFIG);
+        if (!GroupProtocol.CLASSIC.name().equalsIgnoreCase(groupProtocol)) {
+            return;
+        }
+        long connectionsMaxIdleMs = getLong(CONNECTIONS_MAX_IDLE_MS_CONFIG);
+        int maxPollIntervalMs = getInt(MAX_POLL_INTERVAL_MS_CONFIG);
+        if (connectionsMaxIdleMs >= 0 && connectionsMaxIdleMs < maxPollIntervalMs) {
+            log.warn("Configuration '{}' with value '{}' is lower than configuration '{}' with value '{}'. " +
+                    "This may cause the connection to the group coordinator to be closed during an ongoing rebalance, " +
+                    "which can prolong or disrupt group rejoin.",
+                CONNECTIONS_MAX_IDLE_MS_CONFIG, connectionsMaxIdleMs,
+                MAX_POLL_INTERVAL_MS_CONFIG, maxPollIntervalMs);
+        }
+    }
+
     private void maybeOverrideClientId(Map<String, Object> configs) {
         final String clientId = this.getString(CLIENT_ID_CONFIG);
         if (clientId == null || clientId.isEmpty()) {

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedConfig.java

@@ -598,9 +598,22 @@ public String groupId() {
     @Override
     protected Map<String, Object> postProcessParsedConfig(final Map<String, Object> parsedValues) {
         CommonClientConfigs.warnDisablingExponentialBackoff(this);
+        warnIfConnectionsMaxIdleMsLowerThanRebalanceTimeoutMs();
         return super.postProcessParsedConfig(parsedValues);
     }
 
+    private void warnIfConnectionsMaxIdleMsLowerThanRebalanceTimeoutMs() {
+        long connectionsMaxIdleMs = getLong(CommonClientConfigs.CONNECTIONS_MAX_IDLE_MS_CONFIG);
+        int rebalanceTimeoutMs = getInt(REBALANCE_TIMEOUT_MS_CONFIG);
+        if (connectionsMaxIdleMs >= 0 && connectionsMaxIdleMs < rebalanceTimeoutMs) {
+            log.warn("Configuration '{}' with value '{}' is lower than configuration '{}' with value '{}'. " +
+                    "This may cause the connection to the group coordinator to be closed during an ongoing rebalance, " +
+                    "which can prolong or disrupt group rejoin.",
+                CommonClientConfigs.CONNECTIONS_MAX_IDLE_MS_CONFIG, connectionsMaxIdleMs,
+                REBALANCE_TIMEOUT_MS_CONFIG, rebalanceTimeoutMs);
+        }
+    }
+
     public DistributedConfig(Map<String, String> props) {
         this(Crypto.SYSTEM, props);
     }