private void verifyShutdown(boolean taskFailed, boolean cancelled) throws Exception {
        if (cancelled) {
            verify(producer).close(Duration.ZERO);
            verify(producer, times(2)).close(any());
            verify(offsetReader, times(2)).close();
        } else {
            verify(producer).close(any());
            verify(offsetReader).close();
        }

        verify(offsetStore).stop();
        verify(admin).close(any());
        verify(transformationChain).close();
        verify(headerConverter).close();

        if (taskStarted) {
            verify(sourceTask).stop();
        }

        if (taskFailed) {
            verify(statusListener).onFailure(eq(taskId), any());
        } else if (!cancelled) {
            verify(statusListener).onShutdown(taskId);
        }
    }
@return how many times an {@link OffsetStorageWriter#doFlush(Callback) offset flush} took place
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ExactlyOnceWorkerSourceTaskTest.java