exceptionfactory commented on code in PR #7492: URL: https://github.com/apache/nifi/pull/7492#discussion_r1265891962
########## nifi-nar-bundles/nifi-aws-bundle/nifi-aws-schema-registry-service/src/main/java/org/apache/nifi/aws/schemaregistry/GlueSchemaRegistry.java: ########## @@ -0,0 +1,248 @@ +/* + * 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.aws.schemaregistry; + +import com.amazonaws.regions.Regions; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnEnabled; +import org.apache.nifi.aws.schemaregistry.client.CachingSchemaRegistryClient; +import org.apache.nifi.aws.schemaregistry.client.GlueSchemaRegistryClient; +import org.apache.nifi.aws.schemaregistry.client.SchemaRegistryClient; +import org.apache.nifi.components.AllowableValue; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.controller.AbstractControllerService; +import org.apache.nifi.controller.ConfigurationContext; +import org.apache.nifi.expression.ExpressionLanguageScope; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.processors.aws.credentials.provider.service.AWSCredentialsProviderService; +import org.apache.nifi.proxy.ProxyConfiguration; +import org.apache.nifi.proxy.ProxyConfigurationService; +import org.apache.nifi.proxy.ProxySpec; +import org.apache.nifi.schema.access.SchemaField; +import org.apache.nifi.schema.access.SchemaNotFoundException; +import org.apache.nifi.schemaregistry.services.SchemaRegistry; +import org.apache.nifi.serialization.record.RecordSchema; +import org.apache.nifi.serialization.record.SchemaIdentifier; +import org.apache.nifi.ssl.SSLContextService; +import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; +import software.amazon.awssdk.http.FileStoreTlsKeyManagersProvider; +import software.amazon.awssdk.http.SdkHttpClient; +import software.amazon.awssdk.http.TlsKeyManagersProvider; +import software.amazon.awssdk.http.apache.ApacheHttpClient; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.glue.GlueClient; +import software.amazon.awssdk.services.glue.GlueClientBuilder; + +import javax.net.ssl.TrustManager; +import java.io.IOException; +import java.net.Proxy; +import java.net.URI; +import java.nio.file.Path; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.EnumSet; +import java.util.List; +import java.util.OptionalInt; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +@Tags({"schema", "registry", "aws", "avro", "glue"}) +@CapabilityDescription("Provides a Schema Registry that interacts with the AWS Glue Schema Registry so that those Schemas that are stored in the Glue Schema " + + "Registry can be used in NiFi. When a Schema is looked up by name by this registry, it will find a Schema in the Glue Schema Registry with their names.") +public class GlueSchemaRegistry extends AbstractControllerService implements SchemaRegistry { Review Comment: Recommend prefixing the name with `Amazon`: ```suggestion public class AmazonGlueSchemaRegistry extends AbstractControllerService implements SchemaRegistry { ``` ########## nifi-nar-bundles/nifi-aws-bundle/nifi-aws-schema-registry-service/pom.xml: ########## @@ -0,0 +1,72 @@ +<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + + <parent> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-aws-bundle</artifactId> + <version>2.0.0-SNAPSHOT</version> + </parent> + + <artifactId>nifi-aws-schema-registry-service</artifactId> + <packaging>jar</packaging> + <dependencies> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-record</artifactId> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-avro-record-utils</artifactId> + <version>2.0.0-SNAPSHOT</version> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-api</artifactId> + </dependency> + <dependency> + <groupId>software.amazon.awssdk</groupId> + <artifactId>glue</artifactId> + <version>2.20.103</version> Review Comment: This version can be removed since the BOM dependency provides the version. ```suggestion ``` ########## nifi-nar-bundles/nifi-aws-bundle/pom.xml: ########## @@ -38,6 +38,8 @@ <module>nifi-aws-abstract-processors</module> <module>nifi-aws-parameter-value-providers</module> <module>nifi-aws-parameter-providers</module> + <module>nifi-aws-schema-registry-service</module> + <module>nifi-aws-schema-registry-service-nar</module> Review Comment: Have you evaluated the size of this new NAR? It might be better to bundle the new Schema Registry Service together with the existing AWS NAR to avoid duplicative dependencies in multiple NAR files. ########## nifi-nar-bundles/nifi-aws-bundle/nifi-aws-schema-registry-service/src/main/java/org/apache/nifi/aws/schemaregistry/client/GlueSchemaRegistryClient.java: ########## @@ -0,0 +1,109 @@ +/* + * 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.aws.schemaregistry.client; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.avro.Schema; +import org.apache.avro.SchemaParseException; +import org.apache.nifi.avro.AvroTypeUtil; +import org.apache.nifi.schema.access.SchemaNotFoundException; +import org.apache.nifi.serialization.record.RecordSchema; +import org.apache.nifi.serialization.record.SchemaIdentifier; +import software.amazon.awssdk.services.glue.GlueClient; +import software.amazon.awssdk.services.glue.model.GetSchemaVersionRequest; +import software.amazon.awssdk.services.glue.model.GetSchemaVersionResponse; +import software.amazon.awssdk.services.glue.model.SchemaId; +import software.amazon.awssdk.services.glue.model.SchemaVersionNumber; + +import java.io.IOException; + +public class GlueSchemaRegistryClient implements SchemaRegistryClient { + + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + private static final String NAMESPACE_FIELD_NAME = "namespace"; + + private final GlueClient client; + private final String registryName; + + public GlueSchemaRegistryClient(final GlueClient client, final String registryName) { + this.client = client; + this.registryName = registryName; + } + + @Override + public RecordSchema getSchema(String schemaName) throws IOException, SchemaNotFoundException { + final SchemaVersionNumber schemaVersionNumber = SchemaVersionNumber.builder() + .latestVersion(true) + .build(); + + final GetSchemaVersionResponse schemaVersionResponse = getSchemaVersionResponse(schemaName, schemaVersionNumber); + + return createRecordSchema(schemaVersionResponse); + } + + @Override + public RecordSchema getSchema(String schemaName, int version) throws IOException, SchemaNotFoundException { + final SchemaVersionNumber schemaVersionNumber = SchemaVersionNumber.builder() + .versionNumber((long) version) Review Comment: As `SchemaRegistryClient` is an interface internal to this implementation, it seems like the `version` parameter should be a `long` to avoid this hidden casting in the implementation. ########## nifi-nar-bundles/nifi-aws-bundle/nifi-aws-schema-registry-service/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService: ########## @@ -0,0 +1,16 @@ +# 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. + +org.apache.nifi.aws.schemaregistry.GlueSchemaRegistry Review Comment: ```suggestion org.apache.nifi.aws.schemaregistry.AmazonGlueSchemaRegistry ``` ########## nifi-nar-bundles/nifi-aws-bundle/nifi-aws-schema-registry-service/pom.xml: ########## @@ -0,0 +1,72 @@ +<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + + <parent> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-aws-bundle</artifactId> + <version>2.0.0-SNAPSHOT</version> + </parent> + + <artifactId>nifi-aws-schema-registry-service</artifactId> + <packaging>jar</packaging> + <dependencies> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-record</artifactId> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-avro-record-utils</artifactId> + <version>2.0.0-SNAPSHOT</version> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-api</artifactId> + </dependency> + <dependency> + <groupId>software.amazon.awssdk</groupId> + <artifactId>glue</artifactId> + <version>2.20.103</version> + </dependency> + <dependency> + <groupId>software.amazon.awssdk</groupId> + <artifactId>apache-client</artifactId> + <version>2.20.103</version> Review Comment: ```suggestion ``` ########## nifi-nar-bundles/nifi-aws-bundle/nifi-aws-schema-registry-service/src/main/java/org/apache/nifi/aws/schemaregistry/GlueSchemaRegistry.java: ########## @@ -0,0 +1,248 @@ +/* + * 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.aws.schemaregistry; + +import com.amazonaws.regions.Regions; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnEnabled; +import org.apache.nifi.aws.schemaregistry.client.CachingSchemaRegistryClient; +import org.apache.nifi.aws.schemaregistry.client.GlueSchemaRegistryClient; +import org.apache.nifi.aws.schemaregistry.client.SchemaRegistryClient; +import org.apache.nifi.components.AllowableValue; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.controller.AbstractControllerService; +import org.apache.nifi.controller.ConfigurationContext; +import org.apache.nifi.expression.ExpressionLanguageScope; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.processors.aws.credentials.provider.service.AWSCredentialsProviderService; +import org.apache.nifi.proxy.ProxyConfiguration; +import org.apache.nifi.proxy.ProxyConfigurationService; +import org.apache.nifi.proxy.ProxySpec; +import org.apache.nifi.schema.access.SchemaField; +import org.apache.nifi.schema.access.SchemaNotFoundException; +import org.apache.nifi.schemaregistry.services.SchemaRegistry; +import org.apache.nifi.serialization.record.RecordSchema; +import org.apache.nifi.serialization.record.SchemaIdentifier; +import org.apache.nifi.ssl.SSLContextService; +import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; +import software.amazon.awssdk.http.FileStoreTlsKeyManagersProvider; +import software.amazon.awssdk.http.SdkHttpClient; +import software.amazon.awssdk.http.TlsKeyManagersProvider; +import software.amazon.awssdk.http.apache.ApacheHttpClient; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.glue.GlueClient; +import software.amazon.awssdk.services.glue.GlueClientBuilder; + +import javax.net.ssl.TrustManager; +import java.io.IOException; +import java.net.Proxy; +import java.net.URI; +import java.nio.file.Path; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.EnumSet; +import java.util.List; +import java.util.OptionalInt; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +@Tags({"schema", "registry", "aws", "avro", "glue"}) +@CapabilityDescription("Provides a Schema Registry that interacts with the AWS Glue Schema Registry so that those Schemas that are stored in the Glue Schema " + + "Registry can be used in NiFi. When a Schema is looked up by name by this registry, it will find a Schema in the Glue Schema Registry with their names.") +public class GlueSchemaRegistry extends AbstractControllerService implements SchemaRegistry { + + private static final Set<SchemaField> schemaFields = EnumSet.of(SchemaField.SCHEMA_NAME, SchemaField.SCHEMA_TEXT, + SchemaField.SCHEMA_TEXT_FORMAT, SchemaField.SCHEMA_IDENTIFIER, SchemaField.SCHEMA_VERSION); + + static final PropertyDescriptor SCHEMA_REGISTRY_NAME = new PropertyDescriptor.Builder() + .name("schema-registry-name") + .displayName("Schema Registry Name") + .description("The name of the Schema Registry") + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .required(true) + .build(); + + public static final PropertyDescriptor REGION = new PropertyDescriptor.Builder() + .name("region") + .displayName("Region") + .description("The region of the cloud resources") + .required(true) + .allowableValues(getAvailableRegions()) + .defaultValue(createAllowableValue(Regions.DEFAULT_REGION).getValue()) + .build(); + + static final PropertyDescriptor CACHE_SIZE = new PropertyDescriptor.Builder() + .name("cache-size") + .displayName("Cache Size") + .description("Specifies how many Schemas should be cached from the Schema Registry") + .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR) + .defaultValue("1000") + .required(true) + .build(); + + static final PropertyDescriptor CACHE_EXPIRATION = new PropertyDescriptor.Builder() + .name("cache-expiration") + .displayName("Cache Expiration") + .description("Specifies how long a Schema that is cached should remain in the cache. Once this time period elapses, a " + + "cached version of a schema will no longer be used, and the service will have to communicate with the " + + "Schema Registry again in order to obtain the schema.") + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .defaultValue("1 hour") + .required(true) + .build(); + + static final PropertyDescriptor TIMEOUT = new PropertyDescriptor.Builder() + .name("timeout") Review Comment: Recommend a more specific name to match the display name: ```suggestion .name("communications-timeout") ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@nifi.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
