private void failedConnectorRestart(boolean onlyFailed, boolean includeTasks, int expectedConnectorRestarts) throws Exception {
        //as connector is failed we expect 0 task to be started
        Map<String, Integer> expectedTasksStarts = allTasksExpectedRestarts(0);

        // setup up props for the source connector
        Map<String, String> props = defaultSourceConnectorProps(TOPIC_NAME);
        props.put("connector.start.inject.error", "true");
        // build a Connect cluster backed by a Kafka KRaft cluster
        startOrReuseConnectWithNumWorkers(ONE_WORKER);

        // Try to start the connector and its single task.
        connect.configureConnector(connectorName, props);

        connect.assertions().assertConnectorIsFailedAndTasksHaveFailed(connectorName, 0,
                "Connector or tasks are in running state.");

        StartsAndStops beforeSnapshot = connectorHandle.startAndStopCounter().countsSnapshot();

        StartAndStopLatch startLatch = connectorHandle.expectedStarts(expectedConnectorRestarts, expectedTasksStarts, includeTasks);

        // Call the Restart API
        connect.restartConnectorAndTasks(connectorName, onlyFailed, includeTasks, false);

        connect.assertions().assertConnectorIsFailedAndTasksHaveFailed(connectorName, 0,
                "Connector tasks are not all in running state.");
        // Expect that the connector has started again
        assertTrue(startLatch.await(CONNECTOR_SETUP_DURATION_MS, TimeUnit.MILLISECONDS),
                "Failed to start connector and tasks after coordinator failure within "
                        + CONNECTOR_SETUP_DURATION_MS + "ms");
        StartsAndStops afterSnapshot = connectorHandle.startAndStopCounter().countsSnapshot();

        assertEquals(beforeSnapshot.starts() + expectedConnectorRestarts, afterSnapshot.starts());
    }
Test connectors restart API use cases.
connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorRestartApiIntegrationTest.java