@SuppressWarnings("unchecked")
    RestartRequest recordToRestartRequest(ConsumerRecord<String, byte[]> record, SchemaAndValue value) {
        String connectorName = record.key().substring(RESTART_PREFIX.length());
        if (!(value.value() instanceof Map)) {
            log.error("Ignoring restart request because the value is not a Map but is {}", className(value.value()));
            return null;
        }

        Map<String, Object> valueAsMap = (Map<String, Object>) value.value();

        Object failed = valueAsMap.get(ONLY_FAILED_FIELD_NAME);
        boolean onlyFailed;
        if (!(failed instanceof Boolean)) {
            log.warn("Invalid data for restart request '{}' field should be a Boolean but is {}, defaulting to {}", ONLY_FAILED_FIELD_NAME, className(failed), ONLY_FAILED_DEFAULT);
            onlyFailed = ONLY_FAILED_DEFAULT;
        } else {
            onlyFailed = (Boolean) failed;
        }

        Object withTasks = valueAsMap.get(INCLUDE_TASKS_FIELD_NAME);
        boolean includeTasks;
        if (!(withTasks instanceof Boolean)) {
            log.warn("Invalid data for restart request '{}' field should be a Boolean but is {}, defaulting to {}", INCLUDE_TASKS_FIELD_NAME, className(withTasks), INCLUDE_TASKS_DEFAULT);
            includeTasks = INCLUDE_TASKS_DEFAULT;
        } else {
            includeTasks = (Boolean) withTasks;
        }
        return new RestartRequest(connectorName, onlyFailed, includeTasks);
    }
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