@Override
    public void start() {
        if (taskExecutorThread == null) {
            taskExecutorThread = new TaskExecutorThread(name);
            taskExecutorThread.start();
            shutdownGate = new CountDownLatch(1);
        }
    }
@Test
    public void shouldNotFlushOnException() {
        final StreamsException exception = mock(StreamsException.class);
        when(task.process(anyLong())).thenThrow(exception);
        when(taskManager.hasUncaughtException(task.id())).thenReturn(true);

        taskExecutor.start();

        verify(taskManager, timeout(VERIFICATION_TIMEOUT)).unassignTask(task, taskExecutor);
        verify(task, never()).flush();
    }