private void assertTransactionalProducerIsFenced(KafkaProducer<byte[], byte[]> producer, String topic) {
        producer.beginTransaction();
        assertThrows(
                ProducerFencedException.class,
                () -> {
                    producer.send(new ProducerRecord<>(topic, new byte[] {69}, new byte[] {96}));
                    producer.commitTransaction();
                },
                "Producer should be fenced out"
        );
        producer.close(Duration.ZERO);
    }
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