tpalfy commented on code in PR #11499:
URL: https://github.com/apache/nifi/pull/11499#discussion_r4072913652
##########
nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/ProtobufSchemaCompiler.java:
##########
@@ -182,23 +183,40 @@ private void safeDeleteDirectory(final Path directory) {
}
/**
- * Writes a schema definition to the temporary directory structure.
- * If package name is present, creates the appropriate directory structure.
+ * Writes a schema definition to the temporary directory structure using
the name of its identifier.
*
* @param tempDir the temporary directory root
* @param schemaDefinition the schema definition to write
* @throws IOException if unable to create directories or write files
*/
private void writeSchemaToTempDirectory(final Path tempDir, final
SchemaDefinition schemaDefinition) throws IOException {
logger.debug("Writing schema definition to temporary directory.
Identifier: {}", schemaDefinition.getIdentifier());
+ writeSchemaFile(tempDir, generateSchemaFileName(schemaDefinition),
schemaDefinition.getText());
+ }
+
+ /**
+ * Writes schema text to a path relative to the temporary directory,
creating any parent directories.
+ * Import paths may contain directories, such as {@code
airlines/ph/cdm/shared.proto}, so the enclosing
+ * directory structure has to exist before the file is written.
Review Comment:
```suggestion
* Import paths may contain directories, such as {@code
airlines/ph/cdm/shared.proto}, so this method
* creates the enclosing directory structure before writing the file.
```
##########
nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/ProtobufSchemaValidator.java:
##########
@@ -17,47 +17,45 @@
package org.apache.nifi.services.protobuf;
import org.apache.nifi.schemaregistry.services.SchemaDefinition;
-import org.apache.nifi.serialization.record.SchemaIdentifier;
+
+import java.util.Map;
/**
- * Validates Protocol Buffer SchemaDefinition objects and schema identifiers.
+ * Validates the schema references of Protocol Buffer SchemaDefinition objects.
*/
final class ProtobufSchemaValidator {
+ private static final String PROTO_EXTENSION = ".proto";
+
private ProtobufSchemaValidator() {
}
/**
- * Validates that all SchemaDefinition identifiers end with .proto
extension.
- * Performs recursive validation on all referenced schemas.
+ * Validates that every schema reference, at any depth, is keyed by a path
ending in the .proto extension.
+ * <p>
+ * A reference is keyed by the path used in the import statement of the
referencing schema, for example
+ * {@code airlines/ph/cdm/shared.proto}, and the referenced schema is
written to exactly that path so the import
+ * resolves. The extension is required because the compiler only discovers
files named {@code *.proto}; without it
+ * the schema would fail to compile later with an unresolved import that
does not indicate the cause.
+ * <p>
+ * The identifier of a referenced schema is deliberately not validated. It
carries the subject the schema is
+ * registered under, which is unrelated to the import path and
legitimately has no .proto suffix. Under the
+ * Confluent RecordNameStrategy, for instance, a subject is a fully
qualified record name.
Review Comment:
I would avoid referencing Confluent-specific details.
```suggestion
* The identifier of a referenced schema is deliberately not validated. It
identifies the schema
* in its source registry and is unrelated to the Protobuf import path, so it
is not required to
* have a .proto suffix.
```
##########
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.")
+ .required(false)
+ .identifiesControllerService(MessageIndexWriter.class)
+ .build();
+
+ private static final PropertyDescriptor PROTOBUF_SCHEMA_TEXT = 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 static final String PROTO_EXTENSION = ".proto";
+
+ private static final InputStream EMPTY_INPUT_STREAM = new
ByteArrayInputStream(new byte[0]);
+
+ private volatile ProtobufSchemaCompiler schemaCompiler;
+ private volatile MessageNameResolver messageNameResolver;
+ private volatile SchemaReferenceWriter schemaReferenceWriter;
+ private volatile MessageIndexWriter messageIndexWriter;
+ 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();
+ schemaRegistry =
context.getProperty(SCHEMA_REGISTRY).asControllerService(SchemaRegistry.class);
+ schemaReferenceWriter =
context.getProperty(SCHEMA_REFERENCE_WRITER).asControllerService(SchemaReferenceWriter.class);
+ messageIndexWriter =
context.getProperty(MESSAGE_INDEX_WRITER).asControllerService(MessageIndexWriter.class);
+ schemaName = context.getProperty(SCHEMA_NAME);
+ schemaText = context.getProperty(SCHEMA_TEXT);
+ schemaBranchName = context.getProperty(SCHEMA_BRANCH_NAME);
+ schemaVersion = context.getProperty(SCHEMA_VERSION);
+ }
+
+ @Override
+ protected void init(final ControllerServiceInitializationContext config)
throws InitializationException {
+ super.init(config);
+ schemaCompiler = new ProtobufSchemaCompiler(getIdentifier(),
getLogger());
+ }
+
+ @Override
+ public RecordSchema getSchema(final Map<String, String> variables, final
RecordSchema readSchema) throws SchemaNotFoundException, IOException {
+ return createWriteContext(variables).recordSchema();
+ }
+
+ @Override
+ public RecordSetWriter createWriter(final ComponentLog logger, final
RecordSchema schema, final OutputStream out, final Map<String, String>
variables) throws SchemaNotFoundException, IOException {
+ final ProtobufWriteContext context = createWriteContext(variables);
+ return new WriteProtobufResultWithExternalSchema(context.schema(),
context.messageName(), context.recordSchema(),
+ context.schemaDefinition(), schemaReferenceWriter,
messageIndexWriter, variables, out);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = new
ArrayList<>(super.getSupportedPropertyDescriptors());
+ // The Schema Reference Reader is a read-side concern: a writer
determines its schema from the configured
+ // access strategy and writes references through the Schema Reference
Writer instead.
+ properties.removeIf(property ->
SCHEMA_REFERENCE_READER.getName().equals(property.getName()));
+ properties.add(MESSAGE_NAME_RESOLUTION_STRATEGY);
+ properties.add(MESSAGE_NAME_RESOLVER);
+ properties.add(MESSAGE_NAME);
+ properties.add(SCHEMA_REFERENCE_WRITER);
+ properties.add(MESSAGE_INDEX_WRITER);
+ return properties;
+ }
+
+ @Override
+ protected List<AllowableValue> getSchemaAccessStrategyValues() {
+ // Only the strategies that createSchemaDefinition supports are
offered; the inherited list also contains
+ // the Schema Reference Reader strategy, which cannot be used to
obtain a schema for writing.
Review Comment:
Unnecessary comments.
```suggestion
```
##########
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.")
+ .required(false)
+ .identifiesControllerService(MessageIndexWriter.class)
+ .build();
+
+ private static final PropertyDescriptor PROTOBUF_SCHEMA_TEXT = 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 static final String PROTO_EXTENSION = ".proto";
+
+ private static final InputStream EMPTY_INPUT_STREAM = new
ByteArrayInputStream(new byte[0]);
+
+ private volatile ProtobufSchemaCompiler schemaCompiler;
+ private volatile MessageNameResolver messageNameResolver;
+ private volatile SchemaReferenceWriter schemaReferenceWriter;
+ private volatile MessageIndexWriter messageIndexWriter;
+ 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();
+ schemaRegistry =
context.getProperty(SCHEMA_REGISTRY).asControllerService(SchemaRegistry.class);
+ schemaReferenceWriter =
context.getProperty(SCHEMA_REFERENCE_WRITER).asControllerService(SchemaReferenceWriter.class);
+ messageIndexWriter =
context.getProperty(MESSAGE_INDEX_WRITER).asControllerService(MessageIndexWriter.class);
+ schemaName = context.getProperty(SCHEMA_NAME);
+ schemaText = context.getProperty(SCHEMA_TEXT);
+ schemaBranchName = context.getProperty(SCHEMA_BRANCH_NAME);
+ schemaVersion = context.getProperty(SCHEMA_VERSION);
+ }
+
+ @Override
+ protected void init(final ControllerServiceInitializationContext config)
throws InitializationException {
+ super.init(config);
+ schemaCompiler = new ProtobufSchemaCompiler(getIdentifier(),
getLogger());
+ }
+
+ @Override
+ public RecordSchema getSchema(final Map<String, String> variables, final
RecordSchema readSchema) throws SchemaNotFoundException, IOException {
+ return createWriteContext(variables).recordSchema();
+ }
+
+ @Override
+ public RecordSetWriter createWriter(final ComponentLog logger, final
RecordSchema schema, final OutputStream out, final Map<String, String>
variables) throws SchemaNotFoundException, IOException {
+ final ProtobufWriteContext context = createWriteContext(variables);
+ return new WriteProtobufResultWithExternalSchema(context.schema(),
context.messageName(), context.recordSchema(),
+ context.schemaDefinition(), schemaReferenceWriter,
messageIndexWriter, variables, out);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = new
ArrayList<>(super.getSupportedPropertyDescriptors());
+ // The Schema Reference Reader is a read-side concern: a writer
determines its schema from the configured
+ // access strategy and writes references through the Schema Reference
Writer instead.
+ properties.removeIf(property ->
SCHEMA_REFERENCE_READER.getName().equals(property.getName()));
+ properties.add(MESSAGE_NAME_RESOLUTION_STRATEGY);
+ properties.add(MESSAGE_NAME_RESOLVER);
+ properties.add(MESSAGE_NAME);
+ properties.add(SCHEMA_REFERENCE_WRITER);
+ properties.add(MESSAGE_INDEX_WRITER);
+ return properties;
+ }
+
+ @Override
+ protected List<AllowableValue> getSchemaAccessStrategyValues() {
+ // Only the strategies that createSchemaDefinition supports are
offered; the inherited list also contains
+ // the Schema Reference Reader strategy, which cannot be used to
obtain a schema for writing.
+ return List.of(SCHEMA_NAME_PROPERTY, SCHEMA_TEXT_PROPERTY);
+ }
+
+ @Override
+ protected PropertyDescriptor buildSchemaTextProperty() {
+ return PROTOBUF_SCHEMA_TEXT;
+ }
+
+ private ProtobufWriteContext createWriteContext(final Map<String, String>
variables) throws SchemaNotFoundException, IOException {
+ final SchemaDefinition schemaDefinition =
createSchemaDefinition(variables);
+ final Schema schema =
schemaCompiler.compileOrGetFromCache(schemaDefinition);
+ final MessageName messageName =
messageNameResolver.getMessageName(variables, schemaDefinition,
EMPTY_INPUT_STREAM);
+
+ final ProtoSchemaParser schemaParser = new ProtoSchemaParser(schema);
+ final RecordSchema parsedSchema =
schemaParser.createSchema(messageName.getFullyQualifiedName());
+ // Preserve the schema identifier from the SchemaDefinition so the
configured Schema Reference Writer can
+ // write the correct schema id in the Confluent header.
Review Comment:
Useful comment, but better avoid referencing Confluent-specific details.
```suggestion
// Preserve the schema identifier required by the configured Schema
Reference Writer.
```
##########
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:
Better avoid Confluent specific details and add more clarification about the
relationship with Message Index Writer.
```suggestion
.description("The Controller Service used to write schema reference
information before any message index information "
+ "and the Protobuf payload. The selected implementation
must be compatible with the target wire format and any "
+ "configured Message Index Writer. When not configured, no
schema reference information is written.")
```
##########
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:
Better avoid Confluent specific details.
```suggestion
.description("The Controller Service used to write information identifying
the selected message within the Protobuf "
+ "schema. Message index information is written after
any schema reference information and before the Protobuf "
+ "payload. The selected implementation must be
compatible with the target wire format and any configured "
+ "Schema Reference Writer.")
```
##########
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.""")
Review Comment:
Avoid referencing Confluent as part of the generic capability description. A
reminded-type mentioning of the need of both services for Confluent is okay
here though.
```suggestion
@CapabilityDescription("""
Serializes NiFi Records into Protocol Buffers binary format. \
Supports inline schema text and schema registry lookup for determining
the Proto schema. \
Optional Schema Reference Writer and Message Index Writer Controller
Services can add framing before the \
Protobuf payload. When both are configured, schema reference information
is written first, followed by message \
index information. Selected implementations must be compatible with each
other and with the target wire format. \
Confluent Protobuf wire format requires both compatible services. \
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 because 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.""")
```
##########
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.")
+ .required(false)
+ .identifiesControllerService(MessageIndexWriter.class)
+ .build();
+
+ private static final PropertyDescriptor PROTOBUF_SCHEMA_TEXT = 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 static final String PROTO_EXTENSION = ".proto";
+
+ private static final InputStream EMPTY_INPUT_STREAM = new
ByteArrayInputStream(new byte[0]);
+
+ private volatile ProtobufSchemaCompiler schemaCompiler;
+ private volatile MessageNameResolver messageNameResolver;
+ private volatile SchemaReferenceWriter schemaReferenceWriter;
+ private volatile MessageIndexWriter messageIndexWriter;
+ 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();
+ schemaRegistry =
context.getProperty(SCHEMA_REGISTRY).asControllerService(SchemaRegistry.class);
+ schemaReferenceWriter =
context.getProperty(SCHEMA_REFERENCE_WRITER).asControllerService(SchemaReferenceWriter.class);
+ messageIndexWriter =
context.getProperty(MESSAGE_INDEX_WRITER).asControllerService(MessageIndexWriter.class);
+ schemaName = context.getProperty(SCHEMA_NAME);
+ schemaText = context.getProperty(SCHEMA_TEXT);
+ schemaBranchName = context.getProperty(SCHEMA_BRANCH_NAME);
+ schemaVersion = context.getProperty(SCHEMA_VERSION);
+ }
+
+ @Override
+ protected void init(final ControllerServiceInitializationContext config)
throws InitializationException {
+ super.init(config);
+ schemaCompiler = new ProtobufSchemaCompiler(getIdentifier(),
getLogger());
+ }
+
+ @Override
+ public RecordSchema getSchema(final Map<String, String> variables, final
RecordSchema readSchema) throws SchemaNotFoundException, IOException {
+ return createWriteContext(variables).recordSchema();
+ }
+
+ @Override
+ public RecordSetWriter createWriter(final ComponentLog logger, final
RecordSchema schema, final OutputStream out, final Map<String, String>
variables) throws SchemaNotFoundException, IOException {
+ final ProtobufWriteContext context = createWriteContext(variables);
+ return new WriteProtobufResultWithExternalSchema(context.schema(),
context.messageName(), context.recordSchema(),
+ context.schemaDefinition(), schemaReferenceWriter,
messageIndexWriter, variables, out);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = new
ArrayList<>(super.getSupportedPropertyDescriptors());
+ // The Schema Reference Reader is a read-side concern: a writer
determines its schema from the configured
+ // access strategy and writes references through the Schema Reference
Writer instead.
+ properties.removeIf(property ->
SCHEMA_REFERENCE_READER.getName().equals(property.getName()));
+ properties.add(MESSAGE_NAME_RESOLUTION_STRATEGY);
+ properties.add(MESSAGE_NAME_RESOLVER);
+ properties.add(MESSAGE_NAME);
+ properties.add(SCHEMA_REFERENCE_WRITER);
+ properties.add(MESSAGE_INDEX_WRITER);
+ return properties;
+ }
+
+ @Override
+ protected List<AllowableValue> getSchemaAccessStrategyValues() {
+ // Only the strategies that createSchemaDefinition supports are
offered; the inherited list also contains
+ // the Schema Reference Reader strategy, which cannot be used to
obtain a schema for writing.
+ return List.of(SCHEMA_NAME_PROPERTY, SCHEMA_TEXT_PROPERTY);
+ }
+
+ @Override
+ protected PropertyDescriptor buildSchemaTextProperty() {
+ return PROTOBUF_SCHEMA_TEXT;
+ }
+
+ private ProtobufWriteContext createWriteContext(final Map<String, String>
variables) throws SchemaNotFoundException, IOException {
+ final SchemaDefinition schemaDefinition =
createSchemaDefinition(variables);
+ final Schema schema =
schemaCompiler.compileOrGetFromCache(schemaDefinition);
+ final MessageName messageName =
messageNameResolver.getMessageName(variables, schemaDefinition,
EMPTY_INPUT_STREAM);
Review Comment:
While technically correct because the 0-length byte array behaves like an
immutable object and can be reused safely, it is surprising and gives the
code-reader a pause. That being said, I'm okay keeping it like this - but let
me offer an alternative:
```suggestion
final MessageName messageName =
messageNameResolver.getMessageName(variables, schemaDefinition,
InputStream.nullInputStream());
```
##########
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.")
+ .required(false)
+ .identifiesControllerService(MessageIndexWriter.class)
+ .build();
+
+ private static final PropertyDescriptor PROTOBUF_SCHEMA_TEXT = 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 static final String PROTO_EXTENSION = ".proto";
+
+ private static final InputStream EMPTY_INPUT_STREAM = new
ByteArrayInputStream(new byte[0]);
+
+ private volatile ProtobufSchemaCompiler schemaCompiler;
+ private volatile MessageNameResolver messageNameResolver;
+ private volatile SchemaReferenceWriter schemaReferenceWriter;
+ private volatile MessageIndexWriter messageIndexWriter;
+ 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();
+ schemaRegistry =
context.getProperty(SCHEMA_REGISTRY).asControllerService(SchemaRegistry.class);
+ schemaReferenceWriter =
context.getProperty(SCHEMA_REFERENCE_WRITER).asControllerService(SchemaReferenceWriter.class);
+ messageIndexWriter =
context.getProperty(MESSAGE_INDEX_WRITER).asControllerService(MessageIndexWriter.class);
+ schemaName = context.getProperty(SCHEMA_NAME);
+ schemaText = context.getProperty(SCHEMA_TEXT);
+ schemaBranchName = context.getProperty(SCHEMA_BRANCH_NAME);
+ schemaVersion = context.getProperty(SCHEMA_VERSION);
+ }
+
Review Comment:
An inline schema via Schema Text doesn't have a Confluent-compatible
identifier.
Not sure if this is intended, but if it is, we can add a custom validate to
make sure that when the ConfluentEncodedSchemaReferenceWriter is set, the
Schema Name strategy is selected as well.
Also we can add validation for literal Schema Text values.
```suggestion
@Override
protected Collection<ValidationResult> customValidate(final
ValidationContext validationContext) {
final List<ValidationResult> results = new
ArrayList<>(super.customValidate(validationContext));
validateLiteralSchemaText(validationContext, results);
validateSchemaReferenceWriterCompatibility(validationContext, results);
return results;
}
private void validateLiteralSchemaText(final ValidationContext
validationContext, final List<ValidationResult> results) {
final String schemaAccessStrategy =
validationContext.getProperty(SCHEMA_ACCESS_STRATEGY).getValue();
if (!SCHEMA_TEXT_PROPERTY.getValue().equals(schemaAccessStrategy)) {
return;
}
final String schemaTextValue =
validationContext.getProperty(SCHEMA_TEXT).getValue();
if (schemaTextValue == null || schemaTextValue.isBlank()
||
validationContext.isExpressionLanguagePresent(schemaTextValue)) {
return;
}
try {
final SchemaIdentifier schemaIdentifier = SchemaIdentifier.builder()
.name(sha256Hex(schemaTextValue) + PROTO_EXTENSION)
.build();
final SchemaDefinition schemaDefinition = new
StandardSchemaDefinition(
schemaIdentifier, schemaTextValue,
SchemaDefinition.SchemaType.PROTOBUF);
final Schema compiledSchema =
schemaCompiler.compileOrGetFromCache(schemaDefinition);
validateLiteralMessageName(validationContext, compiledSchema,
results);
} catch (final SchemaCompilationException e) {
results.add(new ValidationResult.Builder()
.subject(SCHEMA_TEXT.getDisplayName())
.valid(false)
.explanation("Invalid Protocol Buffers schema: " +
e.getMessage())
.build());
}
}
private void validateLiteralMessageName(
final ValidationContext validationContext,
final Schema compiledSchema,
final List<ValidationResult> results) {
final String resolutionStrategy =
validationContext.getProperty(MESSAGE_NAME_RESOLUTION_STRATEGY).getValue();
if (!MESSAGE_NAME_PROPERTY.getValue().equals(resolutionStrategy)) {
return;
}
final String messageNameValue =
validationContext.getProperty(MESSAGE_NAME).getValue();
if (messageNameValue == null || messageNameValue.isBlank()
||
validationContext.isExpressionLanguagePresent(messageNameValue)) {
return;
}
if (!(compiledSchema.getType(messageNameValue) instanceof MessageType)) {
results.add(new ValidationResult.Builder()
.subject(MESSAGE_NAME.getDisplayName())
.input(messageNameValue)
.valid(false)
.explanation("Message name '%s' does not identify a message in
the configured Protocol Buffers schema"
.formatted(messageNameValue))
.build());
}
}
private void validateSchemaReferenceWriterCompatibility(
final ValidationContext validationContext,
final List<ValidationResult> results) {
if (!validationContext.getProperty(SCHEMA_REFERENCE_WRITER).isSet()) {
return;
}
final SchemaReferenceWriter referenceWriter =
validationContext.getProperty(SCHEMA_REFERENCE_WRITER)
.asControllerService(SchemaReferenceWriter.class);
if (referenceWriter == null) {
return;
}
final Set<SchemaField> missingFields = EnumSet.noneOf(SchemaField.class);
missingFields.addAll(referenceWriter.getRequiredSchemaFields());
missingFields.removeAll(getSuppliedSchemaFields(validationContext));
if (!missingFields.isEmpty()) {
results.add(new ValidationResult.Builder()
.subject(SCHEMA_REFERENCE_WRITER.getDisplayName())
.valid(false)
.explanation("The configured Schema Reference Writer requires
schema fields that are not provided "
+ "by the configured Schema Access Strategy and Schema
Registry: " + missingFields)
.build());
}
}
```
##########
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.")
+ .required(false)
+ .identifiesControllerService(MessageIndexWriter.class)
+ .build();
+
+ private static final PropertyDescriptor PROTOBUF_SCHEMA_TEXT = 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 static final String PROTO_EXTENSION = ".proto";
+
+ private static final InputStream EMPTY_INPUT_STREAM = new
ByteArrayInputStream(new byte[0]);
+
+ private volatile ProtobufSchemaCompiler schemaCompiler;
+ private volatile MessageNameResolver messageNameResolver;
+ private volatile SchemaReferenceWriter schemaReferenceWriter;
+ private volatile MessageIndexWriter messageIndexWriter;
+ 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();
+ schemaRegistry =
context.getProperty(SCHEMA_REGISTRY).asControllerService(SchemaRegistry.class);
+ schemaReferenceWriter =
context.getProperty(SCHEMA_REFERENCE_WRITER).asControllerService(SchemaReferenceWriter.class);
+ messageIndexWriter =
context.getProperty(MESSAGE_INDEX_WRITER).asControllerService(MessageIndexWriter.class);
+ schemaName = context.getProperty(SCHEMA_NAME);
+ schemaText = context.getProperty(SCHEMA_TEXT);
+ schemaBranchName = context.getProperty(SCHEMA_BRANCH_NAME);
+ schemaVersion = context.getProperty(SCHEMA_VERSION);
+ }
+
+ @Override
+ protected void init(final ControllerServiceInitializationContext config)
throws InitializationException {
+ super.init(config);
+ schemaCompiler = new ProtobufSchemaCompiler(getIdentifier(),
getLogger());
+ }
+
+ @Override
+ public RecordSchema getSchema(final Map<String, String> variables, final
RecordSchema readSchema) throws SchemaNotFoundException, IOException {
+ return createWriteContext(variables).recordSchema();
+ }
+
+ @Override
+ public RecordSetWriter createWriter(final ComponentLog logger, final
RecordSchema schema, final OutputStream out, final Map<String, String>
variables) throws SchemaNotFoundException, IOException {
+ final ProtobufWriteContext context = createWriteContext(variables);
Review Comment:
```suggestion
final ProtobufWriteContext context = createWriteContext(variables);
if (schemaReferenceWriter != null) {
schemaReferenceWriter.validateSchema(context.recordSchema());
}
```
##########
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.")
+ .required(false)
+ .identifiesControllerService(MessageIndexWriter.class)
+ .build();
+
+ private static final PropertyDescriptor PROTOBUF_SCHEMA_TEXT = 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 static final String PROTO_EXTENSION = ".proto";
+
+ private static final InputStream EMPTY_INPUT_STREAM = new
ByteArrayInputStream(new byte[0]);
+
+ private volatile ProtobufSchemaCompiler schemaCompiler;
+ private volatile MessageNameResolver messageNameResolver;
+ private volatile SchemaReferenceWriter schemaReferenceWriter;
+ private volatile MessageIndexWriter messageIndexWriter;
+ 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();
+ schemaRegistry =
context.getProperty(SCHEMA_REGISTRY).asControllerService(SchemaRegistry.class);
+ schemaReferenceWriter =
context.getProperty(SCHEMA_REFERENCE_WRITER).asControllerService(SchemaReferenceWriter.class);
+ messageIndexWriter =
context.getProperty(MESSAGE_INDEX_WRITER).asControllerService(MessageIndexWriter.class);
+ schemaName = context.getProperty(SCHEMA_NAME);
+ schemaText = context.getProperty(SCHEMA_TEXT);
+ schemaBranchName = context.getProperty(SCHEMA_BRANCH_NAME);
+ schemaVersion = context.getProperty(SCHEMA_VERSION);
+ }
+
+ @Override
+ protected void init(final ControllerServiceInitializationContext config)
throws InitializationException {
+ super.init(config);
+ schemaCompiler = new ProtobufSchemaCompiler(getIdentifier(),
getLogger());
+ }
+
+ @Override
+ public RecordSchema getSchema(final Map<String, String> variables, final
RecordSchema readSchema) throws SchemaNotFoundException, IOException {
+ return createWriteContext(variables).recordSchema();
+ }
+
+ @Override
+ public RecordSetWriter createWriter(final ComponentLog logger, final
RecordSchema schema, final OutputStream out, final Map<String, String>
variables) throws SchemaNotFoundException, IOException {
+ final ProtobufWriteContext context = createWriteContext(variables);
+ return new WriteProtobufResultWithExternalSchema(context.schema(),
context.messageName(), context.recordSchema(),
+ context.schemaDefinition(), schemaReferenceWriter,
messageIndexWriter, variables, out);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = new
ArrayList<>(super.getSupportedPropertyDescriptors());
+ // The Schema Reference Reader is a read-side concern: a writer
determines its schema from the configured
+ // access strategy and writes references through the Schema Reference
Writer instead.
Review Comment:
Unnecessary (and a bit confusing) comment.
```suggestion
```
--
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]
