@Test
    public void testConnectorReconfiguration() throws Exception {
        // 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);

        Map<String, String> props = new HashMap<>();
        props.put(CONNECTOR_CLASS_CONFIG, MonitorableSourceConnector.class.getName());
        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(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);

        StartAndStopLatch connectorStart = connectorAndTaskStart(3);
        props.put(TASKS_MAX_CONFIG, "3");
        // start a source connector
        connect.configureConnector(CONNECTOR_NAME, props);
        assertConnectorStarted(connectorStart);

        assertProducersAreFencedOnReconfiguration(3, 5, topic, props);
        assertProducersAreFencedOnReconfiguration(5, 1, topic, props);
        assertProducersAreFencedOnReconfiguration(1, 5, topic, props);
        assertProducersAreFencedOnReconfiguration(5, 3, topic, props);

        // Do a final sanity check to make sure that the last generation of tasks is able to run
        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());
        // We used at most five tasks during the tests; each of them should have been able to produce records
        assertExactlyOnceSeqnos(records, 5);
    }
A moderately-complex green-path test that ensures the worker can start up and run tasks for a source
connector that gets reconfigured, and will fence out potential zombie tasks for older generations before
bringing up new task instances.
connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExactlyOnceSourceIntegrationTest.java