KAFKA-18185: remove internal.leave.group.on.close config

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

@@ -45,7 +45,6 @@ public String toString() {
     public final Optional<String> groupInstanceId;
     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);
@@ -75,13 +74,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.
@@ -91,15 +83,13 @@ public GroupRebalanceConfig(final int sessionTimeoutMs,
                                 String groupId,
                                 Optional<String> groupInstanceId,
                                 long retryBackoffMs,
-                                long retryBackoffMaxMs,
-                                boolean leaveGroupOnClose) {
+                                long retryBackoffMaxMs) {
         this.sessionTimeoutMs = sessionTimeoutMs;
         this.rebalanceTimeoutMs = rebalanceTimeoutMs;
         this.heartbeatIntervalMs = heartbeatIntervalMs;
         this.groupId = groupId;
         this.groupInstanceId = groupInstanceId;
         this.retryBackoffMs = retryBackoffMs;
         this.retryBackoffMaxMs = retryBackoffMaxMs;
-        this.leaveGroupOnClose = leaveGroupOnClose;
     }
 }

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

@@ -329,17 +329,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.
@@ -637,10 +626,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,

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(

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> {
                 groupId.get(),
                 groupInstanceId,
                 retryBackoffMs,
-                retryBackoffMaxMs,
-                true
+                retryBackoffMaxMs
             );
             this.coordinator = new ConsumerCoordinator(
                 rebalanceConfig,

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,
@@ -167,8 +162,7 @@ false, false, new SubscriptionState(logContext, AutoOffsetResetStrategy.EARLIEST
                                                                         GROUP_ID,
                                                                         groupInstanceId,
                                                                         retryBackoffMs,
-                                                                        retryBackoffMaxMs,
-                                                                        leaveOnClose);
+                                                                        retryBackoffMaxMs);
         this.coordinator = new DummyCoordinator(rebalanceConfig,
                                                 consumerClient,
                                                 metrics,
@@ -1108,7 +1102,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() {
@@ -1123,11 +1117,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));

clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java

@@ -223,8 +223,7 @@ private GroupRebalanceConfig buildRebalanceConfig(Optional<String> groupInstance
                                         groupId,
                                         groupInstanceId,
                                         retryBackoffMs,
-                                        retryBackoffMaxMs,
-                                        groupInstanceId.isEmpty());
+                                        retryBackoffMaxMs);
     }
 
     @AfterEach

clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatTest.java

@@ -46,8 +46,7 @@ public void setUp() {
                                                                         "group_id",
                                                                         Optional.empty(),
                                                                         retryBackoffMs,
-                                                                        retryBackoffMaxMs,
-                                                                        true);
+                                                                        retryBackoffMaxMs);
         heartbeat = new Heartbeat(rebalanceConfig, time);
     }
 

connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinatorIncrementalTest.java

@@ -150,8 +150,7 @@ public void init(ConnectProtocolCompatibility compatibility) {
             groupId,
             Optional.empty(),
             retryBackoffMs,
-            retryBackoffMaxMs,
-            true);
+            retryBackoffMaxMs);
         this.coordinator = new WorkerCoordinator(rebalanceConfig,
             loggerFactory,
             consumerClient,

connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinatorTest.java

@@ -140,8 +140,7 @@ public void setup(ConnectProtocolCompatibility compatibility) {
                                                         groupId,
                                                         Optional.empty(),
                                                         retryBackoffMs,
-                                                        retryBackoffMaxMs,
-                                                        true);
+                                                        retryBackoffMaxMs);
         this.coordinator = new WorkerCoordinator(rebalanceConfig,
                                                  logContext,
                                                  consumerClient,

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

@@ -25,7 +25,7 @@ import java.time.{Duration => JDuration}
 import java.util.Arrays.asList
 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
@@ -2331,9 +2331,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) =>
@@ -2382,14 +2379,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,
@@ -2402,11 +2392,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())