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();
        if (!hasNamedTopologies) {
            // TODO implement error handling/backoff for non-named topologies (needs KIP)
            log.trace("Task {} processing check for unnamed topology '{}'", task.id(), topologyName);
            return !pausedTopologies.contains(UNNAMED_TOPOLOGY);
        } else {
            if (pausedTopologies.contains(topologyName)) {
                log.debug("Task {} can't be processed: topology '{}' is paused", task.id(), topologyName);
                return false;
            } else {
                log.trace("Task {} processing check for named topology '{}'", task.id(), topologyName);
                final NamedTopologyMetadata metadata = topologyNameToErrorMetadata.get(topologyName);
                return metadata == null || (metadata.canProcess() && metadata.canProcessTask(task, now));

            }
        }
    }

    public boolean canPunctuateTask(final Task task) {
        final String topologyName = task.id().topologyName();

        if (topologyName == null) {
            return !pausedTopologies.contains(UNNAMED_TOPOLOGY);
        } else {
            return !pausedTopologies.contains(topologyName);
        }
    }

    public void registerTaskError(final Task task, final Throwable t, final long now) {
        if (hasNamedTopologies) {
            final String topologyName = task.id().topologyName();
            topologyNameToErrorMetadata.computeIfAbsent(topologyName, n -> new NamedTopologyMetadata(topologyName))
                .registerTaskError(task, t, now);

Review comment (suggestion):
```suggestion
                log.trace("Task {} can't be processed: topology '{}' is paused", task.id(), topologyName);
```

Code after:
    private final ConcurrentHashMap<TaskId, Long> taskToLastNotReadyLogTime = new ConcurrentHashMap<>();
    private final Logger log;

    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) {