exceptionfactory commented on code in PR #10214:
URL: https://github.com/apache/nifi/pull/10214#discussion_r2299702118
##########
nifi-extension-bundles/nifi-standard-services/nifi-schema-registry-service-api/src/main/java/org/apache/nifi/schemaregistry/services/MessageNameResolver.java:
##########
@@ -32,7 +29,7 @@
* message processing and routing based on the resolved message type.
* </p>
*/
-public interface MessageNameResolver extends ControllerService {
+public interface MessageNameResolver {
Review Comment:
Extra space should be removed:
```suggestion
public interface MessageNameResolver {
```
##########
nifi-extension-bundles/nifi-standard-services/nifi-schema-registry-service-api/src/main/java/org/apache/nifi/schemaregistry/services/MessageNameResolverService.java:
##########
@@ -0,0 +1,25 @@
+/*
+ * 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;
+
+/**
+ * A service interface for resolving message names from schema definitions and
input streams.
+ */
+public interface MessageNameResolverService extends ControllerService,
MessageNameResolver {
Review Comment:
This refactoring seems unnecessary, the previous approach with
`MessageNameResolver` extending `ControllerService` is preferred.
##########
nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/ProtobufReader.java:
##########
@@ -54,7 +54,10 @@
import java.util.concurrent.atomic.AtomicReference;
@Tags({"protobuf", "record", "reader", "parser"})
-@CapabilityDescription("Parses a Protocol Buffers message from binary format.")
+@CapabilityDescription("Parses a Protocol Buffers message from binary format.
" +
+ "If the schema can be fetched from a schema registry or pasted inline
into property, StandardProtobufReader should be used instead. " +
+ "Note that this service has the limitation because of the fact that it
reads .proto schema from filesystem local to the NIFI node." +
+ "As a consequence, it requires proto files to be copied to every node
manually when this service is used in a multinode cluster. ")
Review Comment:
This description should use a multi-line string, but it could also use some
adjustments. Instead of providing this description, I recommend keeping the
original description, but adding the `SeeAlso` annotation.
##########
nifi-extension-bundles/nifi-standard-services/nifi-schema-registry-service-api/src/main/java/org/apache/nifi/schemaregistry/services/StandardMessageName.java:
##########
@@ -29,6 +29,18 @@ public StandardMessageName(final Optional<String> namespace,
final String name)
this.namespace = namespace;
}
+ public StandardMessageName(final String name) {
Review Comment:
This convenience constructor seems questionable, especially because it does
not have any method documentation. I recommend removing it and localizing the
parsing to avoid potential confusion around the single parameter.
##########
nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/StandardProtobufReader.java:
##########
@@ -0,0 +1,306 @@
+/*
+ * 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.services.protobuf;
+
+import com.squareup.wire.schema.Schema;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnDisabled;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.components.DescribedValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.PropertyValue;
+import org.apache.nifi.context.PropertyContext;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.schema.access.SchemaNotFoundException;
+import org.apache.nifi.schemaregistry.services.MessageName;
+import org.apache.nifi.schemaregistry.services.MessageNameResolver;
+import org.apache.nifi.schemaregistry.services.MessageNameResolverService;
+import org.apache.nifi.schemaregistry.services.SchemaDefinition;
+import org.apache.nifi.schemaregistry.services.SchemaReferenceReader;
+import org.apache.nifi.schemaregistry.services.SchemaRegistry;
+import org.apache.nifi.schemaregistry.services.StandardMessageName;
+import org.apache.nifi.schemaregistry.services.StandardSchemaDefinition;
+import org.apache.nifi.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.SchemaRegistryService;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.serialization.record.SchemaIdentifier;
+import org.apache.nifi.services.protobuf.schema.ProtoSchemaParser;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static
org.apache.nifi.expression.ExpressionLanguageScope.FLOWFILE_ATTRIBUTES;
+import static
org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_ACCESS_STRATEGY;
+import static
org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_BRANCH_NAME;
+import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_NAME;
+import static
org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_NAME_PROPERTY;
+import static
org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_REFERENCE_READER;
+import static
org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_REFERENCE_READER_PROPERTY;
+import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_REGISTRY;
+import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_TEXT;
+import static
org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_TEXT_PROPERTY;
+import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_VERSION;
+import static
org.apache.nifi.services.protobuf.StandardProtobufReader.MessageNameResolverStrategyName.MESSAGE_NAME_PROPERTY;
+import static
org.apache.nifi.services.protobuf.StandardProtobufReader.MessageNameResolverStrategyName.MESSAGE_NAME_RESOLVER_SERVICE;
+
+@Tags({"protobuf", "record", "reader", "parser"})
+@CapabilityDescription("""
+ Parses Protocol Buffers messages from binary format into NiFi Records. \
+ Supports multiple schema access strategies including inline schema text,
schema registry lookup, \
+ and schema reference readers.
+ Protobuf reader needs to know the Proto schema message name in order to
deserialize the binary payload correctly. \
+ The name of this message can be determined statically using 'Message Name'
property, \
+ or dynamically, using a Message Name Resolver service.""")
+
+public class StandardProtobufReader extends SchemaRegistryService implements
RecordReaderFactory {
+
+ public static final PropertyDescriptor MESSAGE_NAME_RESOLVER_STRATEGY =
new PropertyDescriptor.Builder()
+ .name("Message Name Resolver Strategy")
+ .description("Strategy for determining the Protocol Buffers message
name for processing")
+ .required(true)
+ .allowableValues(MESSAGE_NAME_PROPERTY, MESSAGE_NAME_RESOLVER_SERVICE)
+ .defaultValue(MESSAGE_NAME_PROPERTY)
+ .build();
+
+ public static final PropertyDescriptor MESSAGE_NAME = new
PropertyDescriptor.Builder()
+ .name("Message Name")
+ .description("Fully qualified name of the Protocol Buffers message
including its package (eg. mypackage.MyMessage).")
+ .required(true)
+ .expressionLanguageSupported(FLOWFILE_ATTRIBUTES)
+ .dependsOn(MESSAGE_NAME_RESOLVER_STRATEGY, MESSAGE_NAME_PROPERTY)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor
MESSAGE_NAME_RESOLVER_CONTROLLER_SERVICE = new PropertyDescriptor.Builder()
+ .name("Message Name Resolver Service")
+ .description("Controller service that dynamically resolves Protocol
Buffer message names from FlowFile content or attributes")
+ .required(true)
+ .identifiesControllerService(MessageNameResolverService.class)
+ .dependsOn(MESSAGE_NAME_RESOLVER_STRATEGY,
MESSAGE_NAME_RESOLVER_SERVICE)
+ .build();
+
+ private static final String PROTO_EXTENSION = ".proto";
+
+ private volatile ProtobufSchemaCompiler schemaCompiler;
+ private volatile MessageNameResolver messageNameResolver;
+ private volatile SchemaReferenceReader schemaReferenceReader;
+ private volatile SchemaRegistry schemaRegistry;
+ private volatile String schemaAccessStrategyValue;
+ private volatile PropertyValue schemaText;
+ private volatile PropertyValue schemaName;
+ private volatile PropertyValue schemaBranchName;
+ private volatile PropertyValue schemaVersion;
+
+
+ @OnEnabled
+ public void onEnabled(final ConfigurationContext context) {
+ super.storeSchemaAccessStrategy(context);
+ setupMessageNameResolver(context);
+ schemaAccessStrategyValue =
context.getProperty(SCHEMA_ACCESS_STRATEGY).getValue();
+ schemaReferenceReader =
context.getProperty(SCHEMA_REFERENCE_READER).asControllerService(SchemaReferenceReader.class);
+ schemaRegistry =
context.getProperty(SCHEMA_REGISTRY).asControllerService(SchemaRegistry.class);
+ schemaName = context.getProperty(SCHEMA_NAME);
+ schemaText = context.getProperty(SCHEMA_TEXT);
+ schemaBranchName = context.getProperty(SCHEMA_BRANCH_NAME);
+ schemaVersion = context.getProperty(SCHEMA_VERSION);
+ schemaCompiler = new ProtobufSchemaCompiler(getIdentifier(),
getLogger());
+
+ }
+
+ @OnDisabled
+ public void onDisabled(final ConfigurationContext context) {
+ schemaCompiler = null;
+ }
+
+ @Override
+ public RecordReader createRecordReader(final Map<String, String>
variables, final InputStream in, final long inputLength, final ComponentLog
logger) throws IOException, SchemaNotFoundException {
+ if (SCHEMA_TEXT_PROPERTY.getValue().equals(schemaAccessStrategyValue))
{
+ final SchemaDefinition schemaDefinition =
createSchemaDefinitionFromText(variables);
+ return createProtobufRecordReader(variables, in, schemaDefinition);
+ } else if
(SCHEMA_NAME_PROPERTY.getValue().equals(schemaAccessStrategyValue)) {
+ final SchemaDefinition schemaDefinition =
createSchemaDefinitionFromRegistry(variables);
+ return createProtobufRecordReader(variables, in, schemaDefinition);
+ } else if
(SCHEMA_REFERENCE_READER_PROPERTY.getValue().equals(schemaAccessStrategyValue))
{
+ final SchemaIdentifier schemaIdentifier =
schemaReferenceReader.getSchemaIdentifier(variables, in);
+ final SchemaDefinition schemaDefinition =
schemaRegistry.retrieveSchemaDefinition(schemaIdentifier);
+ logger.debug("Using message name for schema identifier: {}",
schemaDefinition.getIdentifier());
+ return createProtobufRecordReader(variables, in, schemaDefinition);
+ }
+
+ throw new SchemaNotFoundException("Unsupported schema access strategy:
" + schemaAccessStrategyValue);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = new
ArrayList<>(super.getSupportedPropertyDescriptors());
+ properties.add(MESSAGE_NAME_RESOLVER_STRATEGY);
+ properties.add(MESSAGE_NAME_RESOLVER_CONTROLLER_SERVICE);
+ properties.add(MESSAGE_NAME);
+ return properties;
+ }
+
+ @Override
+ protected PropertyDescriptor buildSchemaTextProperty() {
+ return new PropertyDescriptor.Builder()
+ .fromPropertyDescriptor(SCHEMA_TEXT)
+ .required(true)
+ .clearValidators()
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .defaultValue("${proto.schema}")
+ .description("The text of a Proto 3 formatted Schema")
+ .build();
+ }
+
+ private RecordReader createProtobufRecordReader(final Map<String, String>
variables, final InputStream in, final SchemaDefinition schemaDefinition)
throws IOException {
+ final Schema schema =
schemaCompiler.compileOrGetFromCache(schemaDefinition);
+ final ProtoSchemaParser schemaParser = new ProtoSchemaParser(schema);
+ final MessageName messageName =
messageNameResolver.getMessageName(variables, schemaDefinition, in);
+ final RecordSchema recordSchema =
schemaParser.createSchema(messageName.getFullyQualifiedName());
+ return new ProtobufRecordReader(schema,
messageName.getFullyQualifiedName(), in, recordSchema);
+ }
+
+
+ private void setupMessageNameResolver(final ConfigurationContext context) {
+ final MessageNameResolverStrategyName messageNameResolverStrategyName
=
context.getProperty(MESSAGE_NAME_RESOLVER_STRATEGY).asAllowableValue(MessageNameResolverStrategyName.class);
+ messageNameResolver = switch (messageNameResolverStrategyName) {
+ case MESSAGE_NAME_PROPERTY -> new
PropertyMessageNameResolver(context);
+ case MESSAGE_NAME_RESOLVER_SERVICE ->
context.getProperty(MESSAGE_NAME_RESOLVER_CONTROLLER_SERVICE).asControllerService(MessageNameResolverService.class);
+ };
+ }
+
+ private SchemaDefinition createSchemaDefinitionFromText(final Map<String,
String> variables) throws SchemaNotFoundException {
+ final String schemaTextString =
schemaText.evaluateAttributeExpressions(variables).getValue();
+ validateSchemaText(schemaTextString);
+
+ final String sha256hex = sha256Hex(schemaTextString);
+ final SchemaIdentifier schemaIdentifier = SchemaIdentifier.builder()
+ .name(sha256hex + PROTO_EXTENSION)
+ .build();
+
+ return new StandardSchemaDefinition(schemaIdentifier,
schemaTextString, SchemaDefinition.SchemaType.PROTOBUF);
+ }
+
+ private String sha256Hex(final String input) {
+ MessageDigest digest;
+ try {
+ digest = MessageDigest.getInstance("SHA-256");
+ } catch (NoSuchAlgorithmException e) {
+ throw new IllegalStateException(e);
+ }
+ byte[] hash = digest.digest(
+ input.getBytes(StandardCharsets.UTF_8));
+ final StringBuilder hexString = new StringBuilder();
+ for (byte b : hash) {
+ final String hex = Integer.toHexString(0xff & b);
+ if (hex.length() == 1)
+ hexString.append('0');
+ hexString.append(hex);
+ }
Review Comment:
This approach can be replaced
[HexFormat](https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/HexFormat.html)
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
