KAFKA-18185: remove internal.leave.group.on.close config (#19400)

frankvicky · web-flow · commit 68f1da847400 · 2025-10-03T13:21:38.000-04:00
JIRA: KAFKA-18185 This is a follow-up of #17614 The patch is to remove the `internal.leave.group.on.close` config. Reviewers: Sophie Blee-Goldman <ableegoldman@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>, Bill Bejeck <bbejeck@apache.org>
diff --git a/clients/src/main/java/org/apache/kafka/clients/GroupRebalanceConfig.java b/clients/src/main/java/org/apache/kafka/clients/GroupRebalanceConfig.java
@@ -46,7 +46,6 @@ public String toString() {
     public final Optional<String> rackId;
     public final long retryBackoffMs;
     public final long retryBackoffMaxMs;
-    public final boolean leaveGroupOnClose;
 
     public GroupRebalanceConfig(AbstractConfig config, ProtocolType protocolType) {
         this.sessionTimeoutMs = config.getInt(CommonClientConfigs.SESSION_TIMEOUT_MS_CONFIG);
@@ -80,13 +79,6 @@ public GroupRebalanceConfig(AbstractConfig config, ProtocolType protocolType) {
 
         this.retryBackoffMs = config.getLong(CommonClientConfigs.RETRY_BACKOFF_MS_CONFIG);
         this.retryBackoffMaxMs = config.getLong(CommonClientConfigs.RETRY_BACKOFF_MAX_MS_CONFIG);
-
-        // Internal leave group config is only defined in Consumer.
-        if (protocolType == ProtocolType.CONSUMER) {
-            this.leaveGroupOnClose = config.getBoolean("internal.leave.group.on.close");
-        } else {
-            this.leaveGroupOnClose = true;
-        }
     }
 
     // For testing purpose.
@@ -97,8 +89,7 @@ public GroupRebalanceConfig(final int sessionTimeoutMs,
                                 Optional<String> groupInstanceId,
                                 String rackId,
                                 long retryBackoffMs,
-                                long retryBackoffMaxMs,
-                                boolean leaveGroupOnClose) {
+                                long retryBackoffMaxMs) {
         this.sessionTimeoutMs = sessionTimeoutMs;
         this.rebalanceTimeoutMs = rebalanceTimeoutMs;
         this.heartbeatIntervalMs = heartbeatIntervalMs;
@@ -107,6 +98,5 @@ public GroupRebalanceConfig(final int sessionTimeoutMs,
         this.rackId = rackId == null || rackId.isEmpty() ? Optional.empty() : Optional.of(rackId);
         this.retryBackoffMs = retryBackoffMs;
         this.retryBackoffMaxMs = retryBackoffMaxMs;
-        this.leaveGroupOnClose = leaveGroupOnClose;
     }
 }
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
@@ -330,17 +330,6 @@ public class ConsumerConfig extends AbstractConfig {
             "be excluded from the subscription. It is always possible to explicitly subscribe to an internal topic.";
     public static final boolean DEFAULT_EXCLUDE_INTERNAL_TOPICS = true;
 
-    /**
-     * <code>internal.leave.group.on.close</code>
-     * Whether or not the consumer should leave the group on close. If set to <code>false</code> then a rebalance
-     * won't occur until <code>session.timeout.ms</code> expires.
-     *
-     * <p>
-     * Note: this is an internal configuration and could be changed in the future in a backward incompatible way
-     *
-     */
-    static final String LEAVE_GROUP_ON_CLOSE_CONFIG = "internal.leave.group.on.close";
-
     /**
      * <code>internal.throw.on.fetch.stable.offset.unsupported</code>
      * Whether or not the consumer should throw when the new stable offset feature is supported.
@@ -634,10 +623,6 @@ public class ConsumerConfig extends AbstractConfig {
                                         DEFAULT_EXCLUDE_INTERNAL_TOPICS,
                                         Importance.MEDIUM,
                                         EXCLUDE_INTERNAL_TOPICS_DOC)
-                                .defineInternal(LEAVE_GROUP_ON_CLOSE_CONFIG,
-                                        Type.BOOLEAN,
-                                        true,
-                                        Importance.LOW)
                                 .defineInternal(THROW_ON_FETCH_STABLE_OFFSET_UNSUPPORTED,
                                         Type.BOOLEAN,
                                         false,
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
@@ -1170,7 +1170,7 @@ protected void handlePollTimeoutExpiry() {
     public synchronized RequestFuture<Void> maybeLeaveGroup(CloseOptions.GroupMembershipOperation membershipOperation, String leaveReason) {
         RequestFuture<Void> future = null;
 
-        if (rebalanceConfig.leaveGroupOnClose && shouldSendLeaveGroupRequest(membershipOperation)) {
+        if (shouldSendLeaveGroupRequest(membershipOperation)) {
             log.info("Member {} sending LeaveGroup request to coordinator {} due to {}",
                 generation.memberId, coordinator, leaveReason);
             LeaveGroupRequest.Builder request = new LeaveGroupRequest.Builder(
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ClassicKafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ClassicKafkaConsumer.java
@@ -331,8 +331,7 @@ public class ClassicKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
                 groupInstanceId,
                 rackId,
                 retryBackoffMs,
-                retryBackoffMaxMs,
-                true
+                retryBackoffMaxMs
             );
             this.coordinator = new ConsumerCoordinator(
                 rebalanceConfig,
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerMembershipManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerMembershipManager.java
@@ -421,7 +421,7 @@ private void logPausedPartitionsBeingRevoked(Set<TopicPartition> partitionsToRev
     @Override
     public boolean isLeavingGroup() {
         CloseOptions.GroupMembershipOperation leaveGroupOperation = leaveGroupOperation();
-        if (REMAIN_IN_GROUP == leaveGroupOperation) {
+        if (REMAIN_IN_GROUP == leaveGroupOperation && groupInstanceId.isEmpty()) {
             return false;
         }
 
@@ -432,7 +432,8 @@ public boolean isLeavingGroup() {
         boolean hasLeaveOperation = DEFAULT == leaveGroupOperation ||
             // Leave operation: both static and dynamic consumers will send a leave heartbeat
             LEAVE_GROUP == leaveGroupOperation ||
-            // Remain in group: only static consumers will send a leave heartbeat, while dynamic members will not
+            // Remain in group: static consumers will send a leave heartbeat with -2 epoch to reflect that a member using the given
+            // instance id decided to leave the group and would be back within the session timeout.
             groupInstanceId().isPresent();
 
         return isLeavingState && hasLeaveOperation;
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java
@@ -135,12 +135,7 @@ private void setupCoordinator(int retryBackoffMs, int retryBackoffMaxMs) {
             Optional.empty(), Optional.empty());
     }
 
-
     private void setupCoordinator(int retryBackoffMs, int retryBackoffMaxMs, int rebalanceTimeoutMs, Optional<String> groupInstanceId, Optional<Supplier<BaseHeartbeatThread>> heartbeatThreadSupplier) {
-        setupCoordinator(retryBackoffMs, retryBackoffMaxMs, rebalanceTimeoutMs, groupInstanceId, heartbeatThreadSupplier, groupInstanceId.isEmpty());
-    }
-
-    private void setupCoordinator(int retryBackoffMs, int retryBackoffMaxMs, int rebalanceTimeoutMs, Optional<String> groupInstanceId, Optional<Supplier<BaseHeartbeatThread>> heartbeatThreadSupplier, boolean leaveOnClose) {
         LogContext logContext = new LogContext();
         this.mockTime = new MockTime();
         ConsumerMetadata metadata = new ConsumerMetadata(retryBackoffMs, retryBackoffMaxMs, 60 * 60 * 1000L,
@@ -168,8 +163,7 @@ false, false, new SubscriptionState(logContext, AutoOffsetResetStrategy.EARLIEST
                                                                         groupInstanceId,
                                                                         null,
                                                                         retryBackoffMs,
-                                                                        retryBackoffMaxMs,
-                                                                        leaveOnClose);
+                                                                        retryBackoffMaxMs);
         this.coordinator = new DummyCoordinator(rebalanceConfig,
                                                 consumerClient,
                                                 metrics,
@@ -1109,7 +1103,7 @@ public void testLeaveGroupSentWithGroupInstanceIdUnSet() {
     @ParameterizedTest
     @MethodSource("groupInstanceIdAndMembershipOperationMatrix")
     public void testLeaveGroupSentWithGroupInstanceIdUnSetAndDifferentGroupMembershipOperation(Optional<String> groupInstanceId, CloseOptions.GroupMembershipOperation operation) {
-        checkLeaveGroupRequestSent(groupInstanceId, operation, Optional.empty(), true);
+        checkLeaveGroupRequestSent(groupInstanceId, operation, Optional.empty());
     }
 
     private static Stream<Arguments> groupInstanceIdAndMembershipOperationMatrix() {
@@ -1124,11 +1118,11 @@ private static Stream<Arguments> groupInstanceIdAndMembershipOperationMatrix() {
     }
 
     private void checkLeaveGroupRequestSent(Optional<String> groupInstanceId)  {
-        checkLeaveGroupRequestSent(groupInstanceId, CloseOptions.GroupMembershipOperation.DEFAULT, Optional.empty(), groupInstanceId.isEmpty());
+        checkLeaveGroupRequestSent(groupInstanceId, CloseOptions.GroupMembershipOperation.DEFAULT, Optional.empty());
     }
 
-    private void checkLeaveGroupRequestSent(Optional<String> groupInstanceId, CloseOptions.GroupMembershipOperation operation, Optional<Supplier<BaseHeartbeatThread>> heartbeatThreadSupplier, boolean leaveOnClose) {
-        setupCoordinator(RETRY_BACKOFF_MS, RETRY_BACKOFF_MAX_MS, Integer.MAX_VALUE, groupInstanceId, heartbeatThreadSupplier, leaveOnClose);
+    private void checkLeaveGroupRequestSent(Optional<String> groupInstanceId, CloseOptions.GroupMembershipOperation operation, Optional<Supplier<BaseHeartbeatThread>> heartbeatThreadSupplier) {
+        setupCoordinator(RETRY_BACKOFF_MS, RETRY_BACKOFF_MAX_MS, Integer.MAX_VALUE, groupInstanceId, heartbeatThreadSupplier);
 
         mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));
         mockClient.prepareResponse(joinGroupFollowerResponse(1, memberId, leaderId, Errors.NONE));
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
@@ -224,8 +224,7 @@ private GroupRebalanceConfig buildRebalanceConfig(Optional<String> groupInstance
                                         groupInstanceId,
                                         rackId,
                                         retryBackoffMs,
-                                        retryBackoffMaxMs,
-                                        groupInstanceId.isEmpty());
+                                        retryBackoffMaxMs);
     }
 
     @AfterEach
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatTest.java
@@ -47,8 +47,7 @@ public void setUp() {
                                                                         Optional.empty(),
                                                                         null,
                                                                         retryBackoffMs,
-                                                                        retryBackoffMaxMs,
-                                                                        true);
+                                                                        retryBackoffMaxMs);
         heartbeat = new Heartbeat(rebalanceConfig, time);
     }
 
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinatorIncrementalTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinatorIncrementalTest.java
@@ -149,8 +149,7 @@ public void init(ConnectProtocolCompatibility compatibility) {
             Optional.empty(),
             null,
             retryBackoffMs,
-            retryBackoffMaxMs,
-            true);
+            retryBackoffMaxMs);
         this.coordinator = new WorkerCoordinator(rebalanceConfig,
             loggerFactory,
             consumerClient,
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinatorTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinatorTest.java
@@ -140,8 +140,7 @@ public void setup(ConnectProtocolCompatibility compatibility) {
                                                         Optional.empty(),
                                                         null,
                                                         retryBackoffMs,
-                                                        retryBackoffMaxMs,
-                                                        true);
+                                                        retryBackoffMaxMs);
         this.coordinator = new WorkerCoordinator(rebalanceConfig,
                                                  logContext,
                                                  consumerClient,
diff --git a/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala
@@ -24,7 +24,7 @@ import java.lang.{Long => JLong}
 import java.time.{Duration => JDuration}
 import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger}
 import java.util.concurrent.{CountDownLatch, ExecutionException, TimeUnit}
-import java.util.{Collections, Locale, Optional, Properties}
+import java.util.{Collections, Optional, Properties}
 import java.{time, util}
 import kafka.integration.KafkaServerTestHarness
 import kafka.server.KafkaConfig
@@ -2355,9 +2355,6 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
       defaultConsumerConfig.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
       defaultConsumerConfig.setProperty(ConsumerConfig.GROUP_ID_CONFIG, testGroupId)
       defaultConsumerConfig.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, testClientId)
-      // We need to set internal.leave.group.on.close to validate dynamic member removal, but it only works for ClassicConsumer
-      // After KIP-1092, we can control dynamic member removal for both ClassicConsumer and AsyncConsumer
-      defaultConsumerConfig.setProperty("internal.leave.group.on.close", "false")
 
       val backgroundConsumerSet = new BackgroundConsumerSet(defaultConsumerConfig)
       groupInstanceSet.zip(topicSet).foreach { case (groupInstanceId, topic) =>
@@ -2406,14 +2403,7 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
         var testGroupDescription = describeTestGroupResult.describedGroups().get(testGroupId).get()
         assertEquals(testGroupId, testGroupDescription.groupId)
         assertFalse(testGroupDescription.isSimpleConsumerGroup)
-
-        // Although we set `internal.leave.group.on.close` in the consumer, it only works for ClassicConsumer.
-        // After KIP-1092, we can control dynamic member removal in consumer.close()
-        if (groupProtocol == GroupProtocol.CLASSIC.name.toLowerCase(Locale.ROOT)) {
-          assertEquals(3, testGroupDescription.members().size())
-        } else if (groupProtocol == GroupProtocol.CONSUMER.name.toLowerCase(Locale.ROOT)) {
-          assertEquals(2, testGroupDescription.members().size())
-        }
+        assertEquals(2, testGroupDescription.members().size())
 
         // Test delete one static member
         removeMembersResult = client.removeMembersFromConsumerGroup(testGroupId,
@@ -2426,11 +2416,7 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
           new DescribeConsumerGroupsOptions().includeAuthorizedOperations(true))
         testGroupDescription = describeTestGroupResult.describedGroups().get(testGroupId).get()
 
-        if (groupProtocol == GroupProtocol.CLASSIC.name.toLowerCase(Locale.ROOT)) {
-          assertEquals(2, testGroupDescription.members().size())
-        } else if (groupProtocol == GroupProtocol.CONSUMER.name.toLowerCase(Locale.ROOT)) {
-          assertEquals(1, testGroupDescription.members().size())
-        }
+        assertEquals(1, testGroupDescription.members().size())
 
         // Delete all active members remaining
         removeMembersResult = client.removeMembersFromConsumerGroup(testGroupId, new RemoveMembersFromConsumerGroupOptions())
diff --git a/docs/upgrade.html b/docs/upgrade.html
@@ -36,6 +36,9 @@ <h5><a id="upgrade_420_notable" href="#upgrade_420_notable">Notable changes in 4
         The <code>KafkaPrincipalBuilder</code> now extends <code>KafkaPrincipalSerde</code>. Force developer to implement <code>KafkaPrincipalSerde</code> interface for custom <code>KafkaPrincipalBuilder</code>.
         For further details, please refer to <a href="https://cwiki.apache.org/confluence/x/1gq9F">KIP-1157</a>.
     </li>
+    <li>The behavior of <code>org.apache.kafka.streams.KafkaStreams#removeStreamThread</code> has been changed. The consumer has no longer remove once <code>removeStreamThread</code> finished.
+        Instead, consumer would be kicked off from the group after <code>org.apache.kafka.streams.processor.internals.StreamThread</code> completes its <code>run</code> function.
+    </li>
     <li>
         The support for MX4J library, enabled through <code>kafka_mx4jenable</code> system property, was deprecated and will be removed in Kafka 5.0.
     </li>
diff --git a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
diff --git a/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java b/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java
diff --git a/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java b/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java

Original file line number	Diff line number	Diff line change
`@@ -224,8 +224,7 @@ private GroupRebalanceConfig buildRebalanceConfig(Optional<String> groupInstance`
`224`	`224`	`groupInstanceId,`
`225`	`225`	`rackId,`
`226`	`226`	`retryBackoffMs,`
`227`		`- retryBackoffMaxMs,`
`228`		`- groupInstanceId.isEmpty());`
	`227`	`+ retryBackoffMaxMs);`
`229`	`228`	`}`
`230`	`229`
`231`	`230`	`@AfterEach`
Original file line number	Diff line number	Diff line change
`@@ -47,8 +47,7 @@ public void setUp() {`
`47`	`47`	`Optional.empty(),`
`48`	`48`	`null,`
`49`	`49`	`retryBackoffMs,`
`50`		`- retryBackoffMaxMs,`
`51`		`- true);`
	`50`	`+ retryBackoffMaxMs);`
`52`	`51`	`heartbeat = new Heartbeat(rebalanceConfig, time);`
`53`	`52`	`}`
`54`	`53`