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 testNamedTopologiesCanBePausedIndependently() {
        final Set<String> pausedTopologies = new HashSet<>();
        final TaskExecutionMetadata metadata = new TaskExecutionMetadata(NAMED_TOPOLOGIES, pausedTopologies, ProcessingMode.AT_LEAST_ONCE);

        final Task mockTask1 = createMockTask(TOPOLOGY1);
        final Task mockTask2 = createMockTask(TOPOLOGY2);

        assertTrue(metadata.canProcessTask(mockTask1, TIME_ZERO));
        assertTrue(metadata.canProcessTask(mockTask2, TIME_ZERO));

        pausedTopologies.add(TOPOLOGY1);
        assertFalse(metadata.canProcessTask(mockTask1, TIME_ZERO));
        assertTrue(metadata.canProcessTask(mockTask2, TIME_ZERO));

        pausedTopologies.remove(TOPOLOGY1);
        assertTrue(metadata.canProcessTask(mockTask1, TIME_ZERO));
        assertTrue(metadata.canProcessTask(mockTask2, TIME_ZERO));
    }