@Test
    public void testBrokerCoordinator() throws Exception {
        ConnectorHandle connectorHandle = RuntimeHandles.get().connectorHandle(CONNECTOR_NAME);
        workerProps.put(DistributedConfig.SCHEDULED_REBALANCE_MAX_DELAY_MS_CONFIG, String.valueOf(5000));

        useFixedBrokerPort();

        // start the clusters
        connect = connectBuilder.build();
        connect.start();
        int numTasks = 4;
        // create test topic
        connect.kafka().createTopic(TOPIC_NAME, NUM_TOPIC_PARTITIONS);

        // set 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, numTasks,
                "Connector tasks did not start in time.");

        // expect that the connector will be stopped once the coordinator is detected to be down
        StartAndStopLatch stopLatch = connectorHandle.expectedStops(1, false);

        connect.kafka().stopOnlyBrokers();

        // Allow for the workers to discover that the coordinator is unavailable, wait is
        // heartbeat timeout * 2 + 4sec
        Thread.sleep(TimeUnit.SECONDS.toMillis(10));

        connect.requestTimeout(1000);
        assertFalse(
                connect.anyWorkersHealthy(),
                "No workers should be healthy when underlying Kafka cluster is down"
        );
        connect.workers().forEach(worker -> {
            try (Response response = connect.healthCheck(worker)) {
                assertEquals(INTERNAL_SERVER_ERROR.getStatusCode(), response.getStatus());
                assertNotNull(response.getEntity());
                String body = response.getEntity().toString();
                String expectedSubstring = "Worker was unable to handle this request and may be unable to handle other requests";
                assertTrue(
                        body.contains(expectedSubstring),
                        "Response body '" + body + "' did not contain expected message '" + expectedSubstring + "'"
                );
            }
        });
        connect.resetRequestTimeout();

        // Wait for the connector to be stopped
        assertTrue(stopLatch.await(CONNECTOR_SETUP_DURATION_MS, TimeUnit.MILLISECONDS),
                "Failed to stop connector and tasks after coordinator failure within "
                        + CONNECTOR_SETUP_DURATION_MS + "ms");

        StartAndStopLatch startLatch = connectorHandle.expectedStarts(1, false);
        connect.kafka().restartOnlyBrokers();

        // Allow for the kafka brokers to come back online
        Thread.sleep(TimeUnit.SECONDS.toMillis(10));

        connect.assertions().assertExactlyNumWorkersAreUp(NUM_WORKERS,
                "Group of workers did not remain the same within the designated time.");

        // Allow for the workers to rebalance and reach a steady state
        Thread.sleep(TimeUnit.SECONDS.toMillis(10));

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

        // Expect that the connector has started again
        assertTrue(startLatch.await(CONNECTOR_SETUP_DURATION_MS, TimeUnit.MILLISECONDS),
                "Failed to stop connector and tasks after coordinator failure within "
                        + CONNECTOR_SETUP_DURATION_MS + "ms");
    }
Verify that a set of tasks restarts correctly after a broker goes offline and back online
connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectWorkerIntegrationTest.java