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

        if (offsetResetTimestamps.isEmpty())
            return;

        resetPositionsAsync(offsetResetTimestamps);
    }
@Test
    public void testListOffsetNoUpdateMissingEpoch() {
        buildFetcher();

        // Set up metadata with no leader epoch
        subscriptions.assignFromUser(singleton(tp0));
        MetadataResponse metadataWithNoLeaderEpochs = RequestTestUtils.metadataUpdateWithIds(
                "kafka-cluster", 1, Collections.emptyMap(), singletonMap(topicName, 4), tp -> null, topicIds);
        client.updateMetadata(metadataWithNoLeaderEpochs);

        // Return a ListOffsets response with leaderEpoch=1, we should ignore it
        subscriptions.requestOffsetReset(tp0, OffsetResetStrategy.LATEST);
        client.prepareResponse(listOffsetRequestMatcher(ListOffsetsRequest.LATEST_TIMESTAMP),
                listOffsetResponse(tp0, Errors.NONE, 1L, 5L, 1));
        offsetFetcher.resetPositionsIfNeeded();
        consumerClient.pollNoWakeup();

        // Reset should be satisfied and no metadata update requested
        assertFalse(subscriptions.isOffsetResetNeeded(tp0));
        assertFalse(metadata.updateRequested());
        assertFalse(metadata.lastSeenLeaderEpoch(tp0).isPresent());
    }