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

Code before:
     * source topic partitions, or if it is enforced to be processable.
     */
    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 == -1L;
        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 = 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");
        }

Review comment (refactor):
It seems we can extract this into a helper method (`maybeLogNotReady(...)`, just passing in the log message? This code is always the same:
```
final long timeSinceLastLog = wallClockTime - lastNotReadyLogTime;
if (timeSinceLastLog >= NOT_READY_LOG_INTERVAL_MS) {
    log.info(logMessage);
    lastNotReadyLogTime = wallClockTime;
}
```

Code after:
        partitionsToResume.clear();

        transitionTo(State.CLOSED);
    }

    /**
     * An active task is processable if its buffer contains data for all of its input
     * source topic partitions, or if it is enforced to be processable.
     */
    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) {