private void verifyTransactions(int numTaskCommits, int numProducerCommits) throws InterruptedException {
        // these operations happen on every commit opportunity
        VerificationMode commitOpportunities = times(numTaskCommits);
        verify(offsetWriter, commitOpportunities).beginFlush();
        verify(sourceTask, commitOpportunities).commit();
        // these operations only happen on non-empty commits
        VerificationMode commits = times(numProducerCommits);
        verify(producer, commits).beginTransaction();
        verify(producer, commits).commitTransaction();
        verify(offsetWriter, commits).doFlush(any());
    }
@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