@ParameterizedTest
    @MethodSource("listenerCallbacksInvokeSource")
    public void testListenerCallbacksInvoke(List<ConsumerRebalanceListenerMethodName> methodNames,
                                            Optional<RuntimeException> revokedError,
                                            Optional<RuntimeException> assignedError,
                                            Optional<RuntimeException> lostError,
                                            int expectedRevokedCount,
                                            int expectedAssignedCount,
                                            int expectedLostCount,
                                            Optional<RuntimeException> expectedException
                                            ) {
        consumer = newConsumer();
        CounterConsumerRebalanceListener consumerRebalanceListener = new CounterConsumerRebalanceListener(
                revokedError,
                assignedError,
                lostError
        );
        doReturn(Fetch.empty()).when(fetchCollector).collectFetch(any(FetchBuffer.class));
        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.
        // If any error is happening inside the rebalance callbacks, we expect the first exception to be thrown from poll.
        if (expectedException.isPresent()) {
            Exception exception = assertThrows(expectedException.get().getClass(), () -> consumer.poll(Duration.ZERO));
            assertEquals(expectedException.get().getMessage(), exception.getMessage());
            assertEquals(expectedException.get().getCause(), exception.getCause());
        } else {
            assertDoesNotThrow(() -> consumer.poll(Duration.ZERO));
        }

        assertEquals(expectedRevokedCount, consumerRebalanceListener.revokedCount());
        assertEquals(expectedAssignedCount, consumerRebalanceListener.assignedCount());
        assertEquals(expectedLostCount, consumerRebalanceListener.lostCount());
    }
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.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java