private KafkaProducer<byte[], byte[]> transactionalProducer(String clientId, String transactionalId) {
        Map<String, Object> transactionalProducerProps = new HashMap<>();
        transactionalProducerProps.put(CLIENT_ID_CONFIG, clientId);
        transactionalProducerProps.put(ENABLE_IDEMPOTENCE_CONFIG, true);
        transactionalProducerProps.put(TRANSACTIONAL_ID_CONFIG, transactionalId);
        return connect.kafka().createProducer(transactionalProducerProps);
    }
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