public synchronized Map<TopicPartition, OffsetAndMetadata> allConsumed() {
        Map<TopicPartition, OffsetAndMetadata> allConsumed = new HashMap<>();
        assignment.forEach((topicPartition, partitionState) -> {
            if (partitionState.hasValidPosition())
                allConsumed.put(topicPartition, new OffsetAndMetadata(partitionState.position.offset,
                        partitionState.position.offsetEpoch, ""));
        });
        return allConsumed;
    }
Unset the preferred read replica. This causes the fetcher to go back to the leader for fetches.

@param tp The topic partition
@return the removed preferred read replica if set, Empty otherwise.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java