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)
            return !pausedTopologies.contains(UNNAMED_TOPOLOGY);
        } else {
            if (pausedTopologies.contains(topologyName)) {
                return false;
            } else {
                final NamedTopologyMetadata metadata = topologyNameToErrorMetadata.get(topologyName);
                return metadata == null || (metadata.canProcess() && metadata.canProcessTask(task, now));
            }
        }
    }
@Test
    public void testCanProcessWithoutNamedTopologies() {
        final Set<String> topologies = Collections.singleton(UNNAMED_TOPOLOGY);
        final Set<String> pausedTopologies = new HashSet<>();

        final TaskExecutionMetadata metadata = new TaskExecutionMetadata(topologies, pausedTopologies, ProcessingMode.AT_LEAST_ONCE);

        final Task mockTask = createMockTask(UNNAMED_TOPOLOGY);

        assertTrue(metadata.canProcessTask(mockTask, TIME_ZERO));
        // This pauses an UNNAMED_TOPOLOGY / a KafkaStreams instance without named/modular
        // topologies.
        pausedTopologies.add(UNNAMED_TOPOLOGY);
        assertFalse(metadata.canProcessTask(mockTask, TIME_ZERO));
    }