public void awaitRecords(long timeout, TimeUnit unit) throws InterruptedException {
        if (recordsRemainingLatch == null) {
            throw new IllegalStateException("Illegal state encountered. expectedRecords() was not set for this task?");
        }
        if (!recordsRemainingLatch.await(timeout, unit)) {
            String msg = String.format(
                    "Insufficient records seen by task %s in %d millis. Records expected=%d, actual=%d",
                    taskId,
                    unit.toMillis(timeout),
                    expectedRecords,
                    expectedRecords - recordsRemainingLatch.getCount());
            throw new DataException(msg);
        }
        log.debug("Task {} saw {} records, expected {} records",
                  taskId, expectedRecords - recordsRemainingLatch.getCount(), expectedRecords);
    }
Wait up to the specified timeout for this task to meet the expected number of records as
defined by {@code expectedRecords}.

@param timeout duration to wait for records
@param unit    the unit of duration; may not be null
@throws InterruptedException if another threads interrupts this one while waiting for records
connect/runtime/src/test/java/org/apache/kafka/connect/integration/TaskHandle.java