@Test
    public void testReconfigConnector() throws Exception {
        ConnectorHandle connectorHandle = RuntimeHandles.get().connectorHandle(CONNECTOR_NAME);

        // create test topic
        String anotherTopic = "another-topic";
        connect.kafka().createTopic(TOPIC_NAME, NUM_TOPIC_PARTITIONS);
        connect.kafka().createTopic(anotherTopic, NUM_TOPIC_PARTITIONS);

        // setup up props for the source connector
        Map<String, String> props = defaultSourceConnectorProps(TOPIC_NAME);

        // start a source connector
        connect.configureConnector(CONNECTOR_NAME, props);

        connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(CONNECTOR_NAME, NUM_TASKS,
                "Connector tasks did not start in time.");

        int numRecordsProduced = 100;
        long recordTransferDurationMs = TimeUnit.SECONDS.toMillis(30);

        // consume all records from the source topic or fail, to ensure that they were correctly produced
        int recordNum = connect.kafka().consume(numRecordsProduced, recordTransferDurationMs, TOPIC_NAME).count();
        assertTrue(recordNum >= numRecordsProduced,
                "Not enough records produced by source connector. Expected at least: " + numRecordsProduced + " + but got " + recordNum);

        // expect that we're going to restart the connector and its tasks
        StartAndStopLatch restartLatch = connectorHandle.expectedStarts(1);

        // Reconfigure the source connector by changing the Kafka topic used as output
        props.put(TOPIC_CONFIG, anotherTopic);
        connect.configureConnector(CONNECTOR_NAME, props);

        // Wait for the connector *and tasks* to be restarted
        assertTrue(restartLatch.await(CONNECTOR_SETUP_DURATION_MS, TimeUnit.MILLISECONDS),
                "Failed to alter connector configuration and see connector and tasks restart "
                        + "within " + CONNECTOR_SETUP_DURATION_MS + "ms");

        // And wait for the Connect to show the connectors and tasks are running
        connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(CONNECTOR_NAME, NUM_TASKS,
                "Connector tasks did not start in time.");

        // consume all records from the source topic or fail, to ensure that they were correctly produced
        recordNum = connect.kafka().consume(numRecordsProduced, recordTransferDurationMs, anotherTopic).count();
        assertTrue(recordNum >= numRecordsProduced,
                "Not enough records produced by source connector. Expected at least: " + numRecordsProduced + " + but got " + recordNum);
    }
Integration tests for incremental cooperative rebalancing between Connect workers
connect/runtime/src/test/java/org/apache/kafka/connect/integration/RebalanceSourceConnectorsIntegrationTest.java