@Override
    public void removeConnectorConfig(String connector) {
        log.debug("Removing connector configuration for connector '{}'", connector);
        try {
            Timer timer = time.timer(READ_WRITE_TOTAL_TIMEOUT_MS);
            List<ProducerKeyValue> keyValues = Arrays.asList(
                    new ProducerKeyValue(CONNECTOR_KEY(connector), null),
                    new ProducerKeyValue(TARGET_STATE_KEY(connector), null)
            );
            sendPrivileged(keyValues, timer);

            configLog.readToEnd().get(timer.remainingMs(), TimeUnit.MILLISECONDS);
        } catch (InterruptedException | ExecutionException | TimeoutException e) {
            log.error("Failed to remove connector configuration from Kafka: ", e);
            throw new ConnectException("Error removing connector configuration from Kafka", e);
        }
    }
Remove configuration for a given connector. {@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 remove
@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