@Override
        public void onCompletion(Throwable error, ConsumerRecord<String, byte[]> record) {
            if (error != null) {
                log.error("Unexpected in consumer callback for KafkaConfigBackingStore: ", error);
                return;
            }

            final SchemaAndValue value;
            try {
                value = converter.toConnectData(topic, record.value());
            } catch (DataException e) {
                log.error("Failed to convert config data to Kafka Connect format: ", e);
                return;
            }
            // Make the recorded offset match the API used for positions in the consumer -- return the offset of the
            // *next record*, not the last one consumed.
            offset = record.offset() + 1;

            if (record.key().startsWith(TARGET_STATE_PREFIX)) {
                String connectorName = record.key().substring(TARGET_STATE_PREFIX.length());
                processTargetStateRecord(connectorName, value);
            } else if (record.key().startsWith(CONNECTOR_PREFIX)) {
                String connectorName = record.key().substring(CONNECTOR_PREFIX.length());
                processConnectorConfigRecord(connectorName, value);
            } else if (record.key().startsWith(TASK_PREFIX)) {
                ConnectorTaskId taskId = parseTaskId(record.key());
                if (taskId == null) {
                    log.error("Ignoring task configuration because {} couldn't be parsed as a task config key", record.key());
                    return;
                }
                processTaskConfigRecord(taskId, value);
            } else if (record.key().startsWith(COMMIT_TASKS_PREFIX)) {
                String connectorName = record.key().substring(COMMIT_TASKS_PREFIX.length());
                processTasksCommitRecord(connectorName, value);
            } else if (record.key().startsWith(RESTART_PREFIX)) {
                RestartRequest request = recordToRestartRequest(record, value);
                // Only notify the listener if this backing store is already successfully started (having caught up the first time)
                if (request != null && started) {
                    updateListener.onRestartRequest(request);
                }
            } else if (record.key().startsWith(TASK_COUNT_RECORD_PREFIX)) {
                String connectorName = record.key().substring(TASK_COUNT_RECORD_PREFIX.length());
                processTaskCountRecord(connectorName, value);
            } else if (record.key().equals(SESSION_KEY_KEY)) {
                processSessionKeyRecord(value);
            } else if (record.key().startsWith(LOGGER_CLUSTER_PREFIX)) {
                String loggingNamespace = record.key().substring(LOGGER_CLUSTER_PREFIX.length());
                processLoggerLevelRecord(loggingNamespace, value);
            } else {
                log.error("Discarding config update record with invalid key: {}", record.key());
            }
        }
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