private void awaitPolls(int minimum, List<SourceRecord> records) {
        pollRecords.set(records);
        pollLatch.set(new CountDownLatch(minimum));
        ConcurrencyUtils.awaitLatch(pollLatch.get(), "task was not polled " + minimum + " time(s) quickly enough");
    }
@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