dajac commented on code in PR #14640:
URL: https://github.com/apache/kafka/pull/14640#discussion_r1424300228
##########
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:
I see. Ideally, we should mock the background part of it here. If this is
too hard. We can address it separately.
--
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
