kirktrue commented on code in PR #14640:
URL: https://github.com/apache/kafka/pull/14640#discussion_r1424285645
##########
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 attempted this, but due to the way the way the code and tests are
structured, it's very awkward to do so.
Basically, when we create an `AsyncKafkaConsumer`, we start the background
network I/O thread. That background thread pulls the events off of the queue,
so there's a timing issue which could make the test flaky. I'll make another
attempt at this today.
--
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
