synchronized void requestFailed(Set<TopicPartition> partitions, long nextRetryTimeMs) {
        for (TopicPartition partition : partitions) {
            // by the time the request failed, the assignment may no longer
            // contain this partition any more, in which case we would just ignore.
            final TopicPartitionState state = assignedStateOrNull(partition);
            if (state != null)
                state.requestFailed(nextRetryTimeMs);
        }
    }
Enable fetching and updating positions for the given partitions that were added to the
assignment, but waiting for the onPartitionsAssigned callback to complete. This is
expected to be used by the async consumer.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java