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

Code before:
    public TaskExecutionMetadata(final Set<String> allTopologyNames,
                                 final Set<String> pausedTopologies,
                                 final ProcessingMode processingMode) {
        this.hasNamedTopologies = !(allTopologyNames.size() == 1 && allTopologyNames.contains(UNNAMED_TOPOLOGY));
        this.pausedTopologies = pausedTopologies;
        this.processingMode = processingMode;
        this.log = new LogContext("").logger(TaskExecutionMetadata.class);
    }

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

Review comment (suggestion):
```suggestion
            logMessage = String.format("Task %s processing check", task.id());
```

Code after:
    public TaskExecutionMetadata(final Set<String> allTopologyNames,
                                 final Set<String> pausedTopologies,
                                 final ProcessingMode processingMode) {
        this.hasNamedTopologies = !(allTopologyNames.size() == 1 && allTopologyNames.contains(UNNAMED_TOPOLOGY));
        this.pausedTopologies = pausedTopologies;
        this.processingMode = processingMode;
        this.log = new LogContext("").logger(TaskExecutionMetadata.class);
    }

    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 {