private void startOrReuseConnectWithNumWorkers(int numWorkers) throws Exception {
        connect = CONNECT_CLUSTERS.computeIfAbsent(numWorkers, n -> {
            // setup Connect worker properties
            Map<String, String> workerProps = new HashMap<>();
            workerProps.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
            workerProps.put(CONNECTOR_CLIENT_POLICY_CLASS_CONFIG, "All");

            // setup Kafka broker properties
            Properties brokerProps = new Properties();
            brokerProps.put("auto.create.topics.enable", String.valueOf(false));

            EmbeddedConnectCluster.Builder connectBuilder = new EmbeddedConnectCluster.Builder()
                    .name("connect-cluster")
                    .numWorkers(numWorkers)
                    .workerProps(workerProps)
                    .brokerProps(brokerProps)
                    // true is the default, setting here as example
                    .maskExitProcedures(true);
            EmbeddedConnectCluster connect = connectBuilder.build();
            // start the clusters
            connect.start();
            return connect;
        });
    }
Test connectors restart API use cases.
connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorRestartApiIntegrationTest.java