private Set<TopicPartition> collectPartitions(Predicate<TopicPartitionState> filter) {
        Set<TopicPartition> result = new HashSet<>();
        assignment.forEach((topicPartition, topicPartitionState) -> {
            if (filter.test(topicPartitionState)) {
                result.add(topicPartition);
            }
        });
        return result;
    }
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