@Override
    public void putRestartRequest(RestartRequest restartRequest) {
        log.debug("Writing {} to Kafka", restartRequest);
        String key = RESTART_KEY(restartRequest.connectorName());
        Struct value = new Struct(RESTART_REQUEST_V0);
        value.put(INCLUDE_TASKS_FIELD_NAME, restartRequest.includeTasks());
        value.put(ONLY_FAILED_FIELD_NAME, restartRequest.onlyFailed());
        byte[] serializedValue = converter.fromConnectData(topic, value.schema(), value);
        try {
            Timer timer = time.timer(READ_WRITE_TOTAL_TIMEOUT_MS);
            sendPrivileged(key, serializedValue, timer);
            configLog.readToEnd().get(timer.remainingMs(), TimeUnit.MILLISECONDS);
        } catch (InterruptedException | ExecutionException | TimeoutException e) {
            log.error("Failed to write {} to Kafka: ", restartRequest, e);
            throw new ConnectException("Error writing " + restartRequest + " to Kafka", e);
        }
    }
Write a restart request for the connector and optionally its tasks to persistent storage and wait until it has been
acknowledged and read back by tailing the Kafka log with a consumer. {@link #claimWritePrivileges()} must be successfully
invoked before calling this method if the worker is configured to use a fencable producer for writes to the config topic.
@param restartRequest the restart request details
connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java