private void sendPrivileged(List<ProducerKeyValue> keyValues, Timer timer) throws ExecutionException, InterruptedException, TimeoutException {
        if (!usesFencableWriter) {
            List<Future<RecordMetadata>> producerFutures = new ArrayList<>();
            keyValues.forEach(
                    keyValue -> producerFutures.add(configLog.sendWithReceipt(keyValue.key, keyValue.value))
            );

            timer.update();
            for (Future<RecordMetadata> future : producerFutures) {
                future.get(timer.remainingMs(), TimeUnit.MILLISECONDS);
                timer.update();
            }

            return;
        }

        if (fencableProducer == null) {
            throw new IllegalStateException("Cannot produce to config topic without claiming write privileges first");
        }

        try {
            fencableProducer.beginTransaction();
            keyValues.forEach(
                    keyValue -> fencableProducer.send(new ProducerRecord<>(topic, keyValue.key, keyValue.value))
            );
            fencableProducer.commitTransaction();
            timer.update();
        } catch (Exception e) {
            log.warn("Failed to perform fencable send to config topic", e);
            relinquishWritePrivileges();
            throw new PrivilegedWriteException("Failed to perform fencable send to config topic", e);
        }
    }
Send one or more records to the config topic synchronously. Note that {@link #claimWritePrivileges()} must be
successfully invoked before calling this method if this store is configured to use a fencable writer.
@param keyValues the list of producer record key/value pairs
@param timer Timer bounding how long this method can block. The timer is updated before the method returns.
connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java