@Override
    public void putTargetState(String connector, TargetState state) {
        log.debug("Writing target state {} for connector {}", state, connector);
        try {
            configLog.sendWithReceipt(TARGET_STATE_KEY(connector), serializeTargetState(state))
                .get(READ_WRITE_TOTAL_TIMEOUT_MS, TimeUnit.MILLISECONDS);
        } catch (InterruptedException | ExecutionException | TimeoutException e) {
            log.error("Failed to write target state to Kafka", e);
            throw new ConnectException("Error writing target state to Kafka", e);
        }
    }
Write a new {@link TargetState} for the connector. Note that {@link #claimWritePrivileges()} does not need to be
invoked before invoking this method.
@param connector the name of the connector
@param state the desired target state for the connector
connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java