@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);
    }
@Test
    public void testRecordToRestartRequest() {
        ConsumerRecord<String, byte[]> record = new ConsumerRecord<>(TOPIC, 0, 0, 0L, TimestampType.CREATE_TIME, 0, 0, RESTART_CONNECTOR_KEYS.get(0),
                CONFIGS_SERIALIZED.get(0), new RecordHeaders(), Optional.empty());
        Struct struct = RESTART_REQUEST_STRUCTS.get(0);
        SchemaAndValue schemaAndValue = new SchemaAndValue(struct.schema(), structToMap(struct));
        RestartRequest restartRequest = configStorage.recordToRestartRequest(record, schemaAndValue);
        assertEquals(CONNECTOR_1_NAME, restartRequest.connectorName());
        assertEquals(struct.getBoolean(INCLUDE_TASKS_FIELD_NAME), restartRequest.includeTasks());
        assertEquals(struct.getBoolean(ONLY_FAILED_FIELD_NAME), restartRequest.onlyFailed());
    }