@Test
    public void testRemovingWorker() throws Exception {
        // create test topic
        connect.kafka().createTopic(TOPIC_NAME, NUM_TOPIC_PARTITIONS);

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

        // start a source connector
        IntStream.range(0, 4).forEachOrdered(i -> connect.configureConnector(CONNECTOR_NAME + i, props));

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

        connect.removeWorker();

        connect.assertions().assertExactlyNumWorkersAreUp(NUM_WORKERS - 1,
                "Connect workers did not start in time.");

        waitForCondition(this::assertConnectorAndTasksAreUniqueAndBalanced,
                WORKER_SETUP_DURATION_MS, "Connect and tasks are imbalanced between the workers.");
    }
Integration tests for incremental cooperative rebalancing between Connect workers
connect/runtime/src/test/java/org/apache/kafka/connect/integration/RebalanceSourceConnectorsIntegrationTest.java