Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java (Java)

Code before:
    }

    // Visible for testing
    public synchronized List<TopicPartition> fetchablePartitions(Predicate<TopicPartition> isAvailable) {
        // Since this is in the hot-path for fetching, we do this instead of using java.util.stream API
        List<TopicPartition> result = new ArrayList<>();
        assignment.forEach((topicPartition, topicPartitionState) -> {
            // Cheap check is first to avoid evaluating the predicate if possible
            if ((subscriptionType.equals(SubscriptionType.AUTO_TOPICS_SHARE) || isFetchableAndSubscribed(topicPartition, topicPartitionState))
                    && isAvailable.test(topicPartition)) {
                result.add(topicPartition);
            }
        });
        return result;
    }

    /**
     * Check if the partition is fetchable.
     * If the consumer has explicitly subscribed to a list of topic names,
     * this will also check that the partition is contained in the subscription.
     */
    private synchronized boolean isFetchableAndSubscribed(TopicPartition topicPartition, TopicPartitionState topicPartitionState) {
        if (subscriptionType.equals(SubscriptionType.AUTO_TOPICS) && !subscription.contains(topicPartition.topic())) {
            log.debug("Assigned partition {} is not in the subscription {} so will be considered not fetchable.", topicPartition, subscription);
            return false;
        }
        return topicPartitionState.isFetchable();
    }

    public synchronized boolean hasAutoAssignedPartitions() {
        return this.subscriptionType == SubscriptionType.AUTO_TOPICS || this.subscriptionType == SubscriptionType.AUTO_PATTERN
                || this.subscriptionType == SubscriptionType.AUTO_TOPICS_SHARE || this.subscriptionType == SubscriptionType.AUTO_PATTERN_RE2J;
    }

    /**
     * Check if the topic ID has been received in an assignment
     * from the coordinator after subscribing to a broker-side regex.
     */
    public synchronized boolean isAssignedFromRe2j(Uuid topicId) {
        if (topicId == null || !hasRe2JPatternSubscription()) {
            return false;
        }

        return this.assignedTopicIds.contains(topicId);
    }

    public synchronized void position(TopicPartition tp, FetchPosition position) {
        assignedState(tp).position(position);
    }

    /**

Review comment (bug):
Did we explore the option of removing or flagging the errant partition? This debug line has the possibility of being output on every fetch loop, right?

Code after:
    }

    // Visible for testing
    public synchronized List<TopicPartition> fetchablePartitions(Predicate<TopicPartition> isAvailable) {
        // Since this is in the hot-path for fetching, we do this instead of using java.util.stream API
        List<TopicPartition> result = new ArrayList<>();
        assignment.forEach((topicPartition, topicPartitionState) -> {
            // Cheap check is first to avoid evaluating the predicate if possible
            if ((subscriptionType.equals(SubscriptionType.AUTO_TOPICS_SHARE) || isFetchableAndSubscribed(topicPartition, topicPartitionState))
                    && isAvailable.test(topicPartition)) {
                result.add(topicPartition);
            }
        });
        return result;
    }

    /**
     * Check if the partition is fetchable.
     * If the consumer has explicitly subscribed to a list of topic names,
     * this will also check that the partition is contained in the subscription.
     */
    private synchronized boolean isFetchableAndSubscribed(TopicPartition topicPartition, TopicPartitionState topicPartitionState) {
        if (subscriptionType.equals(SubscriptionType.AUTO_TOPICS) && !subscription.contains(topicPartition.topic())) {
            log.trace("Assigned partition {} is not in the subscription {} so will be considered not fetchable.", topicPartition, subscription);
            return false;
        }
        return topicPartitionState.isFetchable();
    }

    public synchronized boolean hasAutoAssignedPartitions() {
        return this.subscriptionType == SubscriptionType.AUTO_TOPICS || this.subscriptionType == SubscriptionType.AUTO_PATTERN
                || this.subscriptionType == SubscriptionType.AUTO_TOPICS_SHARE || this.subscriptionType == SubscriptionType.AUTO_PATTERN_RE2J;
    }

    /**
     * Check if the topic ID has been received in an assignment
     * from the coordinator after subscribing to a broker-side regex.
     */
    public synchronized boolean isAssignedFromRe2j(Uuid topicId) {
        if (topicId == null || !hasRe2JPatternSubscription()) {
            return false;
        }

        return this.assignedTopicIds.contains(topicId);
    }

    public synchronized void position(TopicPartition tp, FetchPosition position) {
        assignedState(tp).position(position);
    }

    /**