This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new da3b5e78ed2 MINOR: Document connections.max.idle.ms dependency on
max.poll.interval.ms for Classic consumer (#22752)
da3b5e78ed2 is described below
commit da3b5e78ed2b3d5a9efec85760df7ff2ea6ad2cb
Author: TengYao Chi <[email protected]>
AuthorDate: Tue Jul 14 17:51:28 2026 +0100
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 <[email protected]>
---
.../apache/kafka/clients/CommonClientConfigs.java | 10 ++++++++--
.../kafka/clients/consumer/ConsumerConfig.java | 21 +++++++++++++++++++++
.../runtime/distributed/DistributedConfig.java | 13 +++++++++++++
3 files changed, 42 insertions(+), 2 deletions(-)
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 33b936b378d..20d7465f113 100644
--- a/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java
+++ b/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 "
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 80130cd7188..3a4e6ceee18 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
@@ -38,6 +38,9 @@ import org.apache.kafka.common.security.auth.SecurityProtocol;
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 @@ import static
org.apache.kafka.common.config.ConfigDef.ValidString.in;
*/
@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 @@ public class ConsumerConfig extends AbstractConfig {
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()) {
diff --git
a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedConfig.java
b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedConfig.java
index 9ff90e7a516..c417371f731 100644
---
a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedConfig.java
+++
b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedConfig.java
@@ -598,9 +598,22 @@ public final class DistributedConfig extends WorkerConfig {
@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);
}
