@Override
    boolean 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 true;
        }

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

        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
                idlePartitionDeadlines.remove(partition);
                queued.add(partition);
            } else {
                final Long fetchedLag = fetchedLags.getOrDefault(partition, -1L);

                logger.trace("Fetched lag for {} is {}", partition, fetchedLag);

                if (fetchedLag == -1L) {
                    // must wait to fetch metadata for the partition
                    idlePartitionDeadlines.remove(partition);
                    logger.trace("Waiting to fetch data for {}", partition);
                    return false;
                } else if (fetchedLag > 0L) {
                    // must wait to poll the data we know to be on the broker
                    idlePartitionDeadlines.remove(partition);
                    logger.trace(
                            "Lag for {} is currently {}, but no data is buffered locally. Waiting to buffer some records.",
                            partition,
                            fetchedLag
                    );
                    return false;
                } 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);
                    final long deadline = idlePartitionDeadlines.get(partition);
                    if (wallClockTime < deadline) {
                        logger.trace(
                                "Lag for {} is currently 0 and current time is {}. Waiting for new data to be produced for configured idle time {} (deadline is {}).",
                                partition,
                                wallClockTime,
                                maxTaskIdleMs,
                                deadline
                        );
                        return false;
                    } else {
                        // this partition is ready for processing due to the task idling deadline passing
                        if (enforced == null) {
                            enforced = new HashMap<>();
                        }
                        enforced.put(partition, deadline);
                    }
                }
            }
        }
        if (enforced == null) {
            logger.trace("All partitions were buffered locally, so this task is ready for processing.");
            return true;
        } else if (queued.isEmpty()) {
            logger.trace("No partitions were buffered locally, so this task is not ready for processing.");
            return false;
        } else {
            enforcedProcessingSensor.record(1.0d, wallClockTime);
            logger.trace("Continuing to process although some partitions are empty on the broker." +
                            "\n\tThere may be out-of-order processing for this task as a result." +
                            "\n\tPartitions with local data: {}." +
                            "\n\tPartitions we gave up waiting for, with their corresponding deadlines: {}." +
                            "\n\tConfigured max.task.idle.ms: {}." +
                            "\n\tCurrent wall-clock time: {}.",
                    queued,
                    enforced,
                    maxTaskIdleMs,
                    wallClockTime);
            return true;
        }
    }
PartitionGroup is used to buffer all co-partitioned records for processing.

In other words, it represents the "same" partition over multiple co-partitioned topics, and it is used
to buffer records from that partition in each of the contained topic-partitions.
Each StreamTask has exactly one PartitionGroup.

PartitionGroup implements the algorithm that determines in what order buffered records are selected for processing.

Specifically, when polled, it returns the record from the topic-partition with the lowest stream-time.
Stream-time for a topic-partition is defined as the highest timestamp
yet observed at the head of that topic-partition.

PartitionGroup also maintains a stream-time for the group as a whole.
This is defined as the highest timestamp of any record yet polled from the PartitionGroup.
Note however that any computation that depends on stream-time should track it on a per-operator basis to obtain an
accurate view of the local time as seen by that processor.

The PartitionGroups's stream-time is initially UNKNOWN (-1), and it set to a known value upon first poll.
As a consequence of the definition, the PartitionGroup's stream-time is non-decreasing
(i.e., it increases or stays the same over time).
streams/src/main/java/org/apache/kafka/streams/processor/internals/PartitionGroup.java