-
Notifications
You must be signed in to change notification settings - Fork 14k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
KAFKA-16675: Refactored and new rebalance callbacks integration tests #15965
Conversation
lianetm
commented
May 15, 2024
- Move existing rebalance callback + consumer.position test to the PlaintextConsumerCallbackTest file (refactored to reuse the new helper funcs available)
- Add new integration tests for callbacks interaction with seek and pause
- Minor cleanup in the callbacks test file
Hey @lucasbru, extending on top of your initial callbacks test file to get more coverage (very nice helper funcs btw). Could you take a look when you have a chance? Thanks! |
} | ||
|
||
override def onPartitionsRevoked(partitions: util.Collection[TopicPartition]): Unit = { | ||
// noop | ||
} | ||
}) | ||
TestUtils.pollUntilTrue(consumer, () => partitionsAssigned.get(), "Timed out before expected rebalance completed") | ||
consumer.close() |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Removed as it's not needed here, the test already has the logic for closing all consumers on (IntegrationTestHarness.tearDown). (Left it on the triggerOnPartitionsRevoked
as it was, where it's needed to assert the revocation)
val startingTimestamp = 0 | ||
sendRecords(producer, totalRecords.toInt, tp, startingTimestamp) | ||
|
||
consumer.subscribe(asList(topic), new ConsumerRebalanceListener { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Could you not use the helper method below here?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I couldn't initially mainly because of the close and poll in the helper, that played against what I needed to test in these 2, but then I myself removed the close and forgot to try again :). So I did integrate the helper here now, with a minor twist to pass the consumer as param. Also it allowed me to simplify the seek/pause test in one, given that we do need to pause to properly check the seek behaviour, so removed the extra test for pause. Good catch, thanks!
Thanks for the PR @lianetm . I just left one question |
Thanks for the helpful comment @lucasbru , addressed. |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
LGTM, thanks!
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
LGTM, thanks!
…apache#15965) Move existing rebalance callback + consumer.position test to the PlaintextConsumerCallbackTest file (refactored to reuse the new helper funcs available) Add new integration tests for callbacks interaction with seek and pause Minor cleanup in the callbacks test file Reviewers: Lucas Brutschy <lbrutschy@confluent.io>
…apache#15965) Move existing rebalance callback + consumer.position test to the PlaintextConsumerCallbackTest file (refactored to reuse the new helper funcs available) Add new integration tests for callbacks interaction with seek and pause Minor cleanup in the callbacks test file Reviewers: Lucas Brutschy <lbrutschy@confluent.io>
…apache#15965) Move existing rebalance callback + consumer.position test to the PlaintextConsumerCallbackTest file (refactored to reuse the new helper funcs available) Add new integration tests for callbacks interaction with seek and pause Minor cleanup in the callbacks test file Reviewers: Lucas Brutschy <lbrutschy@confluent.io>