private void testUpdateFetchPositionsWithFetchCommittedOffsetsTimeout(boolean committedOffsetsEnabled) {
        completeFetchedCommittedOffsetApplicationEventExceptionally(new TimeoutException());
        doReturn(Fetch.empty()).when(fetchCollector).collectFetch(any(FetchBuffer.class));

        consumer.assign(singleton(new TopicPartition("t1", 1)));

        consumer.poll(Duration.ZERO);

        verify(applicationEventHandler, atLeast(1))
            .addAndGet(ArgumentMatchers.isA(ValidatePositionsEvent.class));

        if (committedOffsetsEnabled) {
            // Verify there was an FetchCommittedOffsets event and no ResetPositions event
            verify(applicationEventHandler, atLeast(1))
                .add(ArgumentMatchers.isA(FetchCommittedOffsetsEvent.class));
            verify(applicationEventHandler, never())
                .addAndGet(ArgumentMatchers.isA(ResetPositionsEvent.class));
        } else {
            // Verify there was not any FetchCommittedOffsets event but there should be a ResetPositions
            verify(applicationEventHandler, never())
                .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