private StartAndStopLatch connectorAndTaskStart(int numTasks) {
        connectorHandle.clearTasks();
        IntStream.range(0, numTasks)
                .mapToObj(i -> MonitorableSourceConnector.taskId(CONNECTOR_NAME, i))
                .forEach(connectorHandle::taskHandle);
        return connectorHandle.expectedStarts(1, true);
    }
Clear all existing task handles for the connector, then preemptively create {@code numTasks} many task handles for it,
and return a {@link StartAndStopLatch} that can be used to {@link StartAndStopLatch#await(long, TimeUnit) await}
the startup of that connector and the expected number of tasks.
@param numTasks the number of tasks that should be started
@return a {@link StartAndStopLatch} that will block until the connector and the expected number of tasks have started
connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExactlyOnceSourceIntegrationTest.java