private void processSessionKeyRecord(SchemaAndValue value) {
        if (value.value() == null) {
            log.error("Ignoring session key because it is unexpectedly null");
            return;
        }
        if (!(value.value() instanceof Map)) {
            log.error("Ignoring session key because the value is not a Map but is {}", className(value.value()));
            return;
        }

        @SuppressWarnings("unchecked")
        Map<String, Object> valueAsMap = (Map<String, Object>) value.value();

        Object sessionKey = valueAsMap.get("key");
        if (!(sessionKey instanceof String)) {
            log.error("Invalid data for session key 'key' field should be a String but is {}", className(sessionKey));
            return;
        }
        byte[] key = Base64.getDecoder().decode((String) sessionKey);

        Object keyAlgorithm = valueAsMap.get("algorithm");
        if (!(keyAlgorithm instanceof String)) {
            log.error("Invalid data for session key 'algorithm' field should be a String but it is {}", className(keyAlgorithm));
            return;
        }

        Object creationTimestamp = valueAsMap.get("creation-timestamp");
        if (!(creationTimestamp instanceof Long)) {
            log.error("Invalid data for session key 'creation-timestamp' field should be a long but it is {}", className(creationTimestamp));
            return;
        }
        KafkaConfigBackingStore.this.sessionKey = new SessionKey(
                new SecretKeySpec(key, (String) keyAlgorithm),
                (long) creationTimestamp
        );

        if (started)
            updateListener.onSessionKeyUpdate(KafkaConfigBackingStore.this.sessionKey);
    }
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