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

Code before:
    }

    public boolean hasNamedTopologies() {
        return hasNamedTopologies;
    }

    public ProcessingMode processingMode() {
        return processingMode;
    }

    public boolean canProcessTask(final Task task, final long now) {
        final String topologyName = task.id().topologyName();
        final boolean taskWasReady = !taskToLastNotReadyLogTime.containsKey(task.id());
        final boolean canProcess;
        final String logMessage;
        
        if (!hasNamedTopologies) {
            // TODO implement error handling/backoff for non-named topologies (needs KIP)
            logMessage = String.format("Task %s processing check for unnamed topology '%s'", task.id(), topologyName);
            canProcess = !pausedTopologies.contains(UNNAMED_TOPOLOGY);
        } else {
            if (pausedTopologies.contains(topologyName)) {
                canProcess = false;
                logMessage = String.format("Task %s can't be processed: topology '%s' is paused", task.id(), topologyName);
            } else {
                logMessage = String.format("Task %s processing check for named topology '%s'", task.id(), topologyName);
                final NamedTopologyMetadata metadata = topologyNameToErrorMetadata.get(topologyName);
                canProcess = metadata == null || (metadata.canProcess() && metadata.canProcessTask(task, now));
            }
        }
        
        if (!canProcess) {
            if (taskWasReady) {
                // READY -> NOT_READY - start timer
                taskToLastNotReadyLogTime.put(task.id(), now);
            } else {
                // NOT_READY - check if it should log
                maybeLogNotReady(task.id(), now, logMessage);
            }
        } else {
            // Task is ready - clear the timer
            taskToLastNotReadyLogTime.remove(task.id());
            log.trace(logMessage);
        }
        
        return canProcess;
    }
    
    private void maybeLogNotReady(final TaskId taskId, final long now, final String logMessage) {
        final Long lastLogTime = taskToLastNotReadyLogTime.get(taskId);
        if (lastLogTime == null) {

Review comment (refactor):
As above. Should we move this after `canProcess = metadata == null || (metadata.canProcess() && metadata.canProcessTask(task, now));` and have a different message depending if `canProcess` is true vs false?

Code after:
    }

    public boolean hasNamedTopologies() {
        return hasNamedTopologies;
    }

    public ProcessingMode processingMode() {
        return processingMode;
    }

    public boolean canProcessTask(final Task task, final long now) {
        final String topologyName = task.id().topologyName();
        final boolean taskWasReady = !taskToLastNotReadyLogTime.containsKey(task.id());
        final boolean canProcess;
        final String logMessage;
        
        if (!hasNamedTopologies) {
            // TODO implement error handling/backoff for non-named topologies (needs KIP)
            canProcess = !pausedTopologies.contains(UNNAMED_TOPOLOGY);
            if (canProcess) {
                logMessage = String.format("Task %s can be processed: topology is not paused", task.id());
            } else {
                logMessage = String.format("Task %s can't be processed: topology is paused", task.id());
            }
        } else {
            if (pausedTopologies.contains(topologyName)) {
                canProcess = false;
                logMessage = String.format("Task %s can't be processed: topology '%s' is paused", task.id(), topologyName);
            } else {
                final NamedTopologyMetadata metadata = topologyNameToErrorMetadata.get(topologyName);
                canProcess = metadata == null || (metadata.canProcess() && metadata.canProcessTask(task, now));
                if (canProcess) {
                    logMessage = String.format("Task %s can be processed for named topology '%s'", task.id(), topologyName);
                } else {
                    logMessage = String.format("Task %s can't be processed for named topology '%s'", task.id(), topologyName);
                }
            }
        }
        
        if (!canProcess) {
            if (taskWasReady) {
                // READY -> NOT_READY - start timer
                taskToLastNotReadyLogTime.put(task.id(), now);
            } else {
                // NOT_READY - check if it should log
                maybeLogNotReady(task.id(), now, logMessage);
            }
        } else {
            // Task is ready - clear the timer
            taskToLastNotReadyLogTime.remove(task.id());
            log.trace(logMessage);