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

Code before:

        record = null;
        closeTaskSensor.record();
        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("Stream task {} is already in {} state, skip processing it.", id(), state());

            return false;
        }

        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 (log.isDebugEnabled()) {
                log.debug("Stream task {} has a pending transaction commit, skip processing it.", id());
            }
            return false;
        }
        final boolean readyToProcess = partitionGroup.readyToProcess(wallClockTime);
        if (!readyToProcess) {
            if (timeCurrentIdlingStarted.isEmpty()) {
                timeCurrentIdlingStarted = Optional.of(wallClockTime);
            }
            if (log.isDebugEnabled()) {
                log.debug("Task {} started idling at time {}", id, timeCurrentIdlingStarted.get());
            }
        } else {
            timeCurrentIdlingStarted = Optional.empty();
        }
        return readyToProcess;
    }

    /**
     * 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")

Review comment (bug):
What is the reason for calling `isDebugEnabled()` only at this specific location and not in the former debug level loggings?



Code after:

        record = null;
        closeTaskSensor.record();
        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();