Repository: apache/kafka
File: streams/src/main/java/org/apache/kafka/streams/processor/internals/PartitionGroup.java (Java)

Code before:
    private final Map<TopicPartition, Long> idlePartitionDeadlines = new HashMap<>();
    private final Map<TopicPartition, Long> fetchedLags = new HashMap<>();

    PartitionGroup(final LogContext logContext,
                   final Map<TopicPartition, RecordQueue> partitionQueues,
                   final Function<TopicPartition, OptionalLong> lagProvider,
                   final Sensor recordLatenessSensor,
                   final Sensor enforcedProcessingSensor,
                   final long maxTaskIdleMs) {
        this.logger = logContext.logger(PartitionGroup.class);
        nonEmptyQueuesByTime = new PriorityQueue<>(partitionQueues.size(), Comparator.comparingLong(RecordQueue::headRecordTimestamp));
        this.partitionQueues = partitionQueues;
        this.lagProvider = lagProvider;
        this.enforcedProcessingSensor = enforcedProcessingSensor;
        this.maxTaskIdleMs = maxTaskIdleMs;
        this.recordLatenessSensor = recordLatenessSensor;
        totalBuffered = 0;
        allBuffered = false;
        streamTime = RecordQueue.UNKNOWN;
    }

    @Override
    ReadyToProcessResult readyToProcess(final long wallClockTime) {
        Optional<String> logMessage = Optional.empty();
        if (maxTaskIdleMs == StreamsConfig.MAX_TASK_IDLE_MS_DISABLED) {
            if (!allBuffered && totalBuffered > 0) {
                final Set<TopicPartition> bufferedPartitions = new HashSet<>();
                final Set<TopicPartition> emptyPartitions = new HashSet<>();
                for (final Map.Entry<TopicPartition, RecordQueue> entry : partitionQueues.entrySet()) {
                    if (entry.getValue().isEmpty()) {
                        emptyPartitions.add(entry.getKey());
                    } else {
                        bufferedPartitions.add(entry.getKey());
                    }
                }
                logMessage = Optional.of(String.format(
                    "Ready for processing because max.task.idle.ms is disabled.%n" +
                        "\tThere may be out-of-order processing for this task as a result.%n" +
                        "\tBuffered partitions: %s%n" +
                        "\tNon-buffered partitions: %s",
                    bufferedPartitions,
                    emptyPartitions
                ));
            }
            return new ReadyToProcessResult(true, logMessage);
        }

        final Set<TopicPartition> queued = new HashSet<>();
        Map<TopicPartition, Long> enforced = null;
        final StringBuilder logMessageBuilder = new StringBuilder();


Review comment (suggestion):
I believe we would want to keep this unchanged, and just return `new ReadyToProcessResult(true, Optional.empty());`

The existing trace log is for the case when we are processable, and the idea of `ReadyToProcessResult` is return a log message only for the "not processable" case.

Code after:
    private final Map<TopicPartition, Long> idlePartitionDeadlines = new HashMap<>();
    private final Map<TopicPartition, Long> fetchedLags = new HashMap<>();

    PartitionGroup(final LogContext logContext,
                   final Map<TopicPartition, RecordQueue> partitionQueues,
                   final Function<TopicPartition, OptionalLong> lagProvider,
                   final Sensor recordLatenessSensor,
                   final Sensor enforcedProcessingSensor,
                   final long maxTaskIdleMs) {
        this.logger = logContext.logger(PartitionGroup.class);
        nonEmptyQueuesByTime = new PriorityQueue<>(partitionQueues.size(), Comparator.comparingLong(RecordQueue::headRecordTimestamp));
        this.partitionQueues = partitionQueues;
        this.lagProvider = lagProvider;
        this.enforcedProcessingSensor = enforcedProcessingSensor;
        this.maxTaskIdleMs = maxTaskIdleMs;
        this.recordLatenessSensor = recordLatenessSensor;
        totalBuffered = 0;
        allBuffered = false;
        streamTime = RecordQueue.UNKNOWN;
    }

    @Override
    ReadyToProcessResult readyToProcess(final long wallClockTime) {
        if (maxTaskIdleMs == StreamsConfig.MAX_TASK_IDLE_MS_DISABLED) {
            if (logger.isTraceEnabled() && !allBuffered && totalBuffered > 0) {
                final Set<TopicPartition> bufferedPartitions = new HashSet<>();
                final Set<TopicPartition> emptyPartitions = new HashSet<>();
                for (final Map.Entry<TopicPartition, RecordQueue> entry : partitionQueues.entrySet()) {
                    if (entry.getValue().isEmpty()) {
                        emptyPartitions.add(entry.getKey());
                    } else {
                        bufferedPartitions.add(entry.getKey());
                    }
                }
                logger.trace("Ready for processing because max.task.idle.ms is disabled." +
                                "\n\tThere may be out-of-order processing for this task as a result." +
                                "\n\tBuffered partitions: {}" +
                                "\n\tNon-buffered partitions: {}",
                        bufferedPartitions,
                        emptyPartitions);
            }
            return new ReadyToProcessResult(true, Optional.empty());
        }

        final Set<TopicPartition> queued = new HashSet<>();
        Map<TopicPartition, Long> enforced = null;
        final StringBuilder logMessageBuilder = new StringBuilder();

        for (final Map.Entry<TopicPartition, RecordQueue> entry : partitionQueues.entrySet()) {
            final TopicPartition partition = entry.getKey();
            final RecordQueue queue = entry.getValue();