private void processLoggerLevelRecord(String namespace, SchemaAndValue value) {
        if (value.value() == null) {
            log.error("Ignoring logging level for namespace {} because it is unexpectedly null", namespace);
            return;
        }
        if (!(value.value() instanceof Map)) {
            log.error("Ignoring logging level for namespace {} because the value is not a Map but is {}", namespace, className(value.value()));
            return;
        }

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

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

        if (started) {
            updateListener.onLoggingLevelUpdate(namespace, (String) level);
        } else {
            // TRACE level since there may be many of these records in the config topic
            log.trace(
                    "Ignoring old logging level {} for namespace {} that was writen to the config topic before this worker completed startup",
                    level,
                    namespace
            );
        }
    }
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