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

Code before:
                        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();

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

            if (!queue.isEmpty()) {
                // this partition is ready for processing
                appendLog(logMessageBuilder, String.format("Partition %s has buffered data, ready for processing", partition));
                idlePartitionDeadlines.remove(partition);
                queued.add(partition);
            } else {
                final Long fetchedLag = fetchedLags.getOrDefault(partition, -1L);
                appendLog(logMessageBuilder, String.format("Partition %s has fetched lag for %d", partition, fetchedLag));
                
                if (fetchedLag == -1L) {
                    // must wait to fetch metadata for the partition
                    idlePartitionDeadlines.remove(partition);
                    appendLog(logMessageBuilder, String.format(String.format("\tWaiting to fetch data for %s", partition)));

                    return new ReadyToProcessResult(false, Optional.of(logMessageBuilder.toString()));
                } else if (fetchedLag > 0L) {
                    // must wait to poll the data we know to be on the broker
                    idlePartitionDeadlines.remove(partition);
                    appendLog(logMessageBuilder,
                        String.format("Partition %s has current lag %d, but no data is buffered locally. Waiting to buffer some records.",
                        partition, fetchedLag));

                    return new ReadyToProcessResult(false, Optional.of(logMessageBuilder.toString()));
                } else {
                    // p is known to have zero lag. wait for maxTaskIdleMs to see if more data shows up.
                    // One alternative would be to set the deadline to nullableMetadata.receivedTimestamp + maxTaskIdleMs
                    // instead. That way, we would start the idling timer as of the freshness of our knowledge about the zero
                    // lag instead of when we happen to run this method, but realistically it's probably a small difference

Review comment (suggestion):
Similar to my commend above: for the positive case (we do return `true` later, we don't really care. We would still log at TRACE directly, but the actually interesting case is the "false" case).

Applies further below, too.

Code after:
                    }
                }
                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();

            if (!queue.isEmpty()) {
                // this partition is ready for processing
                logger.trace("Partition {} has buffered data, ready for processing", partition);
                idlePartitionDeadlines.remove(partition);
                queued.add(partition);
            } else {
                final Long fetchedLag = fetchedLags.getOrDefault(partition, -1L);
                appendLog(logMessageBuilder, String.format("Partition %s has fetched lag of %d", partition, fetchedLag));
                
                if (fetchedLag == -1L) {
                    // must wait to fetch metadata for the partition
                    idlePartitionDeadlines.remove(partition);
                    appendLog(logMessageBuilder, String.format("\tWaiting to fetch data for %s", partition));

                    return new ReadyToProcessResult(false, Optional.of(logMessageBuilder.toString()));
                } else if (fetchedLag > 0L) {
                    // must wait to poll the data we know to be on the broker
                    idlePartitionDeadlines.remove(partition);
                    appendLog(logMessageBuilder,
                        String.format("Partition %s has current lag %d, but no data is buffered locally. Waiting to buffer some records.",
                        partition, fetchedLag));

                    return new ReadyToProcessResult(false, Optional.of(logMessageBuilder.toString()));
                } else {
                    // p is known to have zero lag. wait for maxTaskIdleMs to see if more data shows up.
                    // One alternative would be to set the deadline to nullableMetadata.receivedTimestamp + maxTaskIdleMs
                    // instead. That way, we would start the idling timer as of the freshness of our knowledge about the zero
                    // lag instead of when we happen to run this method, but realistically it's probably a small difference
                    // and using wall clock time seems more intuitive for users,
                    // since the log message will be as of wallClockTime.
                    idlePartitionDeadlines.putIfAbsent(partition, wallClockTime + maxTaskIdleMs);