This is an automated email from the ASF dual-hosted git repository.
lianetm 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 096cc96fee5 KAFKA-20522: Refine test coverage for consumer
close-on-interrupt best-effort behavior. (#22138)
096cc96fee5 is described below
commit 096cc96fee5edd924a9c685b2c0084f320914135
Author: ChickenchickenLove <[email protected]>
AuthorDate: Wed Jun 17 03:58:45 2026 +0900
KAFKA-20522: Refine test coverage for consumer close-on-interrupt
best-effort behavior. (#22138)
### Description
In the previous,
`PlaintextConsumerTest.testClassicConsumerCloseLeavesGroupOnInterrupt`
were flaky.
The reason is that `LeaveGroup` on consumer close has best-effort
semantics, but those tests were written under the assumption that the
consumer deterministically leaves the consumer group. In practice,
however, the consumer may fail to leave the group within the bounded
timeout, for example due to metadata expiration.
(https://github.com/apache/kafka/pull/21332)
This PR splits the flaky test into two separate tests.
1. The existing integration test now verifies that `callsToRevoked` is
invoked when `interrupt()` is called. This behavior is still
deterministic even under best-effort semantics.
2. A unit test verifies that an `ApiKeys.LEAVE_GROUP` request is
recorded in `MockClient.requests()` when `interrupt()` is called.
Previously, even if an `ApiKeys.LEAVE_GROUP` request was created,
request processing could be delayed due to metadata expiration. As a
result, the exact point at which the consumer leaves the group was
non-deterministic. Therefore, taking the best-effort semantics into
account, the unit test only verifies that the `ApiKeys.LEAVE_GROUP`
request is recorded.
### Related
- https://github.com/apache/kafka/pull/21332
- KAFKA-20522
- KAFKA-17397
- KAFKA-18031
Reviewers: Lianet Magrans <[email protected]>
---
.../clients/consumer/PlaintextConsumerTest.java | 75 ----------------------
.../kafka/clients/consumer/KafkaConsumerTest.java | 54 ++++++++++++++++
.../consumer/internals/AsyncKafkaConsumerTest.java | 60 +++++++++++++++++
.../kafka/api/PlaintextConsumerTest.scala | 75 ----------------------
4 files changed, 114 insertions(+), 150 deletions(-)
diff --git
a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerTest.java
b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerTest.java
index 6a1b174d188..e5a2c8a831f 100644
---
a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerTest.java
+++
b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerTest.java
@@ -27,7 +27,6 @@ import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.config.TopicConfig;
-import org.apache.kafka.common.errors.InterruptException;
import org.apache.kafka.common.errors.InvalidGroupIdException;
import org.apache.kafka.common.errors.InvalidTopicException;
import org.apache.kafka.common.errors.TimeoutException;
@@ -46,7 +45,6 @@ import org.apache.kafka.common.test.api.ClusterConfigProperty;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestDefaults;
import org.apache.kafka.common.test.api.ClusterTests;
-import org.apache.kafka.common.test.api.Flaky;
import org.apache.kafka.common.test.api.Type;
import org.apache.kafka.server.quota.QuotaType;
import org.apache.kafka.test.MockConsumerInterceptor;
@@ -63,7 +61,6 @@ import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
@@ -1616,78 +1613,6 @@ public class PlaintextConsumerTest {
}
}
- @Flaky("KAFKA-18031")
- @ClusterTest
- public void testClassicConsumerCloseLeavesGroupOnInterrupt() throws
Exception {
- testCloseLeavesGroupOnInterrupt(Map.of(
- GROUP_PROTOCOL_CONFIG,
GroupProtocol.CLASSIC.name().toLowerCase(Locale.ROOT),
- KEY_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class.getName(),
- VALUE_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class.getName(),
- AUTO_OFFSET_RESET_CONFIG, "earliest",
- GROUP_ID_CONFIG, "group_test,",
- BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers()
- ));
- }
-
- @Flaky("KAFKA-18031")
- @ClusterTest
- public void testAsyncConsumerCloseLeavesGroupOnInterrupt() throws
Exception {
- testCloseLeavesGroupOnInterrupt(Map.of(
- GROUP_PROTOCOL_CONFIG,
GroupProtocol.CONSUMER.name().toLowerCase(Locale.ROOT),
- KEY_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class.getName(),
- VALUE_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class.getName(),
- AUTO_OFFSET_RESET_CONFIG, "earliest",
- GROUP_ID_CONFIG, "group_test,",
- BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers()
- ));
- }
-
- private void testCloseLeavesGroupOnInterrupt(Map<String, Object>
consumerConfig) throws Exception {
- try (Consumer<byte[], byte[]> consumer =
cluster.consumer(consumerConfig)) {
- var listener = new TestConsumerReassignmentListener();
- consumer.subscribe(List.of(TOPIC), listener);
- awaitRebalance(consumer, listener);
-
- assertEquals(1, listener.callsToAssigned);
- assertEquals(0, listener.callsToRevoked);
-
- try {
- Thread.currentThread().interrupt();
- assertThrows(InterruptException.class, consumer::close);
- } finally {
- // Clear the interrupted flag so we don't create problems for
subsequent tests.
- Thread.interrupted();
- }
-
- assertEquals(1, listener.callsToAssigned);
- assertEquals(1, listener.callsToRevoked);
-
- Map<String, Object> consumerConfigMap = new
HashMap<>(consumerConfig);
- var config = new ConsumerConfig(consumerConfigMap);
-
- // Set the wait timeout to be only *half* the configured session
timeout. This way we can make sure that the
- // consumer explicitly left the group as opposed to being kicked
out by the broker.
- var leaveGroupTimeoutMs = config.getInt(SESSION_TIMEOUT_MS_CONFIG)
/ 2;
-
- TestUtils.waitForCondition(
- () -> checkGroupMemberEmpty(config),
- leaveGroupTimeoutMs,
- "Consumer did not leave the consumer group within " +
leaveGroupTimeoutMs + " ms of close"
- );
- }
- }
-
- private boolean checkGroupMemberEmpty(ConsumerConfig config) {
- try (var admin = cluster.admin()) {
- var groupId = config.getString(GROUP_ID_CONFIG);
- var result = admin.describeConsumerGroups(List.of(groupId));
- var groupDescription = result.describedGroups().get(groupId).get();
- return groupDescription.members().isEmpty();
- } catch (ExecutionException | InterruptedException e) {
- return false;
- }
- }
-
@ClusterTest
public void testClassicConsumerOffsetRelatedWhenTimeoutZero() throws
Exception {
testOffsetRelatedWhenTimeoutZero(Map.of(
diff --git
a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
index b28a2ce9d45..0d4fc11dc7c 100644
---
a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
@@ -2244,6 +2244,60 @@ public class KafkaConsumerTest {
consumerCloseTest(groupProtocol, Long.MAX_VALUE,
Collections.emptyList(), 0, true);
}
+ @Test
+ public void
testClassicConsumerCloseRunsRevocationCallbackAndAttemptsLeaveGroupWhenInterrupted()
{
+ ConsumerMetadata metadata = createMetadata(subscription);
+ MockClient client = new MockClient(time, metadata);
+
+ initMetadata(client, Map.of(topic, 1));
+ Node node = metadata.fetch().nodes().get(0);
+
+ final KafkaConsumer<String, String> consumer = newConsumer(
+ GroupProtocol.CLASSIC,
+ time,
+ client,
+ subscription,
+ metadata,
+ assignor,
+ false,
+ Optional.empty()
+ );
+
+ AtomicInteger revokedCount = new AtomicInteger(0);
+ AtomicReference<Set<TopicPartition>> revokedPartitions = new
AtomicReference<>();
+
+ ConsumerRebalanceListener listener = new ConsumerRebalanceListener() {
+ @Override
+ public void onPartitionsRevoked(Collection<TopicPartition>
partitions) {
+ assertTrue(Thread.currentThread().isInterrupted());
+ revokedCount.incrementAndGet();
+ revokedPartitions.set(Set.copyOf(partitions));
+ }
+
+ @Override
+ public void onPartitionsAssigned(Collection<TopicPartition>
partitions) {
+ // Preserve the existing helper behavior so assignment setup
remains equivalent.
+ for (TopicPartition partition : partitions)
+ consumer.seek(partition, 0);
+ }
+ };
+
+ consumer.subscribe(Set.of(topic), listener);
+ prepareRebalance(client, node, assignor, List.of(tp0), null);
+ consumer.updateAssignmentMetadataIfNeeded(time.timer(Long.MAX_VALUE));
+
+ try {
+ Thread.currentThread().interrupt();
+ assertThrows(InterruptException.class, () ->
consumer.close(CloseOptions.timeout(Duration.ofMillis(Long.MAX_VALUE))));
+ } finally {
+ Thread.interrupted();
+ }
+
+ assertEquals(1, revokedCount.get());
+ assertEquals(Set.of(tp0), revokedPartitions.get());
+ assertTrue(requestGenerated(client, ApiKeys.LEAVE_GROUP));
+ }
+
@ParameterizedTest
@EnumSource(GroupProtocol.class)
public void testCloseShouldBeIdempotent(GroupProtocol groupProtocol) {
diff --git
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
index 95566f2f5b3..6f9ef98a66f 100644
---
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
@@ -905,6 +905,66 @@ public class AsyncKafkaConsumerTest {
verify(applicationEventHandler).addAndGet(any(LeaveGroupOnCloseEvent.class));
}
+ @Test
+ public void
testCloseRunsRevocationCallbackAndSendsLeaveGroupEventOnInterrupt() {
+ final String topicName = "topic";
+ final Set<TopicPartition> partitions = singleton(new
TopicPartition(topicName, 0));
+
+ final AtomicReference<Set<TopicPartition>> revokedPartitions = new
AtomicReference<>();
+ final AtomicBoolean revocationCallbackCalled = new
AtomicBoolean(false);
+ final AtomicReference<LeaveGroupOnCloseEvent> leaveGroupEvent = new
AtomicReference<>();
+
+ final ConsumerRebalanceListener listener = new
ConsumerRebalanceListener() {
+ @Override
+ public void onPartitionsRevoked(final Collection<TopicPartition>
partitions) {
+ assertTrue(Thread.currentThread().isInterrupted());
+ revocationCallbackCalled.set(true);
+ revokedPartitions.set(Set.copyOf(partitions));
+ }
+
+ @Override
+ public void onPartitionsAssigned(final Collection<TopicPartition>
partitions) {
+ // no-op
+ }
+
+ @Override
+ public void onPartitionsLost(final Collection<TopicPartition>
partitions) {
+ fail("Expected assigned partitions to be revoked on close");
+ }
+ };
+
+ try (final MockedStatic<RequestManagers> requestManagers =
mockStatic(RequestManagers.class)) {
+ consumer =
newConsumer(requiredConsumerConfigAndGroupId("consumerGroup"));
+ completeTopicSubscriptionChangeEventSuccessfully();
+ consumer.subscribe(singletonList(topicName), listener);
+ consumer.subscriptions().assignFromSubscribed(partitions);
+ consumer.setGroupAssignmentSnapshot(partitions);
+
+ final MemberStateListener groupMetadataUpdateListener =
captureGroupMetadataUpdateListener(requestManagers);
+ groupMetadataUpdateListener.onMemberEpochUpdated(Optional.of(1),
"memberId");
+
+ doAnswer(invocation -> {
+ LeaveGroupOnCloseEvent event = invocation.getArgument(0);
+ leaveGroupEvent.set(event);
+ assertTrue(Thread.currentThread().isInterrupted());
+ throw new InterruptException("Thread was interrupted");
+
}).when(applicationEventHandler).addAndGet(ArgumentMatchers.isA(LeaveGroupOnCloseEvent.class));
+
+ try {
+ Thread.currentThread().interrupt();
+ assertThrows(InterruptException.class, () ->
consumer.close(CloseOptions.timeout(Duration.ZERO)));
+ } finally {
+ Thread.interrupted();
+ }
+ }
+
+ assertTrue(revocationCallbackCalled.get());
+ assertEquals(partitions, revokedPartitions.get());
+
verify(applicationEventHandler).addAndGet(ArgumentMatchers.isA(LeaveGroupOnCloseEvent.class));
+ assertNotNull(leaveGroupEvent.get());
+ assertEquals(CloseOptions.GroupMembershipOperation.DEFAULT,
leaveGroupEvent.get().membershipOperation());
+ }
+
@Test
public void testCommitSyncAllConsumed() {
SubscriptionState subscriptions = new SubscriptionState(new
LogContext(), AutoOffsetResetStrategy.NONE);
diff --git
a/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala
b/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala
deleted file mode 100644
index bbc4e6c350c..00000000000
--- a/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala
+++ /dev/null
@@ -1,75 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
contributor license agreements. See the NOTICE
- * file distributed with this work for additional information regarding
copyright ownership. The ASF licenses this file
- * to You under the Apache License, Version 2.0 (the "License"); you may not
use this file except in compliance with the
- * License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on
- * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
express or implied. See the License for the
- * specific language governing permissions and limitations under the License.
- */
-package kafka.api
-
-import java.util
-import kafka.utils.{TestInfoUtils, TestUtils}
-import org.apache.kafka.clients.consumer._
-import org.apache.kafka.common.errors.InterruptException
-import org.apache.kafka.common.test.api.Flaky
-import org.junit.jupiter.api.Assertions._
-import org.junit.jupiter.api.Timeout
-import org.junit.jupiter.params.ParameterizedTest
-import org.junit.jupiter.params.provider.MethodSource
-
-import java.util.concurrent.ExecutionException
-
-@Timeout(60)
-class PlaintextConsumerTest extends AbstractConsumerTest {
-
- @Flaky("KAFKA-18031")
- @ParameterizedTest(name =
TestInfoUtils.TestWithParameterizedGroupProtocolNames)
- @MethodSource(Array("getTestGroupProtocolParametersAll"))
- def testCloseLeavesGroupOnInterrupt(groupProtocol: String): Unit = {
- val adminClient = createAdminClient()
- val consumer = createConsumer()
- val listener = new TestConsumerReassignmentListener()
- consumer.subscribe(java.util.List.of(topic), listener)
- awaitRebalance(consumer, listener)
-
- assertEquals(1, listener.callsToAssigned)
- assertEquals(0, listener.callsToRevoked)
-
- try {
- Thread.currentThread().interrupt()
- assertThrows(classOf[InterruptException], () => consumer.close())
- } finally {
- // Clear the interrupted flag so we don't create problems for subsequent
tests.
- Thread.interrupted()
- }
-
- assertEquals(1, listener.callsToAssigned)
- assertEquals(1, listener.callsToRevoked)
-
- val config = new ConsumerConfig(consumerConfig)
-
- // Set the wait timeout to be only *half* the configured session timeout.
This way we can make sure that the
- // consumer explicitly left the group as opposed to being kicked out by
the broker.
- val leaveGroupTimeoutMs =
config.getInt(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG) / 2
-
- TestUtils.waitUntilTrue(
- () => {
- try {
- val groupId = config.getString(ConsumerConfig.GROUP_ID_CONFIG)
- val groupDescription =
adminClient.describeConsumerGroups(util.List.of(groupId)).describedGroups.get(groupId).get
- groupDescription.members.isEmpty
- } catch {
- case _: ExecutionException | _: InterruptedException =>
- false
- }
- },
- msg=s"Consumer did not leave the consumer group within
$leaveGroupTimeoutMs ms of close",
- waitTimeMs=leaveGroupTimeoutMs
- )
- }
-}
