class SchemaValueConverter {

    public static Map<String, Object> convertMap(Schema schema, Map<String, Object> value) {
        Map<String, Schema.Type> fieldToType = schema.getFields().stream()
                .collect(toMap(Schema.Field::getName, Schema.Field::getType));

        List<Schema.Field> requiredFields = schema.getFields().stream()
                .filter(field -> !field.isOptional()).collect(toList());

        List<String> fieldsNotFoundInSchema = value.keySet().stream()
                .filter(f -> !(fieldToType.containsKey(f))).collect(toList());

        List<Schema.Field> requiredFieldsNotFoundInValue = requiredFields.stream()
                .filter(f -> !(value.containsKey(f.getName()))).collect(toList());

        if (!fieldsNotFoundInSchema.isEmpty()) {
            throw SchemaValidationFailedException.fieldsNotFoundInSchema(fieldsNotFoundInSchema);
        }

        if (!requiredFieldsNotFoundInValue.isEmpty()) {
            throw SchemaValidationFailedException.requiredFieldsNotFoundInValue(
                    requiredFieldsNotFoundInValue.stream().map(Schema.Field::getName).collect(toList()));
        }

        Map<String, Object> result = new HashMap<>();
        value.forEach((k, v) -> {
            if (v == null) {
                throw SchemaValidationFailedException.nullValueForField(k);
            }
            result.put(k, convert(fieldToType.get(k), v));
        });
        return result;
    }

    private  SchemaValueConverter();

}