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

Code before:
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import static org.apache.kafka.streams.processor.internals.TopologyMetadata.UNNAMED_TOPOLOGY;

/**
 * Multi-threaded class that tracks the status of active tasks being processed. A single instance of this class is
 * shared between all StreamThreads.
 */
public class TaskExecutionMetadata {
    // TODO: implement exponential backoff, for now we just wait 5s
    private static final long CONSTANT_BACKOFF_MS = 5_000L;
    private static final long NOT_READY_LOG_INTERVAL_MS = 120_000L; // Log interval of not being ready (2 minutes)

    private final boolean hasNamedTopologies;
    private final Set<String> pausedTopologies;
    private final ProcessingMode processingMode;
    private final Collection<Task> successfullyProcessed = new HashSet<>();
    // map of topologies experiencing errors/currently under backoff
    private final ConcurrentHashMap<String, NamedTopologyMetadata> topologyNameToErrorMetadata = new ConcurrentHashMap<>();
    // map of task to last not ready logging time (not present means ready, >= 0 means not ready and tracks last log time)
    private final ConcurrentHashMap<TaskId, Long> taskToLastNotReadyLogTime = new ConcurrentHashMap<>();
    // map of task to time when idling started
    private final ConcurrentHashMap<TaskId, Long> taskToIdlingStartTime = 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;

Review comment (suggestion):
I don't think we need this idle-timer here -- the difference is, that a user can call `KafkaStreams#pause()`, what we track here.

The idle-time inside a task, has a different purpose: it's for the case that there is more than one input topic partition for a task, and the task cannot make progress because one partitions has no data. For this case, we don't want to block the task forever, and unblock it to process the data we have after "max idle time" passed.

But if a task is `paused()`, we are not "idel" waiting for new input data.

Code after:
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import static org.apache.kafka.streams.processor.internals.TopologyMetadata.UNNAMED_TOPOLOGY;

/**
 * Multi-threaded class that tracks the status of active tasks being processed. A single instance of this class is
 * shared between all StreamThreads.
 */
public class TaskExecutionMetadata {
    // TODO: implement exponential backoff, for now we just wait 5s
    private static final long CONSTANT_BACKOFF_MS = 5_000L;
    private static final long NOT_READY_LOG_INTERVAL_MS = 120_000L; // Log interval of not being ready (2 minutes)

    private final boolean hasNamedTopologies;
    private final Set<String> pausedTopologies;
    private final ProcessingMode processingMode;
    private final Collection<Task> successfullyProcessed = new HashSet<>();
    // map of topologies experiencing errors/currently under backoff
    private final ConcurrentHashMap<String, NamedTopologyMetadata> topologyNameToErrorMetadata = new ConcurrentHashMap<>();
    // map of task to last not ready logging time (not present means ready, >= 0 means not ready and tracks last log time)
    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)