lianetm commented on code in PR #15613:
URL: https://github.com/apache/kafka/pull/15613#discussion_r1554162586
##########
core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala:
##########
@@ -654,6 +654,64 @@ class PlaintextConsumerTest extends BaseConsumerTest {
assertEquals(7, consumer.committed(Set(tp2).asJava).get(tp2).offset)
}
+ @ParameterizedTest(name =
TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+ @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
+ def testCommitAsyncCompletedConsumerCloses(quorum: String, groupProtocol:
String): Unit = {
+ // This is testing the contract that asynchronous offset commit are
completed before the consumer
+ // is closed.
Review Comment:
I would extend with something like : ...before the consumer is closed,
**even when no commit sync is performed as part of the close (due to
auto-commit disabled, or simply because there no consumed offsets).**
That's the key as I see it, fixed in this PR, and being tested here. If the
call to consumer.close performs an actual commit sync (needs auto-commit
enabled and non-empty consumed offsets), then the async callbacks were always
called I expect. The contract was not being respected in case the commit sync
did not happen for some of the reasons mentioned above. Agree?
--
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]
