dajac commented on code in PR #14640:
URL: https://github.com/apache/kafka/pull/14640#discussion_r1424116779
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java:
##########
@@ -860,6 +871,78 @@ public void testGroupMetadataUpdateSingleCall() {
}
}
+ /**
+ * Tests that the consumer correctly invokes the callbacks for {@link
ConsumerRebalanceListener} that was
+ * specified. We don't go through the full effort to emulate heartbeats
and correct group management here. We're
+ * simply exercising the background {@link EventProcessor} does the
correct thing when
+ * {@link AsyncKafkaConsumer#poll(Duration)} is called.
+ *
+ * Note that we test {@link ConsumerRebalanceListener} that throws errors
in its different callbacks. Failed
+ * callback execution does <em>not</em> immediately errors. Instead, those
errors are forwarded to the
+ * application event thread for the {@link MembershipManagerImpl} to
handle.
+ */
+ @ParameterizedTest
+ @MethodSource("listenerCallbacksInvokeSource")
+ public void
testListenerCallbacksInvoke(List<ConsumerRebalanceListenerMethodName>
methodNames,
+ Optional<RuntimeException>
revokedError,
+ Optional<RuntimeException>
assignedError,
+ Optional<RuntimeException>
lostError,
+ int expectedRevokedCount,
+ int expectedAssignedCount,
+ int expectedLostCount) {
+ CounterConsumerRebalanceListener consumerRebalanceListener = new
CounterConsumerRebalanceListener(
+ revokedError,
+ assignedError,
+ lostError
+ );
+ consumer.subscribe(Collections.singletonList("topic"),
consumerRebalanceListener);
+ SortedSet<TopicPartition> partitions = Collections.emptySortedSet();
+
+ for (ConsumerRebalanceListenerMethodName methodName : methodNames) {
+ CompletableBackgroundEvent<Void> e = new
ConsumerRebalanceListenerCallbackNeededEvent(methodName, partitions);
+ backgroundEventQueue.add(e);
+
+ // This will trigger the background event queue to process our
background event message.
+ consumer.poll(Duration.ZERO);
+ }
+
+ assertEquals(expectedRevokedCount,
consumerRebalanceListener.revokedCount());
+ assertEquals(expectedAssignedCount,
consumerRebalanceListener.assignedCount());
+ assertEquals(expectedLostCount, consumerRebalanceListener.lostCount());
Review Comment:
Should we also verify that an event is pushed back to the queue after the
execution of the callback?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java:
##########
@@ -790,6 +812,197 @@ public void
testOnSubscriptionUpdatedTransitionsToJoiningOnlyIfNotInGroup() {
verify(membershipManager, never()).transitionToJoining();
}
+ @Test
+ public void testListenerCallbacksBasic() {
Review Comment:
If we cannot reuse the existing integration tests, we will need to add new
ones. We could do it in a separate PR though.
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
