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

Code before:
            if (wasReady) {
                // READY -> NOT_READY - start timer
                lastNotReadyLogTime = wallClockTime;
            } else {
                // NOT_READY - check if it should log
                final long timeSinceLastLog = wallClockTime - lastNotReadyLogTime;
                if (timeSinceLastLog >= NOT_READY_LOG_INTERVAL_MS) {
                    log.info("Task is not ready to process: has pending transaction commit");
                    lastNotReadyLogTime = wallClockTime;
                }
            }
            return false;
        }

        final boolean readyToProcess = partitionGroup.readyToProcess(wallClockTime);
        if (!readyToProcess) {
            if (wasReady) {
                // READY -> NOT_READY - start the timer
                lastNotReadyLogTime = wallClockTime;
                timeCurrentIdlingStarted = Optional.of(wallClockTime);
            } else {
                // NOT_READY - check if it should log
                final long timeSinceLastLog = wallClockTime - lastNotReadyLogTime;
                if (timeSinceLastLog >= NOT_READY_LOG_INTERVAL_MS) {
                    final String partitionStatus = getNotReadyPartitionStatus();
                    log.info("Task is not ready to process (started idling at time {}): {}", timeCurrentIdlingStarted.orElse(-1L), partitionStatus);
                    lastNotReadyLogTime = wallClockTime;
                }
            }
        } else {
            // Task is ready - clear the timer
            lastNotReadyLogTime = -1L;
            timeCurrentIdlingStarted = Optional.empty();
            log.trace("Task is ready to process");
        }
        return readyToProcess;
    }

    /**
     * Get the reason why the task is not ready to process.
     * This method is called when the task has been not ready for a while.
     */
    private String getNotReadyPartitionStatus() {
        final Set<TopicPartition> emptyPartitions = new HashSet<>();
        final Set<TopicPartition> bufferedPartitions = new HashSet<>();

        for (final TopicPartition partition : partitionGroup.partitions()) {
            if (partitionGroup.numBuffered(partition) == 0) {
                emptyPartitions.add(partition);
            } else {
                bufferedPartitions.add(partition);

Review comment (refactor):
It seems this message is rather generic -- guess it's the question you asked? -- I think it would be good to modify `partitionGroup.readyToProcess(...)` to allow us to get a better error message.

I see three ways:
1. change the return type to `String` (or `Optional<String>`) and use `null`/empty to indicate "read" -- not super clean but would work
2. add a second parameter `readyToProcess(long wallClockTime, Optional<String> returnedLogMessage)` (or similar), which we set before `return false` -- also not ideal to use an input parameter but again, would work.
3. add a new helper class that wraps `boolean readyToProcess` and `Optional<String> logMessage` and use a return type of `readyToProcess` (this is technically the cleanest way, but most involved change)

Code after:
            return false;
        }

        final boolean wasReady = lastNotReadyLogTime.isEmpty();
        if (hasPendingTxCommit) {
            // if the task has a pending TX commit, we should just retry the commit but not process any records
            // thus, the task is not processable, even if there is available data in the record queue
            if (wasReady) {
                // READY -> NOT_READY - start timer
                lastNotReadyLogTime = Optional.of(wallClockTime);
            } else {
                maybeLogNotReady(wallClockTime, "Task is not ready to process: has pending transaction commit");
            }
            return false;
        }

        final AbstractPartitionGroup.ReadyToProcessResult readyToProcess = partitionGroup.readyToProcess(wallClockTime);
        if (!readyToProcess.isReady()) {
            if (wasReady) {
                // READY -> NOT_READY - start the timer
                lastNotReadyLogTime = Optional.of(wallClockTime);
                timeCurrentIdlingStarted = Optional.of(wallClockTime);
            } else {
                maybeLogNotReady(wallClockTime, readyToProcess.getLogMessage().orElseThrow());
            }
        } else {
            // Task is ready - clear the timer
            lastNotReadyLogTime = Optional.empty();
            timeCurrentIdlingStarted = Optional.empty();
            log.trace("Task is ready to process");
        }
        return readyToProcess.isReady();
    }

    private void maybeLogNotReady(final long wallClockTime, final String logMessage) {
        // NOT_READY - check if it should log
        final long timeSinceLastLog = lastNotReadyLogTime.map(aLong -> wallClockTime - aLong).orElse(-1L);
        if (timeSinceLastLog >= NOT_READY_LOG_INTERVAL_MS) {
            log.info(logMessage);
            lastNotReadyLogTime = Optional.of(wallClockTime);
        }
    }

    /**
     * Process one record.
     *
     * @return true if this method processes a record, false if it does not process a record.
     * @throws TaskMigratedException if the task producer got fenced (EOS only)
     */
    @SuppressWarnings("unchecked")
    public boolean process(final long wallClockTime) {