KAFKA-19110: Add missing unit test for Streams-consumer integration #19457
Conversation
Signed-off-by: PoAn Yang <payang@apache.org>
Signed-off-by: PoAn Yang <payang@apache.org>
FrankYang0529
changed the title
| HANDLER | ||
| ) | ||
| ); | ||
| assertEquals("Named topologies and the CONSUMER protocol cannot be used at the same time.", exception.getMessage()); |
There was a problem hiding this comment.
This exception message is actually wrong. "Named topologies and the STREAMS protocol" should be what it says. Can you update the test and the production code?
There was a problem hiding this comment.
Addressed it. Thanks.
Signed-off-by: PoAn Yang <payang@apache.org>
| @@ -220,7 +228,7 @@ private AsyncKafkaConsumer<String, String> newConsumer(Properties props) { | |||
| (a, b, c, d, e, f, g) -> fetchCollector, | |||
There was a problem hiding this comment.
I know this is pre-existing, but I'm wondering if it's worth doing a follow-up PR to use more meaningful names than a, b etc.
There was a problem hiding this comment.
I will handle this in a follow-up PR. Thanks.
There was a problem hiding this comment.
Follow-up PR: #19550
| taskManager | ||
| ); | ||
|
|
||
| private void setup(final StreamsRebalanceData streamsRebalanceData) { |
There was a problem hiding this comment.
nit: It took a second to realize what was going on with creating a DefaultStreamsRebalanceListener in two places. Maybe change the name from setup() with something like createRebalanceListenerWithRebalanceData
There was a problem hiding this comment.
Updated it. Thanks.
Signed-off-by: PoAn Yang <payang@apache.org>
Replace names like a, b, c, ... with meaningful names in AsyncKafkaConsumerTest. Follow-up: #19457 (comment) Signed-off-by: PoAn Yang <payang@apache.org> Reviewers: Bill Bejeck <bbejeck@apache.org>, Ken Huang <s7133700@gmail.com>
AsyncKafkaConsumerconstructor and verify that theRequestManagers.supplier()contains Streams-specific data structures.RequestManagersconstructs the Streams request managerscorrectly
StreamsGroupHeartbeatManager#resetPollTimer()StreamsOnTasksRevokedCallbackCompletedEvent,StreamsOnTasksAssignedCallbackCompletedEvent, andStreamsOnAllTasksLostCallbackCompletedEventinApplicationEventProcessorDefaultStreamsRebalanceListenerStreamThread.handleStreamsRebalanceData.StreamsRebalanceData.Reviewers: Lucas Brutschy lbrutschy@confluent.io