@Test
    public void testEnsurePollEventSentOnConsumerPoll() {
        SubscriptionState subscriptions = new SubscriptionState(new LogContext(), OffsetResetStrategy.NONE);
        consumer = newConsumer(
                mock(FetchBuffer.class),
                new ConsumerInterceptors<>(Collections.emptyList()),
                mock(ConsumerRebalanceListenerInvoker.class),
                subscriptions,
                "group-id",
                "client-id");
        final TopicPartition tp = new TopicPartition("topic", 0);
        final List<ConsumerRecord<String, String>> records = singletonList(
                new ConsumerRecord<>("topic", 0, 2, "key1", "value1"));
        doAnswer(invocation -> Fetch.forPartition(tp, records, true))
                .when(fetchCollector)
                .collectFetch(Mockito.any(FetchBuffer.class));

        consumer.subscribe(singletonList("topic1"));
        consumer.poll(Duration.ofMillis(100));
        verify(applicationEventHandler).add(any(PollEvent.class));
    }
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