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

Code before:
            log.info("Task is already in {} state, skip processing it.", state());

            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().orElse("Task is not ready to process"));
            }
        } 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(0L);
        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)
     */

Review comment (bug):
For this case, we should always have a log message, right? Wondering if we should use `orElseThrow` to surface a potential bug in our code directly?

Code after:
    public boolean isProcessable(final long wallClockTime) {
        if (state() == State.CLOSED) {
            // a task is only closing / closed when 1) task manager is closing, 2) a rebalance is undergoing;
            // in either case we can just log it and move on without notifying the thread since the consumer
            // would soon be updated to not return any records for this task anymore.
            log.info("Task is already in {} state, skip processing it.", state());

            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);
        }
    }

    /**