public void resetPositionsIfNeeded() {
        Map<TopicPartition, Long> offsetResetTimestamps = offsetFetcherUtils.getOffsetResetTimestamp();

        if (offsetResetTimestamps.isEmpty())
            return;

        resetPositionsAsync(offsetResetTimestamps);
    }
@Test
    public void testEarlierOffsetResetArrivesLate() {
        buildFetcher();
        assignFromUser(singleton(tp0));

        subscriptions.requestOffsetReset(tp0, OffsetResetStrategy.EARLIEST);
        offsetFetcher.resetPositionsIfNeeded();

        client.prepareResponse(req -> {
            if (listOffsetMatchesExpectedReset(tp0, OffsetResetStrategy.EARLIEST, req)) {
                // Before the response is handled, we get a request to reset to the latest offset
                subscriptions.requestOffsetReset(tp0, OffsetResetStrategy.LATEST);
                return true;
            } else {
                return false;
            }
        }, listOffsetResponse(Errors.NONE, 1L, 0L));
        consumerClient.pollNoWakeup();

        // The list offset result should be ignored
        assertTrue(subscriptions.isOffsetResetNeeded(tp0));
        assertEquals(OffsetResetStrategy.LATEST, subscriptions.resetStrategy(tp0));

        offsetFetcher.resetPositionsIfNeeded();
        client.prepareResponse(
            req -> listOffsetMatchesExpectedReset(tp0, OffsetResetStrategy.LATEST, req),
            listOffsetResponse(Errors.NONE, 1L, 10L)
        );
        consumerClient.pollNoWakeup();

        assertFalse(subscriptions.isOffsetResetNeeded(tp0));
        assertEquals(10, subscriptions.position(tp0).offset);
    }