This is an automated email from the ASF dual-hosted git repository. exceptionfactory pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push: new 6bca79cb37 NIFI-11627 Added JsonSchemaRegistry for ValidateJson 6bca79cb37 is described below commit 6bca79cb3720395352bfa587449422e09bec0233 Author: dan-s1 <dsti...@gmail.com> AuthorDate: Fri Nov 10 17:36:42 2023 +0000 NIFI-11627 Added JsonSchemaRegistry for ValidateJson - Added nifi-json-schema-api to nifi-commons - Added StandardJsonSchemaRegistry implementation of JsonSchemaRegistry - Added strategy configuration properties to ValidateJson This closes #8005 Signed-off-by: David Handermann <exceptionfact...@apache.org> --- nifi-commons/nifi-json-schema-api/pom.xml | 33 ++++ .../org/apache/nifi/json/schema/JsonSchema.java | 39 ++++ .../org/apache/nifi/json/schema/SchemaVersion.java | 56 ++++++ nifi-commons/nifi-json-schema-shared/pom.xml | 37 ++++ .../schema/access/JsonSchemaRegistryComponent.java | 32 ++++ nifi-commons/pom.xml | 2 + .../nifi-registry-service/pom.xml | 20 ++ .../services/StandardJsonSchemaRegistry.java | 158 ++++++++++++++++ .../org.apache.nifi.controller.ControllerService | 3 +- .../services/TestStandardJsonSchemaRegistry.java | 123 ++++++++++++ .../nifi-standard-processors/pom.xml | 10 + .../nifi/processors/standard/ValidateJson.java | 210 +++++++++++++++------ .../nifi/processors/standard/TestValidateJson.java | 118 ++++++++++-- .../nifi-schema-registry-service-api/pom.xml | 5 + .../services/JsonSchemaRegistry.java | 37 ++++ 15 files changed, 815 insertions(+), 68 deletions(-) diff --git a/nifi-commons/nifi-json-schema-api/pom.xml b/nifi-commons/nifi-json-schema-api/pom.xml new file mode 100644 index 0000000000..eacabf87a7 --- /dev/null +++ b/nifi-commons/nifi-json-schema-api/pom.xml @@ -0,0 +1,33 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0" + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-commons</artifactId> + <version>2.0.0-SNAPSHOT</version> + </parent> + <artifactId>nifi-json-schema-api</artifactId> + <dependencies> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-api</artifactId> + <version>2.0.0-SNAPSHOT</version> + </dependency> + </dependencies> +</project> \ No newline at end of file diff --git a/nifi-commons/nifi-json-schema-api/src/main/java/org/apache/nifi/json/schema/JsonSchema.java b/nifi-commons/nifi-json-schema-api/src/main/java/org/apache/nifi/json/schema/JsonSchema.java new file mode 100644 index 0000000000..94b1fd18a3 --- /dev/null +++ b/nifi-commons/nifi-json-schema-api/src/main/java/org/apache/nifi/json/schema/JsonSchema.java @@ -0,0 +1,39 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.json.schema; + +import java.util.Objects; + +public class JsonSchema { + private final SchemaVersion schemaVersion; + private final String schemaText; + + public JsonSchema(SchemaVersion schemaVersion, String schemaText) { + Objects.requireNonNull(schemaVersion, "Schema version cannot be null"); + Objects.requireNonNull(schemaText, "The text of the schema cannot be null"); + this.schemaVersion = schemaVersion; + this.schemaText = schemaText; + } + + public SchemaVersion getSchemaVersion() { + return schemaVersion; + } + + public String getSchemaText() { + return schemaText; + } +} diff --git a/nifi-commons/nifi-json-schema-api/src/main/java/org/apache/nifi/json/schema/SchemaVersion.java b/nifi-commons/nifi-json-schema-api/src/main/java/org/apache/nifi/json/schema/SchemaVersion.java new file mode 100644 index 0000000000..2f386a63e9 --- /dev/null +++ b/nifi-commons/nifi-json-schema-api/src/main/java/org/apache/nifi/json/schema/SchemaVersion.java @@ -0,0 +1,56 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.json.schema; + +import org.apache.nifi.components.DescribedValue; + +public enum SchemaVersion implements DescribedValue { + DRAFT_4("Draft Version 4", "Draft 4", "https://json-schema.org/draft-04/schema"), + DRAFT_6("Draft Version 6", "Draft 6", "https://json-schema.org/draft-06/schema"), + DRAFT_7("Draft Version 7", "Draft 7", "https://json-schema.org/draft-07/schema"), + DRAFT_2019_09("Draft Version 2019-09", "Draft 2019-09", "https://json-schema.org/draft/2019-09/schema"), + DRAFT_2020_12("Draft Version 2020-12", "Draft 2020-12", "https://json-schema.org/draft/2020-12/schema"); + + private final String description; + private final String displayName; + private final String uri; + + SchemaVersion(String description, String displayName, String uri) { + this.description = description; + this.displayName = displayName; + this.uri = uri; + } + + @Override + public String getValue() { + return name(); + } + + @Override + public String getDisplayName() { + return displayName; + } + + @Override + public String getDescription() { + return description; + } + + public String getUri() { + return uri; + } +} diff --git a/nifi-commons/nifi-json-schema-shared/pom.xml b/nifi-commons/nifi-json-schema-shared/pom.xml new file mode 100644 index 0000000000..35149bd83f --- /dev/null +++ b/nifi-commons/nifi-json-schema-shared/pom.xml @@ -0,0 +1,37 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0" + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-commons</artifactId> + <version>2.0.0-SNAPSHOT</version> + </parent> + <artifactId>nifi-json-schema-shared</artifactId> + <dependencies> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-api</artifactId> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-json-schema-api</artifactId> + <version>2.0.0-SNAPSHOT</version> + </dependency> + </dependencies> +</project> \ No newline at end of file diff --git a/nifi-commons/nifi-json-schema-shared/src/main/java/org/apache/nifi/schema/access/JsonSchemaRegistryComponent.java b/nifi-commons/nifi-json-schema-shared/src/main/java/org/apache/nifi/schema/access/JsonSchemaRegistryComponent.java new file mode 100644 index 0000000000..0fc53bed30 --- /dev/null +++ b/nifi-commons/nifi-json-schema-shared/src/main/java/org/apache/nifi/schema/access/JsonSchemaRegistryComponent.java @@ -0,0 +1,32 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.schema.access; + +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.json.schema.SchemaVersion; + +public interface JsonSchemaRegistryComponent { + PropertyDescriptor SCHEMA_VERSION = new PropertyDescriptor + .Builder() + .name("JSON Schema Version") + .displayName("JSON Schema Version") + .description("The JSON schema specification") + .required(true) + .allowableValues(SchemaVersion.class) + .defaultValue(SchemaVersion.DRAFT_2020_12.getValue()) + .build(); +} diff --git a/nifi-commons/pom.xml b/nifi-commons/pom.xml index dddf9b7d86..034b39f123 100644 --- a/nifi-commons/pom.xml +++ b/nifi-commons/pom.xml @@ -30,6 +30,8 @@ <module>nifi-flowfile-packager</module> <module>nifi-flow-encryptor</module> <module>nifi-hl7-query-language</module> + <module>nifi-json-schema-api</module> + <module>nifi-json-schema-shared</module> <module>nifi-json-utils</module> <module>nifi-jetty-configuration</module> <module>nifi-kubernetes-client</module> diff --git a/nifi-nar-bundles/nifi-registry-bundle/nifi-registry-service/pom.xml b/nifi-nar-bundles/nifi-registry-bundle/nifi-registry-service/pom.xml index b879530b07..be9603ff9b 100644 --- a/nifi-nar-bundles/nifi-registry-bundle/nifi-registry-service/pom.xml +++ b/nifi-nar-bundles/nifi-registry-bundle/nifi-registry-service/pom.xml @@ -44,6 +44,11 @@ language governing permissions and limitations under the License. --> <groupId>org.apache.nifi</groupId> <artifactId>nifi-utils</artifactId> </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-json-schema-api</artifactId> + <version>2.0.0-SNAPSHOT</version> + </dependency> <dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-lang3</artifactId> @@ -52,5 +57,20 @@ language governing permissions and limitations under the License. --> <groupId>commons-io</groupId> <artifactId>commons-io</artifactId> </dependency> + <dependency> + <groupId>com.networknt</groupId> + <artifactId>json-schema-validator</artifactId> + <version>1.0.87</version> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-json-schema-shared</artifactId> + <version>2.0.0-SNAPSHOT</version> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-mock</artifactId> + <scope>test</scope> + </dependency> </dependencies> </project> diff --git a/nifi-nar-bundles/nifi-registry-bundle/nifi-registry-service/src/main/java/org/apache/nifi/schemaregistry/services/StandardJsonSchemaRegistry.java b/nifi-nar-bundles/nifi-registry-bundle/nifi-registry-service/src/main/java/org/apache/nifi/schemaregistry/services/StandardJsonSchemaRegistry.java new file mode 100644 index 0000000000..3e877ef8e1 --- /dev/null +++ b/nifi-nar-bundles/nifi-registry-bundle/nifi-registry-service/src/main/java/org/apache/nifi/schemaregistry/services/StandardJsonSchemaRegistry.java @@ -0,0 +1,158 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.schemaregistry.services; + +import com.networknt.schema.JsonSchemaFactory; +import com.networknt.schema.SpecVersion; +import org.apache.nifi.annotation.behavior.DynamicProperty; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.ValidationContext; +import org.apache.nifi.components.ValidationResult; +import org.apache.nifi.controller.AbstractControllerService; +import org.apache.nifi.expression.ExpressionLanguageScope; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.json.schema.JsonSchema; +import org.apache.nifi.schema.access.JsonSchemaRegistryComponent; +import org.apache.nifi.schema.access.SchemaNotFoundException; +import org.apache.nifi.json.schema.SchemaVersion; + +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.function.Function; +import java.util.stream.Collectors; + +import static org.apache.commons.lang3.StringUtils.isBlank; +import static org.apache.commons.lang3.StringUtils.isNotBlank; + +@Tags({"schema", "registry", "json"}) +@CapabilityDescription("Provides a service for registering and accessing JSON schemas. One can register a schema " + + "as a dynamic property where 'name' represents the schema name and 'value' represents the textual " + + "representation of the actual schema following the syntax and semantics of the JSON Schema format. " + + "Empty schemas and schemas only consisting of whitespace are not acceptable schemas." + + "The registry is heterogeneous registry as it can store schemas of different schema draft versions. " + + "By default the registry is configured to store schemas of Draft 2020-12. When a schema is added, the version " + + "which is currently is set, is what the schema is saved as.") +@DynamicProperty(name = "Schema Name", value = "Schema Content", + description = "Adds a named schema using the JSON string representation of a JSON schema", + expressionLanguageScope = ExpressionLanguageScope.NONE) +public class StandardJsonSchemaRegistry extends AbstractControllerService implements JsonSchemaRegistry, JsonSchemaRegistryComponent { + + private static final List<PropertyDescriptor> PROPERTY_DESCRIPTORS = Collections.singletonList(SCHEMA_VERSION); + + private final ConcurrentMap<String, JsonSchema> jsonSchemas; + private final ConcurrentMap<SchemaVersion, JsonSchemaFactory> schemaFactories; + private volatile SchemaVersion schemaVersion; + + public StandardJsonSchemaRegistry() { + jsonSchemas = new ConcurrentHashMap<>(); + schemaFactories = Arrays.stream(SchemaVersion.values()) + .collect(Collectors.toConcurrentMap(Function.identity(), + schemaDraftVersion -> JsonSchemaFactory.getInstance(SpecVersion.VersionFlag.fromId(schemaDraftVersion.getUri()).get()))); + schemaVersion = SchemaVersion.valueOf(SCHEMA_VERSION.getDefaultValue()); + } + + @Override + public void onPropertyModified(final PropertyDescriptor descriptor, final String oldValue, final String newValue) { + if (SCHEMA_VERSION.getName().equals(descriptor.getName()) && !newValue.equals(oldValue)) { + schemaVersion = SchemaVersion.valueOf(newValue); + } else if(descriptor.isDynamic() && isBlank(newValue)) { + jsonSchemas.remove(descriptor.getName()); + } else if (descriptor.isDynamic() && isNotBlank(newValue)) { + try { + final String schemaName = descriptor.getName(); + final JsonSchemaFactory jsonSchemaFactory = schemaFactories.get(schemaVersion); + jsonSchemaFactory.getSchema(newValue); + jsonSchemas.put(schemaName, new JsonSchema(schemaVersion, newValue)); + } catch (final Exception e) { + getLogger().debug("Exception thrown when changing value of schema name '{}' from '{}' to '{}'", + descriptor.getName(), oldValue, newValue, e); + } + } + } + + @Override + protected Collection<ValidationResult> customValidate(ValidationContext validationContext) { + final Set<ValidationResult> results = new HashSet<>(); + + final boolean noSchemasConfigured = validationContext.getProperties().keySet().stream() + .noneMatch(PropertyDescriptor::isDynamic); + if (noSchemasConfigured) { + results.add(new ValidationResult.Builder() + .subject("Supported Dynamic Property Descriptor") + .valid(false) + .explanation("There must be at least one JSON schema specified") + .build()); + } else { + // Iterate over dynamic properties, validating only newly added schemas, and adding results + schemaVersion = SchemaVersion.valueOf(validationContext.getProperty(JsonSchemaRegistryComponent.SCHEMA_VERSION).getValue()); + validationContext.getProperties().entrySet().stream() + .filter(entry -> entry.getKey().isDynamic() && !jsonSchemas.containsKey(entry.getKey().getName())) + .forEach(entry -> { + String subject = entry.getKey().getName(); + String input = entry.getValue(); + if (isNotBlank(input)) { + try { + final JsonSchemaFactory jsonSchemaFactory = schemaFactories.get(schemaVersion); + jsonSchemaFactory.getSchema(input); + } catch (Exception e) { + results.add(new ValidationResult.Builder() + .input(input) + .subject(subject) + .valid(false) + .explanation("Not a valid JSON Schema: " + e.getMessage()) + .build()); + } + } + }); + } + + return results; + } + + @Override + public JsonSchema retrieveSchema(final String schemaName) throws SchemaNotFoundException { + JsonSchema jsonSchema = jsonSchemas.get(schemaName); + if (jsonSchema == null) { + throw new SchemaNotFoundException("Unable to find schema with name '" + schemaName + "'"); + } + return jsonSchema; + } + + @Override + protected List<PropertyDescriptor> getSupportedPropertyDescriptors() { + return PROPERTY_DESCRIPTORS; + } + + @Override + protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(final String propertyDescriptorName) { + return new PropertyDescriptor.Builder() + .name(propertyDescriptorName) + .required(false) + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .dynamic(true) + .expressionLanguageSupported(ExpressionLanguageScope.NONE) + .build(); + } +} diff --git a/nifi-nar-bundles/nifi-registry-bundle/nifi-registry-service/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService b/nifi-nar-bundles/nifi-registry-bundle/nifi-registry-service/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService index a000cd743d..1cb051bdac 100644 --- a/nifi-nar-bundles/nifi-registry-bundle/nifi-registry-service/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService +++ b/nifi-nar-bundles/nifi-registry-bundle/nifi-registry-service/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService @@ -12,4 +12,5 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. -org.apache.nifi.schemaregistry.services.AvroSchemaRegistry \ No newline at end of file +org.apache.nifi.schemaregistry.services.AvroSchemaRegistry +org.apache.nifi.schemaregistry.services.StandardJsonSchemaRegistry \ No newline at end of file diff --git a/nifi-nar-bundles/nifi-registry-bundle/nifi-registry-service/src/test/java/org/apache/nifi/schemaregistry/services/TestStandardJsonSchemaRegistry.java b/nifi-nar-bundles/nifi-registry-bundle/nifi-registry-service/src/test/java/org/apache/nifi/schemaregistry/services/TestStandardJsonSchemaRegistry.java new file mode 100644 index 0000000000..87560ee01e --- /dev/null +++ b/nifi-nar-bundles/nifi-registry-bundle/nifi-registry-service/src/test/java/org/apache/nifi/schemaregistry/services/TestStandardJsonSchemaRegistry.java @@ -0,0 +1,123 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.schemaregistry.services; + +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.PropertyValue; +import org.apache.nifi.components.ValidationContext; +import org.apache.nifi.components.ValidationResult; +import org.apache.nifi.expression.ExpressionLanguageScope; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.reporting.InitializationException; +import org.apache.nifi.schema.access.JsonSchemaRegistryComponent; +import org.apache.nifi.schema.access.SchemaNotFoundException; +import org.apache.nifi.util.NoOpProcessor; +import org.apache.nifi.util.TestRunner; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import java.util.Collection; +import java.util.HashMap; +import java.util.Map; +import java.util.stream.Stream; +import org.apache.nifi.util.TestRunners; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.when; + +@ExtendWith(MockitoExtension.class) +class TestStandardJsonSchemaRegistry { + private static final String SCHEMA_NAME = "fooSchema"; + private static final PropertyDescriptor SUPPORTED_DYNAMIC_PROPERTY_DESCRIPTOR = new PropertyDescriptor.Builder() + .name(SCHEMA_NAME) + .required(false) + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .dynamic(true) + .expressionLanguageSupported(ExpressionLanguageScope.NONE) + .build(); + + @Mock + private ValidationContext validationContext; + @Mock + private PropertyValue propertyValue; + private Map<PropertyDescriptor, String> properties; + private StandardJsonSchemaRegistry delegate; + + @BeforeEach + void setUp() throws InitializationException { + properties = new HashMap<>(); + delegate = new StandardJsonSchemaRegistry(); + TestRunner runner = TestRunners.newTestRunner(NoOpProcessor.class); + runner.addControllerService("jsonSchemaRegistry", delegate); + } + + @Test + void testCustomValidateWithoutAnySchemaSpecified() { + when(validationContext.getProperties()).thenReturn(properties); + + final Collection<ValidationResult> results = delegate.customValidate(validationContext); + assertEquals(1, results.size()); + final ValidationResult result = results.iterator().next(); + assertFalse(result.isValid()); + assertTrue(result.getExplanation().contains("at least one JSON schema specified")); + } + + @ParameterizedTest(name = "{3}") + @MethodSource("dynamicProperties") + void testCustomValidateWithSchemaRegistrationFromDynamicProperties(PropertyDescriptor propertyDescriptor, String schema, int numValidationErrors) { + when(validationContext.getProperty(JsonSchemaRegistryComponent.SCHEMA_VERSION)).thenReturn(propertyValue); + when(propertyValue.getValue()).thenReturn(JsonSchemaRegistryComponent.SCHEMA_VERSION.getDefaultValue()); + when(validationContext.getProperties()).thenReturn(properties); + properties.put(propertyDescriptor, schema); + delegate.getSupportedPropertyDescriptors().forEach(prop -> properties.put(prop, prop.getDisplayName())); + + assertEquals(numValidationErrors, delegate.customValidate(validationContext).size()); + } + + @ParameterizedTest(name = "{3}") + @MethodSource("dynamicProperties") + void testSchemaRetrieval(PropertyDescriptor propertyDescriptor, String schema, int numValidationErrors) throws SchemaNotFoundException { + delegate.onPropertyModified(propertyDescriptor, null, schema); + boolean validSchema = numValidationErrors == 0; + + if(validSchema) { + assertDoesNotThrow(() -> delegate.retrieveSchema(SCHEMA_NAME)); + assertNotNull(delegate.retrieveSchema(SCHEMA_NAME)); + } else { + assertThrows(SchemaNotFoundException.class, () -> delegate.retrieveSchema(SCHEMA_NAME)); + } + } + + private static Stream<Arguments> dynamicProperties() { + return Stream.of( + Arguments.of(SUPPORTED_DYNAMIC_PROPERTY_DESCRIPTOR, "{}", 0, "empty object schema"), + Arguments.of(SUPPORTED_DYNAMIC_PROPERTY_DESCRIPTOR, "[]", 0, "empty array schema"), + Arguments.of(SUPPORTED_DYNAMIC_PROPERTY_DESCRIPTOR, "not a schema", 1, "non whitespace") + ); + } +} diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml index 5becb29639..7bf8071ead 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml @@ -499,6 +499,16 @@ <artifactId>mockwebserver</artifactId> <scope>test</scope> </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-json-schema-api</artifactId> + <version>2.0.0-SNAPSHOT</version> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-json-schema-shared</artifactId> + <version>2.0.0-SNAPSHOT</version> + </dependency> </dependencies> <build> <plugins> diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ValidateJson.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ValidateJson.java index 25b1499135..29a10456e9 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ValidateJson.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ValidateJson.java @@ -19,9 +19,8 @@ package org.apache.nifi.processors.standard; import com.fasterxml.jackson.core.JsonParser; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; -import com.networknt.schema.JsonSchema; import com.networknt.schema.JsonSchemaFactory; -import com.networknt.schema.SpecVersion.VersionFlag; +import com.networknt.schema.SpecVersion; import com.networknt.schema.ValidationMessage; import org.apache.nifi.annotation.behavior.InputRequirement; import org.apache.nifi.annotation.behavior.InputRequirement.Requirement; @@ -39,23 +38,35 @@ import org.apache.nifi.annotation.lifecycle.OnScheduled; import org.apache.nifi.components.DescribedValue; import org.apache.nifi.components.PropertyDescriptor; import org.apache.nifi.components.RequiredPermission; +import org.apache.nifi.components.ValidationContext; +import org.apache.nifi.components.ValidationResult; import org.apache.nifi.components.resource.ResourceCardinality; import org.apache.nifi.components.resource.ResourceType; +import org.apache.nifi.context.PropertyContext; +import org.apache.nifi.expression.ExpressionLanguageScope; import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.migration.PropertyConfiguration; import org.apache.nifi.processor.AbstractProcessor; import org.apache.nifi.processor.ProcessContext; import org.apache.nifi.processor.ProcessSession; import org.apache.nifi.processor.Relationship; import org.apache.nifi.processor.exception.ProcessException; import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.json.schema.JsonSchema; +import org.apache.nifi.schema.access.JsonSchemaRegistryComponent; +import org.apache.nifi.json.schema.SchemaVersion; +import org.apache.nifi.schemaregistry.services.JsonSchemaRegistry; import java.io.IOException; import java.io.InputStream; +import java.util.ArrayList; import java.util.Arrays; -import java.util.Collections; -import java.util.HashSet; +import java.util.Collection; import java.util.List; import java.util.Set; +import java.util.concurrent.ConcurrentMap; +import java.util.function.Function; +import java.util.stream.Collectors; @SideEffectFree @SupportsBatching @@ -79,23 +90,22 @@ import java.util.Set; } ) public class ValidateJson extends AbstractProcessor { - public enum SchemaVersion implements DescribedValue { - DRAFT_4("Draft Version 4", "Draft 4", VersionFlag.V4), - DRAFT_6("Draft Version 6", "Draft 6", VersionFlag.V6), - DRAFT_7("Draft Version 7", "Draft 7", VersionFlag.V7), - DRAFT_2019_09("Draft Version 2019-09", "Draft 2019-09", VersionFlag.V201909), - DRAFT_2020_12("Draft Version 2020-12", "Draft 2020-12", VersionFlag.V202012); + public enum JsonSchemaStrategy implements DescribedValue { + SCHEMA_NAME_PROPERTY(SCHEMA_NAME_PROPERTY_NAME + " Property", + "The name of the Schema to use is specified by the '" + SCHEMA_NAME_PROPERTY_NAME + + "' Property. The value of this property is used to lookup the Schema in the configured JSON Schema Registry Service."), + SCHEMA_CONTENT_PROPERTY(SCHEMA_CONTENT_PROPERTY_NAME + " Property", + "A URL or file path to the JSON schema or the actual JSON schema is specified by the '" + SCHEMA_CONTENT_PROPERTY_NAME + "' Property. " + + "No matter how the JSON schema is specified, it must be a valid JSON schema"); - private final String description; - private final String displayName; - private final VersionFlag versionFlag; - - SchemaVersion(String description, String displayName, VersionFlag versionFlag) { - this.description = description; + JsonSchemaStrategy(String displayName, String description) { this.displayName = displayName; - this.versionFlag = versionFlag; + this.description = description; } + private final String displayName; + private final String description; + @Override public String getValue() { return name(); @@ -110,37 +120,62 @@ public class ValidateJson extends AbstractProcessor { public String getDescription() { return description; } - - public VersionFlag getVersionFlag() { - return versionFlag; - } } - public static final String ERROR_ATTRIBUTE_KEY = "json.validation.errors"; + protected static final String ERROR_ATTRIBUTE_KEY = "json.validation.errors"; + private static final String SCHEMA_NAME_PROPERTY_NAME = "Schema Name"; + private static final String SCHEMA_CONTENT_PROPERTY_NAME = "JSON Schema"; + + public static final PropertyDescriptor SCHEMA_ACCESS_STRATEGY = new PropertyDescriptor.Builder() + .name("Schema Access Strategy") + .displayName("Schema Access Strategy") + .description("Specifies how to obtain the schema that is to be used for interpreting the data.") + .allowableValues(JsonSchemaStrategy.class) + .defaultValue(JsonSchemaStrategy.SCHEMA_CONTENT_PROPERTY.getValue()) + .required(true) + .build(); + + public static final PropertyDescriptor SCHEMA_NAME = new PropertyDescriptor.Builder() + .name(SCHEMA_NAME_PROPERTY_NAME) + .displayName(SCHEMA_NAME_PROPERTY_NAME) + .description("Specifies the name of the schema to lookup in the Schema Registry property") + .required(true) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) + .defaultValue("${schema.name}") + .dependsOn(SCHEMA_ACCESS_STRATEGY, JsonSchemaStrategy.SCHEMA_NAME_PROPERTY) + .build(); - public static final PropertyDescriptor SCHEMA_CONTENT = new PropertyDescriptor - .Builder().name("JSON Schema") - .displayName("JSON Schema") - .description("The content of a JSON Schema") + public static final PropertyDescriptor SCHEMA_REGISTRY = new PropertyDescriptor.Builder() + .name("JSON Schema Registry") + .displayName("JSON Schema Registry") + .description("Specifies the Controller Service to use for the JSON Schema Registry") + .identifiesControllerService(JsonSchemaRegistry.class) + .required(true) + .dependsOn(SCHEMA_ACCESS_STRATEGY, JsonSchemaStrategy.SCHEMA_NAME_PROPERTY) + .build(); + + public static final PropertyDescriptor SCHEMA_CONTENT = new PropertyDescriptor.Builder() + .name(SCHEMA_CONTENT_PROPERTY_NAME) + .displayName(SCHEMA_CONTENT_PROPERTY_NAME) + .description("A URL or file path to the JSON schema or the actual JSON schema content") .required(true) .identifiesExternalResource(ResourceCardinality.SINGLE, ResourceType.FILE, ResourceType.URL, ResourceType.TEXT) .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .dependsOn(SCHEMA_ACCESS_STRATEGY, JsonSchemaStrategy.SCHEMA_CONTENT_PROPERTY) .build(); - public static final PropertyDescriptor SCHEMA_VERSION = new PropertyDescriptor - .Builder().name("Schema Version") - .displayName("Schema Version") - .description("The JSON schema specification") - .required(true) - .allowableValues(SchemaVersion.class) - .defaultValue(SchemaVersion.DRAFT_2020_12.getValue()) - .build(); + public static final PropertyDescriptor SCHEMA_VERSION = new PropertyDescriptor.Builder() + .fromPropertyDescriptor(JsonSchemaRegistryComponent.SCHEMA_VERSION) + .dependsOn(SCHEMA_ACCESS_STRATEGY, JsonSchemaStrategy.SCHEMA_CONTENT_PROPERTY) + .build(); - public static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList( - Arrays.asList( - SCHEMA_CONTENT, - SCHEMA_VERSION - ) + private static final List<PropertyDescriptor> PROPERTIES = List.of( + SCHEMA_ACCESS_STRATEGY, + SCHEMA_NAME, + SCHEMA_REGISTRY, + SCHEMA_CONTENT, + SCHEMA_VERSION ); public static final Relationship REL_VALID = new Relationship.Builder() @@ -158,21 +193,28 @@ public class ValidateJson extends AbstractProcessor { .description("FlowFiles that cannot be read as JSON are routed to this relationship") .build(); - public static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>( - Arrays.asList( - REL_VALID, - REL_INVALID, - REL_FAILURE - )) + private static final Set<Relationship> RELATIONSHIPS = Set.of( + REL_VALID, + REL_INVALID, + REL_FAILURE ); - private static final ObjectMapper MAPPER; - static { - MAPPER = new ObjectMapper(); - MAPPER.configure(JsonParser.Feature.ALLOW_COMMENTS, true); - } + private static final ObjectMapper MAPPER = new ObjectMapper().configure(JsonParser.Feature.ALLOW_COMMENTS, true); - private JsonSchema schema; + private final ConcurrentMap<SchemaVersion, JsonSchemaFactory> schemaFactories = Arrays.stream(SchemaVersion.values()) + .collect( + Collectors.toConcurrentMap( + Function.identity(), + schemaDraftVersion -> JsonSchemaFactory.getInstance(SpecVersion.VersionFlag.fromId(schemaDraftVersion.getUri()).get()) + ) + ); + private volatile com.networknt.schema.JsonSchema schema; + private volatile JsonSchemaRegistry jsonSchemaRegistry; + + @Override + public void migrateProperties(final PropertyConfiguration config) { + config.renameProperty("Schema Version", SCHEMA_VERSION.getName()); + } @Override public Set<Relationship> getRelationships() { @@ -184,12 +226,38 @@ public class ValidateJson extends AbstractProcessor { return PROPERTIES; } + @Override + protected Collection<ValidationResult> customValidate(ValidationContext validationContext) { + Collection<ValidationResult> validationResults = new ArrayList<>(); + final String schemaAccessStrategy = getSchemaAccessStrategy(validationContext); + + if (isNameStrategy(validationContext) && !validationContext.getProperty(SCHEMA_REGISTRY).isSet()) { + validationResults.add(new ValidationResult.Builder() + .subject(SCHEMA_REGISTRY.getDisplayName()) + .explanation(getPropertyValidateMessage(schemaAccessStrategy, SCHEMA_REGISTRY)) + .valid(false) + .build()); + } else if (isContentStrategy(validationContext) && !validationContext.getProperty(SCHEMA_CONTENT).isSet()) { + validationResults.add(new ValidationResult.Builder() + .subject(SCHEMA_CONTENT.getDisplayName()) + .explanation(getPropertyValidateMessage(schemaAccessStrategy, SCHEMA_CONTENT)) + .valid(false) + .build()); + } + + return validationResults; + } + @OnScheduled public void onScheduled(final ProcessContext context) throws IOException { - try (final InputStream inputStream = context.getProperty(SCHEMA_CONTENT).asResource().read()) { - final SchemaVersion schemaVersion = SchemaVersion.valueOf(context.getProperty(SCHEMA_VERSION).getValue()); - final JsonSchemaFactory factory = JsonSchemaFactory.getInstance(schemaVersion.getVersionFlag()); - schema = factory.getSchema(inputStream); + if (isNameStrategy(context)) { + jsonSchemaRegistry = context.getProperty(SCHEMA_REGISTRY).asControllerService(JsonSchemaRegistry.class); + } else if (isContentStrategy(context)) { + try (final InputStream inputStream = context.getProperty(SCHEMA_CONTENT).asResource().read()) { + final SchemaVersion schemaVersion = SchemaVersion.valueOf(context.getProperty(SCHEMA_VERSION).getValue()); + final JsonSchemaFactory factory = schemaFactories.get(schemaVersion); + schema = factory.getSchema(inputStream); + } } } @@ -200,6 +268,20 @@ public class ValidateJson extends AbstractProcessor { return; } + if (isNameStrategy(context)) { + try { + final String schemaName = context.getProperty(SCHEMA_NAME).evaluateAttributeExpressions(flowFile).getValue(); + final JsonSchema jsonSchema = jsonSchemaRegistry.retrieveSchema(schemaName); + final JsonSchemaFactory factory = schemaFactories.get(jsonSchema.getSchemaVersion()); + schema = factory.getSchema(jsonSchema.getSchemaText()); + } catch (Exception e) { + getLogger().error("Could not retrieve JSON schema for {}", flowFile, e); + session.getProvenanceReporter().route(flowFile, REL_FAILURE); + session.transfer(flowFile, REL_FAILURE); + return; + } + } + try (final InputStream in = session.read(flowFile)) { final JsonNode node = MAPPER.readTree(in); final Set<ValidationMessage> errors = schema.validate(node); @@ -221,4 +303,22 @@ public class ValidateJson extends AbstractProcessor { session.transfer(flowFile, REL_FAILURE); } } + + private String getPropertyValidateMessage(String schemaAccessStrategy, PropertyDescriptor property) { + return "The '" + schemaAccessStrategy + "' Schema Access Strategy requires that the " + property.getDisplayName() + " property be set."; + } + + private boolean isNameStrategy(PropertyContext context) { + final String schemaAccessStrategy = getSchemaAccessStrategy(context); + return JsonSchemaStrategy.SCHEMA_NAME_PROPERTY.getValue().equals(schemaAccessStrategy); + } + + private String getSchemaAccessStrategy(PropertyContext context) { + return context.getProperty(SCHEMA_ACCESS_STRATEGY).getValue(); + } + + private boolean isContentStrategy(PropertyContext context) { + final String schemaAccessStrategy = getSchemaAccessStrategy(context); + return JsonSchemaStrategy.SCHEMA_CONTENT_PROPERTY.getValue().equals(schemaAccessStrategy); + } } diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestValidateJson.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestValidateJson.java index 14688ce706..c7b44177ef 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestValidateJson.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestValidateJson.java @@ -17,12 +17,22 @@ package org.apache.nifi.processors.standard; import org.apache.commons.io.IOUtils; +import org.apache.nifi.controller.AbstractControllerService; import org.apache.nifi.processor.Relationship; import org.apache.nifi.provenance.ProvenanceEventType; +import org.apache.nifi.reporting.InitializationException; +import org.apache.nifi.json.schema.JsonSchema; +import org.apache.nifi.schema.access.JsonSchemaRegistryComponent; +import org.apache.nifi.schema.access.SchemaNotFoundException; +import org.apache.nifi.json.schema.SchemaVersion; +import org.apache.nifi.schemaregistry.services.JsonSchemaRegistry; import org.apache.nifi.util.TestRunner; import org.apache.nifi.util.TestRunners; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; import org.opentest4j.AssertionFailedError; import java.io.IOException; @@ -30,18 +40,22 @@ import java.io.InputStream; import java.io.UncheckedIOException; import java.net.URL; import java.nio.charset.StandardCharsets; +import java.util.HashMap; import java.util.Map; import java.util.Objects; +import java.util.stream.Stream; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; class TestValidateJson { private static final String JSON = getFileContent("simple-example.json"); + private static final String SIMPLE_SCHEMA = getFileContent("schema-simple-example.json"); private static final String NON_JSON = "Not JSON"; - private static final String SCHEMA_VERSION = ValidateJson.SchemaVersion.DRAFT_7.getValue(); + private static final String SCHEMA_VERSION = SchemaVersion.DRAFT_7.getValue(); private TestRunner runner; @BeforeEach @@ -49,11 +63,39 @@ class TestValidateJson { runner = TestRunners.newTestRunner(ValidateJson.class); } + @ParameterizedTest(name = "{2}") + @MethodSource("customValidateArgs") + void testCustomValidateMissingProperty(String strategy, String errMsg) { + runner.setProperty(ValidateJson.SCHEMA_ACCESS_STRATEGY, strategy); + runner.enqueue(JSON); + + AssertionFailedError e = assertThrows(AssertionFailedError.class, () -> runner.run()); + assertTrue(e.getMessage().contains(errMsg)); + } + @Test void testPassSchema() { final String schemaPath = getFilePath("schema-simple-example.json"); runner.setProperty(ValidateJson.SCHEMA_CONTENT, schemaPath); - runner.setProperty(ValidateJson.SCHEMA_VERSION, SCHEMA_VERSION); + runner.setProperty(JsonSchemaRegistryComponent.SCHEMA_VERSION, SCHEMA_VERSION); + + runner.enqueue(JSON); + + runner.run(); + + runner.assertTransferCount(ValidateJson.REL_FAILURE, 0); + runner.assertTransferCount(ValidateJson.REL_INVALID, 0); + runner.assertTransferCount(ValidateJson.REL_VALID, 1); + + assertValidationErrors(ValidateJson.REL_VALID, false); + assertEquals(1, runner.getProvenanceEvents().size()); + assertEquals(ProvenanceEventType.ROUTE, runner.getProvenanceEvents().get(0).getEventType()); + } + + @Test + void testNoSchemaVersionSpecified() { + final String schemaPath = getFilePath("schema-simple-example.json"); + runner.setProperty(ValidateJson.SCHEMA_CONTENT, schemaPath); runner.enqueue(JSON); @@ -71,7 +113,7 @@ class TestValidateJson { @Test void testEmptySchema() { runner.setProperty(ValidateJson.SCHEMA_CONTENT, "{}"); - runner.setProperty(ValidateJson.SCHEMA_VERSION, SCHEMA_VERSION); + runner.setProperty(JsonSchemaRegistryComponent.SCHEMA_VERSION, SCHEMA_VERSION); runner.enqueue(JSON); runner.run(); @@ -89,7 +131,7 @@ class TestValidateJson { void testAllUnknownKeywordsSchema() { runner.setProperty(ValidateJson.SCHEMA_CONTENT, "{\"fruit\": \"Apple\", \"size\": \"Large\", \"color\": \"Red\"}"); - runner.setProperty(ValidateJson.SCHEMA_VERSION, SCHEMA_VERSION); + runner.setProperty(JsonSchemaRegistryComponent.SCHEMA_VERSION, SCHEMA_VERSION); runner.enqueue(JSON); runner.run(); @@ -107,7 +149,7 @@ class TestValidateJson { void testPatternSchemaCheck() { final String schemaPath = getFilePath("schema-simple-example-unmatched-pattern.json"); runner.setProperty(ValidateJson.SCHEMA_CONTENT, schemaPath); - runner.setProperty(ValidateJson.SCHEMA_VERSION, SCHEMA_VERSION); + runner.setProperty(JsonSchemaRegistryComponent.SCHEMA_VERSION, SCHEMA_VERSION); runner.enqueue(JSON); runner.run(); @@ -125,7 +167,7 @@ class TestValidateJson { void testMissingRequiredValue() { final String schema = getFileContent("schema-simple-example-missing-required.json"); runner.setProperty(ValidateJson.SCHEMA_CONTENT, schema); - runner.setProperty(ValidateJson.SCHEMA_VERSION, SCHEMA_VERSION); + runner.setProperty(JsonSchemaRegistryComponent.SCHEMA_VERSION, SCHEMA_VERSION); runner.enqueue(JSON); runner.run(); @@ -141,9 +183,8 @@ class TestValidateJson { @Test void testInvalidJson() { - final String schema = getFileContent("schema-simple-example.json"); - runner.setProperty(ValidateJson.SCHEMA_CONTENT, schema); - runner.setProperty(ValidateJson.SCHEMA_VERSION, SCHEMA_VERSION); + runner.setProperty(ValidateJson.SCHEMA_CONTENT, SIMPLE_SCHEMA); + runner.setProperty(JsonSchemaRegistryComponent.SCHEMA_VERSION, SCHEMA_VERSION); runner.enqueue(NON_JSON); runner.run(); @@ -160,7 +201,7 @@ class TestValidateJson { @Test void testNonExistingSchema() { runner.setProperty(ValidateJson.SCHEMA_CONTENT, "not-found.json"); - runner.setProperty(ValidateJson.SCHEMA_VERSION, SCHEMA_VERSION); + runner.setProperty(JsonSchemaRegistryComponent.SCHEMA_VERSION, SCHEMA_VERSION); runner.enqueue(JSON); assertThrows(AssertionFailedError.class, () -> runner.run()); @@ -169,7 +210,7 @@ class TestValidateJson { @Test void testBadSchema() { runner.setProperty(ValidateJson.SCHEMA_CONTENT, NON_JSON); - runner.setProperty(ValidateJson.SCHEMA_VERSION, SCHEMA_VERSION); + runner.setProperty(JsonSchemaRegistryComponent.SCHEMA_VERSION, SCHEMA_VERSION); runner.enqueue(JSON); assertThrows(AssertionFailedError.class, () -> runner.run()); @@ -179,7 +220,7 @@ class TestValidateJson { void testJsonWithComments() { final String schemaPath = getFilePath("schema-simple-example.json"); runner.setProperty(ValidateJson.SCHEMA_CONTENT, schemaPath); - runner.setProperty(ValidateJson.SCHEMA_VERSION, SCHEMA_VERSION); + runner.setProperty(JsonSchemaRegistryComponent.SCHEMA_VERSION, SCHEMA_VERSION); runner.enqueue(getFileContent("simple-example-with-comments.json")); @@ -191,6 +232,28 @@ class TestValidateJson { assertValidationErrors(ValidateJson.REL_VALID, false); } + + @Test + void testSchemaRetrievalFromRegistry() throws InitializationException { + final String registryIdentifier = "registry"; + final String schemaName = "someSchema"; + final JsonSchemaRegistry validJsonSchemaRegistry = new SampleJsonSchemaRegistry(registryIdentifier, schemaName); + runner.addControllerService(registryIdentifier, validJsonSchemaRegistry); + runner.enableControllerService(validJsonSchemaRegistry); + runner.assertValid(validJsonSchemaRegistry); + runner.setProperty(ValidateJson.SCHEMA_ACCESS_STRATEGY, ValidateJson.JsonSchemaStrategy.SCHEMA_NAME_PROPERTY.getValue()); + runner.setProperty(ValidateJson.SCHEMA_REGISTRY, registryIdentifier); + + Map<String, String> attributes = new HashMap<>(); + attributes.put("schema.name", schemaName); + runner.enqueue(JSON, attributes); + runner.run(); + + runner.assertTransferCount(ValidateJson.REL_FAILURE, 0); + runner.assertTransferCount(ValidateJson.REL_INVALID, 0); + runner.assertTransferCount(ValidateJson.REL_VALID, 1); + } + private void assertValidationErrors(Relationship relationship, boolean expected) { final Map<String, String> attributes = runner.getFlowFilesForRelationship(relationship).get(0).getAttributes(); @@ -202,6 +265,13 @@ class TestValidateJson { } } + private static Stream<Arguments> customValidateArgs() { + return Stream.of( + Arguments.of(ValidateJson.JsonSchemaStrategy.SCHEMA_NAME_PROPERTY.getValue(), "requires that the JSON Schema Registry property be set", "no registry set"), + Arguments.of(ValidateJson.JsonSchemaStrategy.SCHEMA_CONTENT_PROPERTY.getValue(), "requires that the JSON Schema property be set", "no content specified") + ); + } + private static String getFilePath(final String filename) { final String path = getRelativeResourcePath(filename); final URL url = Objects.requireNonNull(TestValidateJson.class.getResource(path), "Resource not found"); @@ -221,4 +291,28 @@ class TestValidateJson { private static String getRelativeResourcePath(final String filename) { return String.format("/%s/%s", TestValidateJson.class.getSimpleName(), filename); } + + private static class SampleJsonSchemaRegistry extends AbstractControllerService implements JsonSchemaRegistry { + private final String identifier; + private final String schemaName; + + public SampleJsonSchemaRegistry(String identifier, String schemaName) { + this.identifier = identifier; + this.schemaName = schemaName; + } + + @Override + public String getIdentifier() { + return identifier; + } + + @Override + public JsonSchema retrieveSchema(String schemaName) throws SchemaNotFoundException { + if (this.schemaName.equals(schemaName)) { + return new JsonSchema(SchemaVersion.DRAFT_2020_12, "{}"); + } else { + throw new SchemaNotFoundException(""); + } + } + } } diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-schema-registry-service-api/pom.xml b/nifi-nar-bundles/nifi-standard-services/nifi-schema-registry-service-api/pom.xml index d5f5032580..8b5b204eaa 100644 --- a/nifi-nar-bundles/nifi-standard-services/nifi-schema-registry-service-api/pom.xml +++ b/nifi-nar-bundles/nifi-standard-services/nifi-schema-registry-service-api/pom.xml @@ -31,5 +31,10 @@ <groupId>org.apache.nifi</groupId> <artifactId>nifi-record</artifactId> </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-json-schema-api</artifactId> + <version>2.0.0-SNAPSHOT</version> + </dependency> </dependencies> </project> diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-schema-registry-service-api/src/main/java/org/apache/nifi/schemaregistry/services/JsonSchemaRegistry.java b/nifi-nar-bundles/nifi-standard-services/nifi-schema-registry-service-api/src/main/java/org/apache/nifi/schemaregistry/services/JsonSchemaRegistry.java new file mode 100644 index 0000000000..364546fd10 --- /dev/null +++ b/nifi-nar-bundles/nifi-standard-services/nifi-schema-registry-service-api/src/main/java/org/apache/nifi/schemaregistry/services/JsonSchemaRegistry.java @@ -0,0 +1,37 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.schemaregistry.services; + +import org.apache.nifi.controller.ControllerService; +import org.apache.nifi.schema.access.SchemaNotFoundException; +import org.apache.nifi.json.schema.JsonSchema; +import java.io.IOException; + +/** + * Represents {@link ControllerService} strategy to expose internal and/or + * integrate with external Schema Registry + */ +public interface JsonSchemaRegistry extends ControllerService { + /** + * Retrieves the schema based on the provided schema name + * @param schemaName The name of the schema + * @return the schema for the given descriptor + * @throws IOException if unable to communicate with the backing store + * @throws SchemaNotFoundException if unable to find the schema based on the given descriptor + */ + JsonSchema retrieveSchema(String schemaName) throws IOException, SchemaNotFoundException; +}