public ConnectorStateInfo restartConnectorAndTasks(String connName, boolean onlyFailed, boolean includeTasks, boolean onlyCallOnEmptyWorker) {
        ObjectMapper mapper = new ObjectMapper();
        String restartPath = String.format("connectors/%s/restart?onlyFailed=" + onlyFailed + "&includeTasks=" + includeTasks, connName);
        String restartEndpoint;
        if (onlyCallOnEmptyWorker) {
            restartEndpoint = endpointForResourceNotRunningConnector(restartPath, connName);
        } else {
            restartEndpoint = endpointForResource(restartPath);
        }
        Response response = requestPost(restartEndpoint, "", Collections.emptyMap());
        try {
            if (response.getStatus() < Response.Status.BAD_REQUEST.getStatusCode()) {
                //only the 202 stauts returns a body
                if (response.getStatus() == Response.Status.ACCEPTED.getStatusCode()) {
                    return mapper.readerFor(ConnectorStateInfo.class)
                            .readValue(responseToString(response));
                }
            }
            return null;
        } catch (IOException e) {
            log.error("Could not read connector state from response: {}",
                    responseToString(response), e);
            throw new ConnectException("Could not not parse connector state", e);
        }
    }
Restart an existing connector and its tasks.

@param connName  name of the connector to be restarted
@param onlyFailed    true if only failed instances should be restarted
@param includeTasks  true if tasks should be restarted, or false if only the connector should be restarted
@param onlyCallOnEmptyWorker true if the REST API call should be called on a worker not running this connector or its tasks
@throws ConnectRestException if the REST API returns error status
@throws ConnectException for any other error.
connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedConnect.java