private void assertProducersAreFencedOnReconfiguration(
            int currentNumTasks,
            int newNumTasks,
            String topic,
            Map<String, String> baseConnectorProps) throws InterruptedException {

        // create a collection of producers that simulate the producers used for the existing tasks
        List<KafkaProducer<byte[], byte[]>> producers = IntStream.range(0, currentNumTasks)
                .mapToObj(i -> transactionalProducer(
                        "simulated-task-producer-" + CONNECTOR_NAME + "-" + i,
                        Worker.taskTransactionalId(CLUSTER_GROUP_ID, CONNECTOR_NAME, i)
                )).collect(Collectors.toList());

        producers.forEach(KafkaProducer::initTransactions);

        // reconfigure the connector with a new number of tasks
        StartAndStopLatch connectorStart = connectorAndTaskStart(newNumTasks);
        baseConnectorProps.put(TASKS_MAX_CONFIG, Integer.toString(newNumTasks));
        log.info("Reconfiguring connector from {} tasks to {}", currentNumTasks, newNumTasks);
        connect.configureConnector(CONNECTOR_NAME, baseConnectorProps);
        assertConnectorStarted(connectorStart);

        // validate that the old producers were fenced out
        producers.forEach(producer -> assertTransactionalProducerIsFenced(producer, topic));
    }
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