@Test
    public void testFencedLeaderRecovery() throws Exception {
        connectBuilder.numWorkers(1);
        // Much slower offset commit interval; should never be triggered during this test
        workerProps.put(WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_CONFIG, "600000");
        startConnect();

        String topic = "test-topic";
        connect.kafka().createTopic(topic, 3);

        int numTasks = 1;

        Map<String, String> props = new HashMap<>();
        props.put(CONNECTOR_CLASS_CONFIG, MonitorableSourceConnector.class.getName());
        props.put(TASKS_MAX_CONFIG, Integer.toString(numTasks));
        props.put(TOPIC_CONFIG, topic);
        props.put(KEY_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());
        props.put(VALUE_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());
        props.put(NAME_CONFIG, CONNECTOR_NAME);
        props.put(TRANSACTION_BOUNDARY_CONFIG, POLL.toString());
        props.put(MESSAGES_PER_POLL_CONFIG, MESSAGES_PER_POLL);
        props.put(MAX_MESSAGES_PER_SECOND_CONFIG, MESSAGES_PER_SECOND);

        // expect all records to be consumed and committed by the connector
        connectorHandle.expectedRecords(MINIMUM_MESSAGES);
        connectorHandle.expectedCommits(MINIMUM_MESSAGES);

        // fence out the leader of the cluster
        Producer<?, ?> zombieLeader = transactionalProducer(
                "simulated-zombie-leader",
                DistributedConfig.transactionalProducerId(CLUSTER_GROUP_ID)
        );
        zombieLeader.initTransactions();
        zombieLeader.close();

        // start a source connector--should fail the first time
        assertThrows(ConnectRestException.class, () -> connect.configureConnector(CONNECTOR_NAME, props));

        // the second request should succeed because the leader has reclaimed write privileges for the config topic
        connect.configureConnector(CONNECTOR_NAME, props);

        log.info("Waiting for records to be provided to worker by task");
        // wait for the connector tasks to produce enough records
        connectorHandle.awaitRecords(SOURCE_TASK_PRODUCE_TIMEOUT_MS);

        log.info("Waiting for records to be committed to Kafka by worker");
        // wait for the connector tasks to commit enough records
        connectorHandle.awaitCommits(TimeUnit.MINUTES.toMillis(1));

        StartAndStopLatch connectorStop = connectorHandle.expectedStops(1, true);
        connect.deleteConnector(CONNECTOR_NAME);
        assertConnectorStopped(connectorStop);

        // consume all records from the source topic or fail, to ensure that they were correctly produced
        ConsumerRecords<byte[], byte[]> records = connect.kafka().consumeAll(
                CONSUME_RECORDS_TIMEOUT_MS,
                Collections.singletonMap(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed"),
                null,
                topic
        );
        assertTrue(records.count() >= MINIMUM_MESSAGES,
                "Not enough records produced by source connector. Expected at least: " + MINIMUM_MESSAGES + " + but got " + records.count());
        assertExactlyOnceSeqnos(records, numTasks);
    }
Brings up a one-node cluster, then intentionally fences out the transactional producer used by the leader
for writes to the config topic to simulate a zombie leader being active in the cluster. The leader should
automatically recover, verify that it is still the leader, and then succeed to create a connector when the
user resends the request.
connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExactlyOnceSourceIntegrationTest.java