lianetm commented on code in PR #16686:
URL: https://github.com/apache/kafka/pull/16686#discussion_r1702092692
##########
core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala:
##########
@@ -911,4 +911,42 @@ class PlaintextConsumerTest extends BaseConsumerTest {
assertThrows(classOf[WakeupException], () =>
consumer.position(topicPartition, Duration.ofSeconds(100)))
}
+
+ @ParameterizedTest(name =
TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+
@MethodSource(Array("getTestQuorumAndGroupProtocolParametersConsumerGroupProtocolOnly"))
+ def testCloseLeavesGroupOnInterrupt(quorum: String, groupProtocol: String):
Unit = {
+ val adminClient = createAdminClient()
+ val consumer = createConsumer()
+ val groupId = consumerConfig.getProperty("group.id")
+
+ def hasMembers: Boolean = {
+ try {
+ val groupDescription = adminClient.describeConsumerGroups
(Collections.singletonList (groupId) ).describedGroups.get (groupId).get
+ groupDescription.members.size() > 0
+ } catch {
+ case _: ExecutionException | _: InterruptedException =>
+ false
+ }
+ }
+
+ val listener = new TestConsumerReassignmentListener()
+ consumer.subscribe(List(topic).asJava, listener)
+ awaitRebalance(consumer, listener)
+
+ assertEquals(1, listener.callsToAssigned)
+ assertEquals(0, listener.callsToRevoked)
+ TestUtils.waitUntilTrue(() => hasMembers, s"Consumer did not join the
consumer group within ${JTestUtils.DEFAULT_MAX_WAIT_MS} of subscribe")
+
+ 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)
+ TestUtils.waitUntilTrue(() => !hasMembers, s"Consumer did not leave the
consumer group within ${JTestUtils.DEFAULT_MAX_WAIT_MS} of interrupt/close")
Review Comment:
Won't this check always be true, regardless of the changes on this PR, the
moment the session timeout expires? We should set a long session timeout, to be
sure that the group ends up empty because the consumer left it intentionally
even if interrupted (and not because its session expired and the broker kicked
it out).
We can override the broker configs with the existing
[brokerPropertiesOverrides](https://github.com/apache/kafka/blob/704476885ffb40cd3bf9b8f5c368c01eaee0a737/core/src/test/scala/integration/kafka/api/AbstractConsumerTest.scala#L70),
and we should also be careful to do it only for this test (to not affect
behaviour in others, like it's done
[here](https://github.com/apache/kafka/blob/704476885ffb40cd3bf9b8f5c368c01eaee0a737/core/src/test/scala/integration/kafka/api/BaseAdminIntegrationTest.scala#L202)
I believe).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
