public synchronized Set<TopicPartition> partitionsNeedingReset(long nowMs) {
        return collectPartitions(state -> state.awaitingReset() && !state.awaitingRetryBackoff(nowMs));
    }
Note: this will not attempt to reset partitions that are in the process of being assigned
and are pending the completion of any {@link ConsumerRebalanceListener#onPartitionsAssigned(Collection)}
callbacks.

<p/>

This method only appears to be invoked the by the {@link KafkaConsumer} during its
{@link KafkaConsumer#poll(Duration)} logic. <em>Direct</em> calls to methods like
{@link #requestOffsetReset(TopicPartition)}, {@link #requestOffsetResetIfPartitionAssigned(TopicPartition)},
etc. do <em>not</em> skip partitions pending assignment.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java