int punctuate() {
        int punctuated = 0;

        for (final Task task : tasks.activeTasks()) {
            try {
                if (executionMetadata.canPunctuateTask(task)) {
                    if (task.maybePunctuateStreamTime()) {
                        punctuated++;
                    }
                    if (task.maybePunctuateSystemTime()) {
                        punctuated++;
                    }
                }
            } catch (final TaskMigratedException e) {
                log.info("Failed to punctuate stream task {} since it got migrated to another thread already. " +
                    "Will trigger a new rebalance and close all tasks as zombies together.", task.id());
                throw e;
            } catch (final StreamsException e) {
                log.error("Failed to punctuate stream task {} due to the following error:", task.id(), e);
                e.setTaskId(task.id());
                throw e;
            } catch (final KafkaException e) {
                log.error("Failed to punctuate stream task {} due to the following error:", task.id(), e);
                throw new StreamsException(e, task.id());
            }
        }

        return punctuated;
    }
@Test
    public void testPunctuateWithPause() {
        final Tasks tasks = mock(Tasks.class);
        final TaskManager taskManager = mock(TaskManager.class);
        final TaskExecutionMetadata metadata = mock(TaskExecutionMetadata.class);

        final TaskExecutor taskExecutor = new TaskExecutor(tasks, taskManager, metadata, new LogContext());

        taskExecutor.punctuate();
        verify(tasks).activeTasks();
    }