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

Code before:
                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)
     */
    @SuppressWarnings("unchecked")
    public boolean process(final long wallClockTime) {
        if (record == null) {
            if (!isProcessable(wallClockTime)) {
                return false;
            }

            // get the next record to process
            record = partitionGroup.nextRecord(recordInfo, wallClockTime);

            // if there is no record to process, return immediately
            if (record == null) {
                log.trace("Task has no next record to process.");

Review comment (suggestion):
```suggestion
        final long timeSinceLastLog = lastNotReadyLogTime.map(aLong -> wallClockTime - aLong).orElse(-1L);
```
We never logged, so `-1L` is maybe a little be more expressive -- does not really change much; just seems to be a tiny bit cleaner.

Code after:
            // 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) {
        if (record == null) {
            if (!isProcessable(wallClockTime)) {
                return false;
            }

            // get the next record to process