@Override
    public void putConnectorConfig(String connector, Map<String, String> properties, TargetState targetState) {
        log.debug("Writing connector configuration for connector '{}'", connector);
        Struct connectConfig = new Struct(CONNECTOR_CONFIGURATION_V0);
        connectConfig.put("properties", properties);
        byte[] serializedConfig = converter.fromConnectData(topic, CONNECTOR_CONFIGURATION_V0, connectConfig);
        try {
            Timer timer = time.timer(READ_WRITE_TOTAL_TIMEOUT_MS);
            List<ProducerKeyValue> keyValues = new ArrayList<>();
            if (targetState != null) {
                log.debug("Writing target state {} for connector {}", targetState, connector);
                keyValues.add(new ProducerKeyValue(TARGET_STATE_KEY(connector), serializeTargetState(targetState)));
            }
            keyValues.add(new ProducerKeyValue(CONNECTOR_KEY(connector), serializedConfig));
            sendPrivileged(keyValues, timer);
            configLog.readToEnd().get(timer.remainingMs(), TimeUnit.MILLISECONDS);
        } catch (InterruptedException | ExecutionException | TimeoutException e) {
            log.error("Failed to write connector configuration to Kafka: ", e);
            throw new ConnectException("Error writing connector configuration to Kafka", e);
        }
    }
Write this connector configuration (and optionally a target state) 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 connector  name of the connector to write data for
@param properties the configuration to write
@param targetState the desired target state for the connector; may be {@code null} if no target state change is desired. Note that the default
                   target state is {@link TargetState#STARTED} if no target state exists previously
@throws IllegalStateException if {@link #claimWritePrivileges()} is required, but was not successfully invoked before
this method was called
@throws PrivilegedWriteException if the worker is configured to use a fencable producer for writes to the config topic
and the write fails
connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java