public ProcessId poll(final TaskId task, final Function<ProcessId, Boolean> extraConstraint) {
        final Set<ProcessId> invalidPolledClients = new HashSet<>();
        while (!clientsByTaskLoad.isEmpty()) {
            final ProcessId candidateClient = pollNextClient();
            if (constraint.apply(candidateClient, task) && extraConstraint.apply(candidateClient)) {
                // then we found the lightest, valid client
                offerAll(invalidPolledClients);
                return candidateClient;
            } else {
                // remember this client and try again later
                invalidPolledClients.add(candidateClient);
            }
        }
        // we tried all the clients, and none met the constraint (or there are no clients)
        offerAll(invalidPolledClients);
        return null;
    }
@return the next least loaded client that satisfies the given criteria, or null if none do
streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/ConstrainedPrioritySet.java