public synchronized Optional<Integer> clearPreferredReadReplica(TopicPartition tp) {
        final TopicPartitionState topicPartitionState = assignedStateOrNull(tp);
        if (topicPartitionState == null) {
            return Optional.empty();
        } else {
            return topicPartitionState.clearPreferredReadReplica();
        }
    }
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