@Override
    public void putSessionKey(SessionKey sessionKey) {
        log.debug("Distributing new session key");
        Struct sessionKeyStruct = new Struct(SESSION_KEY_V0);
        sessionKeyStruct.put("key", Base64.getEncoder().encodeToString(sessionKey.key().getEncoded()));
        sessionKeyStruct.put("algorithm", sessionKey.key().getAlgorithm());
        sessionKeyStruct.put("creation-timestamp", sessionKey.creationTimestamp());
        byte[] serializedSessionKey = converter.fromConnectData(topic, SESSION_KEY_V0, sessionKeyStruct);
        try {
            Timer timer = time.timer(READ_WRITE_TOTAL_TIMEOUT_MS);
            sendPrivileged(SESSION_KEY_KEY, serializedSessionKey, timer);
            configLog.readToEnd().get(timer.remainingMs(), TimeUnit.MILLISECONDS);
        } catch (InterruptedException | ExecutionException | TimeoutException e) {
            log.error("Failed to write session key to Kafka: ", e);
            throw new ConnectException("Error writing session key to Kafka", e);
        }
    }
Write a session key 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 sessionKey the session key to distributed
@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