private void testRefreshCommittedOffsetsSuccess(Set<TopicPartition> partitions,
                                                    Map<TopicPartition, OffsetAndMetadata> committedOffsets) {
        completeFetchedCommittedOffsetApplicationEventSuccessfully(committedOffsets);
        doReturn(Fetch.empty()).when(fetchCollector).collectFetch(any(FetchBuffer.class));
        doReturn(LeaderAndEpoch.noLeaderOrEpoch()).when(metadata).currentLeader(any());

        consumer.assign(partitions);

        consumer.poll(Duration.ZERO);

        verify(applicationEventHandler, atLeast(1))
            .addAndGet(ArgumentMatchers.isA(ValidatePositionsEvent.class));
        verify(applicationEventHandler, atLeast(1))
            .add(ArgumentMatchers.isA(FetchCommittedOffsetsEvent.class));
        verify(applicationEventHandler, atLeast(1))
            .addAndGet(ArgumentMatchers.isA(ResetPositionsEvent.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