prankstrisse commented on code in PR #11499: URL: https://github.com/apache/nifi/pull/11499#discussion_r4094806112
########## nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/StandardProtobufWriter.java: ########## @@ -0,0 +1,356 @@ +/* + * 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.OnEnabled; +import org.apache.nifi.components.AllowableValue; +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.AbstractControllerService; +import org.apache.nifi.controller.ConfigurationContext; +import org.apache.nifi.controller.ControllerServiceInitializationContext; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.reporting.InitializationException; +import org.apache.nifi.schema.access.SchemaNotFoundException; +import org.apache.nifi.schemaregistry.services.MessageIndexWriter; +import org.apache.nifi.schemaregistry.services.MessageName; +import org.apache.nifi.schemaregistry.services.MessageNameResolver; +import org.apache.nifi.schemaregistry.services.SchemaDefinition; +import org.apache.nifi.schemaregistry.services.SchemaReferenceWriter; +import org.apache.nifi.schemaregistry.services.SchemaRegistry; +import org.apache.nifi.schemaregistry.services.StandardMessageNameFactory; +import org.apache.nifi.schemaregistry.services.StandardSchemaDefinition; +import org.apache.nifi.serialization.RecordSetWriter; +import org.apache.nifi.serialization.RecordSetWriterFactory; +import org.apache.nifi.serialization.SchemaRegistryService; +import org.apache.nifi.serialization.SimpleRecordSchema; +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.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.util.ArrayList; +import java.util.HexFormat; +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_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.StandardProtobufWriter.MessageNameResolverStrategy.MESSAGE_NAME_PROPERTY; + +@Tags({"protobuf", "record", "writer", "serializer", "confluent"}) +@CapabilityDescription(""" + Serializes NiFi Records into Protocol Buffers binary format. \ + Supports inline schema text and schema registry lookup for determining the Proto schema. \ + When a Schema Reference Writer is configured, a Confluent wire-format header is written; when a \ + Message Index Writer is also configured, the Confluent message index array is written after the header. \ + The target Proto message name can be determined statically using the 'Message Name' property, \ + or dynamically using a Message Name Resolver service. + A single record is written per FlowFile, since concatenated Protocol Buffers messages cannot be delimited. \ + The 'google.protobuf.Any' well-known type is not expanded on write; a Record derived from an Any-typed message \ + is serialized as an ordinary nested message rather than being re-wrapped as an Any.""") +public class StandardProtobufWriter extends SchemaRegistryService implements RecordSetWriterFactory { + + public static final PropertyDescriptor MESSAGE_NAME_RESOLUTION_STRATEGY = new PropertyDescriptor.Builder() + .name("Message Name Resolution Strategy") + .description("Strategy for determining the Protocol Buffers message name for serialization") + .required(true) + .allowableValues(MESSAGE_NAME_PROPERTY, MessageNameResolverStrategy.MESSAGE_NAME_RESOLVER) + .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_RESOLUTION_STRATEGY, MESSAGE_NAME_PROPERTY) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor MESSAGE_NAME_RESOLVER = new PropertyDescriptor.Builder() + .name("Message Name Resolver") + .description("Service that dynamically resolves Protocol Buffer message names from FlowFile attributes. " + + "On the write side the resolver is invoked with an empty content stream, so only resolvers that derive the " + + "message name from attributes are supported; resolvers that read the message name from message content " + + "(such as the Confluent wire-format resolver used on the read side) are not applicable here.") + .required(true) + .identifiesControllerService(MessageNameResolver.class) + .dependsOn(MESSAGE_NAME_RESOLUTION_STRATEGY, MessageNameResolverStrategy.MESSAGE_NAME_RESOLVER) + .build(); + + public static final PropertyDescriptor SCHEMA_REFERENCE_WRITER = new PropertyDescriptor.Builder() + .name("Schema Reference Writer") + .description("Service used to write schema reference information, such as a Confluent wire-format header, before the serialized Protobuf content. " + + "When not configured, plain Protobuf content is written without any header.") Review Comment: thanks, done ########## nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/StandardProtobufWriter.java: ########## @@ -0,0 +1,356 @@ +/* + * 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.OnEnabled; +import org.apache.nifi.components.AllowableValue; +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.AbstractControllerService; +import org.apache.nifi.controller.ConfigurationContext; +import org.apache.nifi.controller.ControllerServiceInitializationContext; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.reporting.InitializationException; +import org.apache.nifi.schema.access.SchemaNotFoundException; +import org.apache.nifi.schemaregistry.services.MessageIndexWriter; +import org.apache.nifi.schemaregistry.services.MessageName; +import org.apache.nifi.schemaregistry.services.MessageNameResolver; +import org.apache.nifi.schemaregistry.services.SchemaDefinition; +import org.apache.nifi.schemaregistry.services.SchemaReferenceWriter; +import org.apache.nifi.schemaregistry.services.SchemaRegistry; +import org.apache.nifi.schemaregistry.services.StandardMessageNameFactory; +import org.apache.nifi.schemaregistry.services.StandardSchemaDefinition; +import org.apache.nifi.serialization.RecordSetWriter; +import org.apache.nifi.serialization.RecordSetWriterFactory; +import org.apache.nifi.serialization.SchemaRegistryService; +import org.apache.nifi.serialization.SimpleRecordSchema; +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.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.util.ArrayList; +import java.util.HexFormat; +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_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.StandardProtobufWriter.MessageNameResolverStrategy.MESSAGE_NAME_PROPERTY; + +@Tags({"protobuf", "record", "writer", "serializer", "confluent"}) +@CapabilityDescription(""" + Serializes NiFi Records into Protocol Buffers binary format. \ + Supports inline schema text and schema registry lookup for determining the Proto schema. \ + When a Schema Reference Writer is configured, a Confluent wire-format header is written; when a \ + Message Index Writer is also configured, the Confluent message index array is written after the header. \ + The target Proto message name can be determined statically using the 'Message Name' property, \ + or dynamically using a Message Name Resolver service. + A single record is written per FlowFile, since concatenated Protocol Buffers messages cannot be delimited. \ + The 'google.protobuf.Any' well-known type is not expanded on write; a Record derived from an Any-typed message \ + is serialized as an ordinary nested message rather than being re-wrapped as an Any.""") +public class StandardProtobufWriter extends SchemaRegistryService implements RecordSetWriterFactory { + + public static final PropertyDescriptor MESSAGE_NAME_RESOLUTION_STRATEGY = new PropertyDescriptor.Builder() + .name("Message Name Resolution Strategy") + .description("Strategy for determining the Protocol Buffers message name for serialization") + .required(true) + .allowableValues(MESSAGE_NAME_PROPERTY, MessageNameResolverStrategy.MESSAGE_NAME_RESOLVER) + .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_RESOLUTION_STRATEGY, MESSAGE_NAME_PROPERTY) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor MESSAGE_NAME_RESOLVER = new PropertyDescriptor.Builder() + .name("Message Name Resolver") + .description("Service that dynamically resolves Protocol Buffer message names from FlowFile attributes. " + + "On the write side the resolver is invoked with an empty content stream, so only resolvers that derive the " + + "message name from attributes are supported; resolvers that read the message name from message content " + + "(such as the Confluent wire-format resolver used on the read side) are not applicable here.") + .required(true) + .identifiesControllerService(MessageNameResolver.class) + .dependsOn(MESSAGE_NAME_RESOLUTION_STRATEGY, MessageNameResolverStrategy.MESSAGE_NAME_RESOLVER) + .build(); + + public static final PropertyDescriptor SCHEMA_REFERENCE_WRITER = new PropertyDescriptor.Builder() + .name("Schema Reference Writer") + .description("Service used to write schema reference information, such as a Confluent wire-format header, before the serialized Protobuf content. " + + "When not configured, plain Protobuf content is written without any header.") + .required(false) + .identifiesControllerService(SchemaReferenceWriter.class) + .build(); + + public static final PropertyDescriptor MESSAGE_INDEX_WRITER = new PropertyDescriptor.Builder() + .name("Message Index Writer") + .description("Service used to write the Confluent message index array identifying the target message within the schema, written after the Schema Reference Writer header. " + + "Applicable only when producing Confluent wire-format content.") Review Comment: thanks, done -- 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]
