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

        if (offsetResetTimestamps.isEmpty())
            return;

        resetPositionsAsync(offsetResetTimestamps);
    }
@Test
    public void testFetchingPendingPartitionsBeforeAndAfterSubscriptionReset() {
        buildFetcher();
        assignFromUser(singleton(tp0));
        subscriptions.seek(tp0, 100);
        assertEquals(100, subscriptions.position(tp0).offset);

        assertTrue(subscriptions.isFetchable(tp0));

        subscriptions.markPendingRevocation(singleton(tp0));
        offsetFetcher.resetPositionsIfNeeded();

        // once a partition is marked pending, it should not be fetchable
        assertFalse(subscriptions.isOffsetResetNeeded(tp0));
        assertFalse(subscriptions.isFetchable(tp0));
        assertTrue(subscriptions.hasValidPosition(tp0));
        assertEquals(100, subscriptions.position(tp0).offset);

        subscriptions.seek(tp0, 100);
        assertEquals(100, subscriptions.position(tp0).offset);

        // reassignment should enable fetching of the same partition
        subscriptions.unsubscribe();
        assignFromUser(singleton(tp0));
        subscriptions.seek(tp0, 100);
        assertEquals(100, subscriptions.position(tp0).offset);
        assertTrue(subscriptions.isFetchable(tp0));
    }