This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 4abef54d295 NIFI-16349 Strip custom Protobuf options before Wire
schema compilation (#11710)
4abef54d295 is described below
commit 4abef54d295f2c3f55c8d09fb1e61f78625259eb
Author: Alaksiej Ščarbaty <[email protected]>
AuthorDate: Wed Sep 23 19:41:06 2026 +0200
NIFI-16349 Strip custom Protobuf options before Wire schema compilation
(#11710)
Signed-off-by: David Handermann <[email protected]>
---
.../protobuf/ProtobufCustomOptionRemover.java | 268 ++++++++++++++
.../services/protobuf/ProtobufSchemaCompiler.java | 7 +-
.../protobuf/TestProtobufSchemaCompiler.java | 403 +++++++++++++++++++++
.../protobuf/TestStandardProtobufReader.java | 78 ++++
.../test/custom_options_on_all_elements.proto | 66 ++++
.../test/nested_wrapper_with_field_meta.proto | 72 ++++
6 files changed, 891 insertions(+), 3 deletions(-)
diff --git
a/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/ProtobufCustomOptionRemover.java
b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/ProtobufCustomOptionRemover.java
new file mode 100644
index 00000000000..e147d604662
--- /dev/null
+++
b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/ProtobufCustomOptionRemover.java
@@ -0,0 +1,268 @@
+/*
+ * 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.Location;
+import com.squareup.wire.schema.internal.parser.EnumConstantElement;
+import com.squareup.wire.schema.internal.parser.EnumElement;
+import com.squareup.wire.schema.internal.parser.ExtendElement;
+import com.squareup.wire.schema.internal.parser.ExtensionsElement;
+import com.squareup.wire.schema.internal.parser.FieldElement;
+import com.squareup.wire.schema.internal.parser.GroupElement;
+import com.squareup.wire.schema.internal.parser.MessageElement;
+import com.squareup.wire.schema.internal.parser.OneOfElement;
+import com.squareup.wire.schema.internal.parser.OptionElement;
+import com.squareup.wire.schema.internal.parser.ProtoFileElement;
+import com.squareup.wire.schema.internal.parser.ProtoParser;
+import com.squareup.wire.schema.internal.parser.RpcElement;
+import com.squareup.wire.schema.internal.parser.ServiceElement;
+import com.squareup.wire.schema.internal.parser.TypeElement;
+import org.apache.nifi.logging.ComponentLog;
+
+import java.util.ArrayList;
+import java.util.List;
+
+final class ProtobufCustomOptionRemover {
+
+ private ProtobufCustomOptionRemover() {
+ }
+
+ /**
+ * Removes custom (parenthesized) Protocol Buffer options from schema text.
+ * Built-in options such as {@code packed} and {@code java_package} are
retained.
+ * When the text cannot be parsed, the original string is returned.
+ *
+ * @param schemaText source proto schema to remove custom options from
+ * @param location source location used for parse diagnostics and
regenerated comments
+ * @param logger component logger
+ * @return schema text without custom options, or the original text when
parsing fails or no custom options are present
+ */
+ static String remove(final String schemaText, final Location location,
final ComponentLog logger) {
+ if (schemaText == null || schemaText.isBlank()) {
+ return schemaText;
+ }
+
+ final ProtoFileElement parsed;
+ try {
+ parsed = ProtoParser.Companion.parse(location, schemaText);
+ } catch (final IllegalStateException e) {
+ logger.warn("Failed to parse Protobuf schema [{}] while removing
custom options; using the original schema text", location, e);
+ return schemaText;
+ }
+
+ final ProtoFileElement withoutCustomOptions = removeFromFile(parsed);
+ if (parsed.equals(withoutCustomOptions)) {
+ return schemaText;
+ }
+
+ return withoutCustomOptions.toSchema();
+ }
+
+ private static ProtoFileElement removeFromFile(final ProtoFileElement
file) {
+ return new ProtoFileElement(
+ file.getLocation(),
+ file.getPackageName(),
+ file.getSyntax(),
+ file.getImports(),
+ file.getPublicImports(),
+ file.getWeakImports(),
+ removeFromTypes(file.getTypes()),
+ removeFromServices(file.getServices()),
+ removeFromExtendDeclarations(file.getExtendDeclarations()),
+ removeCustomOptions(file.getOptions())
+ );
+ }
+
+ private static List<TypeElement> removeFromTypes(final List<TypeElement>
types) {
+ final List<TypeElement> updatedTypes = new ArrayList<>(types.size());
+ for (final TypeElement type : types) {
+ updatedTypes.add(removeFromType(type));
+ }
+
+ return updatedTypes;
+ }
+
+ private static TypeElement removeFromType(final TypeElement type) {
+ if (type instanceof MessageElement message) {
+ return new MessageElement(
+ message.getLocation(),
+ message.getName(),
+ message.getDocumentation(),
+ removeFromTypes(message.getNestedTypes()),
+ removeCustomOptions(message.getOptions()),
+ message.getReserveds(),
+ removeFromFields(message.getFields()),
+ removeFromOneOfs(message.getOneOfs()),
+ removeFromExtensions(message.getExtensions()),
+ removeFromGroups(message.getGroups()),
+ removeFromExtendDeclarations(message.getExtendDeclarations())
+ );
+ }
+
+ if (type instanceof EnumElement enumeration) {
+ return new EnumElement(
+ enumeration.getLocation(),
+ enumeration.getName(),
+ enumeration.getDocumentation(),
+ removeCustomOptions(enumeration.getOptions()),
+ removeFromEnumConstants(enumeration.getConstants()),
+ enumeration.getReserveds()
+ );
+ }
+
+ return type;
+ }
+
+ private static List<FieldElement> removeFromFields(final
List<FieldElement> fields) {
+ final List<FieldElement> updatedFields = new
ArrayList<>(fields.size());
+ for (final FieldElement field : fields) {
+ updatedFields.add(new FieldElement(
+ field.getLocation(),
+ field.getLabel(),
+ field.getType(),
+ field.getName(),
+ field.getDefaultValue(),
+ field.getJsonName(),
+ field.getTag(),
+ field.getDocumentation(),
+ removeCustomOptions(field.getOptions())
+ ));
+ }
+
+ return updatedFields;
+ }
+
+ private static List<OneOfElement> removeFromOneOfs(final
List<OneOfElement> oneOfs) {
+ final List<OneOfElement> updatedOneOfs = new
ArrayList<>(oneOfs.size());
+ for (final OneOfElement oneOf : oneOfs) {
+ updatedOneOfs.add(new OneOfElement(
+ oneOf.getName(),
+ oneOf.getDocumentation(),
+ removeFromFields(oneOf.getFields()),
+ removeFromGroups(oneOf.getGroups()),
+ removeCustomOptions(oneOf.getOptions()),
+ oneOf.getLocation()
+ ));
+ }
+
+ return updatedOneOfs;
+ }
+
+ private static List<GroupElement> removeFromGroups(final
List<GroupElement> groups) {
+ final List<GroupElement> updatedGroups = new
ArrayList<>(groups.size());
+ for (final GroupElement group : groups) {
+ updatedGroups.add(new GroupElement(
+ group.getLabel(),
+ group.getLocation(),
+ group.getName(),
+ group.getTag(),
+ group.getDocumentation(),
+ removeFromFields(group.getFields())
+ ));
+ }
+
+ return updatedGroups;
+ }
+
+ private static List<ExtensionsElement> removeFromExtensions(final
List<ExtensionsElement> extensions) {
+ final List<ExtensionsElement> updatedExtensions = new
ArrayList<>(extensions.size());
+ for (final ExtensionsElement extension : extensions) {
+ updatedExtensions.add(new ExtensionsElement(
+ extension.getLocation(),
+ extension.getDocumentation(),
+ extension.getValues(),
+ removeCustomOptions(extension.getOptions())
+ ));
+ }
+
+ return updatedExtensions;
+ }
+
+ private static List<ExtendElement> removeFromExtendDeclarations(final
List<ExtendElement> extendDeclarations) {
+ final List<ExtendElement> updatedDeclarations = new
ArrayList<>(extendDeclarations.size());
+ for (final ExtendElement extendDeclaration : extendDeclarations) {
+ updatedDeclarations.add(new ExtendElement(
+ extendDeclaration.getLocation(),
+ extendDeclaration.getName(),
+ extendDeclaration.getDocumentation(),
+ removeFromFields(extendDeclaration.getFields())
+ ));
+ }
+
+ return updatedDeclarations;
+ }
+
+ private static List<EnumConstantElement> removeFromEnumConstants(final
List<EnumConstantElement> constants) {
+ final List<EnumConstantElement> updatedConstants = new
ArrayList<>(constants.size());
+ for (final EnumConstantElement constant : constants) {
+ updatedConstants.add(new EnumConstantElement(
+ constant.getLocation(),
+ constant.getName(),
+ constant.getTag(),
+ constant.getDocumentation(),
+ removeCustomOptions(constant.getOptions())
+ ));
+ }
+
+ return updatedConstants;
+ }
+
+ private static List<ServiceElement> removeFromServices(final
List<ServiceElement> services) {
+ final List<ServiceElement> updatedServices = new
ArrayList<>(services.size());
+ for (final ServiceElement service : services) {
+ updatedServices.add(new ServiceElement(
+ service.getLocation(),
+ service.getName(),
+ service.getDocumentation(),
+ removeFromRpcs(service.getRpcs()),
+ removeCustomOptions(service.getOptions())
+ ));
+ }
+
+ return updatedServices;
+ }
+
+ private static List<RpcElement> removeFromRpcs(final List<RpcElement>
rpcs) {
+ final List<RpcElement> updatedRpcs = new ArrayList<>(rpcs.size());
+ for (final RpcElement rpc : rpcs) {
+ updatedRpcs.add(new RpcElement(
+ rpc.getLocation(),
+ rpc.getName(),
+ rpc.getDocumentation(),
+ rpc.getRequestType(),
+ rpc.getResponseType(),
+ rpc.getRequestStreaming(),
+ rpc.getResponseStreaming(),
+ removeCustomOptions(rpc.getOptions())
+ ));
+ }
+
+ return updatedRpcs;
+ }
+
+ private static List<OptionElement> removeCustomOptions(final
List<OptionElement> options) {
+ final List<OptionElement> retainedOptions = new
ArrayList<>(options.size());
+ for (final OptionElement option : options) {
+ // Custom options are always parenthesized. Built-in options are
not.
+ if (!option.isParenthesized()) {
+ retainedOptions.add(option);
+ }
+ }
+
+ return retainedOptions;
+ }
+}
diff --git
a/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/ProtobufSchemaCompiler.java
b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/ProtobufSchemaCompiler.java
index da4f20eac55..6e264a19292 100644
---
a/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/ProtobufSchemaCompiler.java
+++
b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/ProtobufSchemaCompiler.java
@@ -29,6 +29,7 @@ import org.apache.nifi.serialization.record.SchemaIdentifier;
import java.io.IOException;
import java.io.UncheckedIOException;
+import java.nio.charset.StandardCharsets;
import java.nio.file.FileSystems;
import java.nio.file.Files;
import java.nio.file.Path;
@@ -195,10 +196,10 @@ final class ProtobufSchemaCompiler {
final String schemaFileName = generateSchemaFileName(schemaDefinition);
final Path schemaFile = tempDir.resolve(schemaFileName);
- // Write schema text to file
- Files.write(schemaFile, schemaDefinition.getText().getBytes(), CREATE,
WRITE, TRUNCATE_EXISTING);
+ final String preparedSchemaDefinition =
ProtobufCustomOptionRemover.remove(schemaDefinition.getText(),
Location.get(schemaFileName), logger);
+ Files.write(schemaFile,
preparedSchemaDefinition.getBytes(StandardCharsets.UTF_8), CREATE, WRITE,
TRUNCATE_EXISTING);
logger.debug("Successfully wrote schema to file: {} (string length:
{})",
- schemaFile, schemaDefinition.getText().length());
+ schemaFile, preparedSchemaDefinition.length());
}
/**
diff --git
a/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/java/org/apache/nifi/services/protobuf/TestProtobufSchemaCompiler.java
b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/java/org/apache/nifi/services/protobuf/TestProtobufSchemaCompiler.java
new file mode 100644
index 00000000000..64128d24439
--- /dev/null
+++
b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/java/org/apache/nifi/services/protobuf/TestProtobufSchemaCompiler.java
@@ -0,0 +1,403 @@
+/*
+ * 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.EnumConstant;
+import com.squareup.wire.schema.EnumType;
+import com.squareup.wire.schema.Field;
+import com.squareup.wire.schema.MessageType;
+import com.squareup.wire.schema.OneOf;
+import com.squareup.wire.schema.Options;
+import com.squareup.wire.schema.ProtoFile;
+import com.squareup.wire.schema.Rpc;
+import com.squareup.wire.schema.Schema;
+import com.squareup.wire.schema.SchemaException;
+import com.squareup.wire.schema.Service;
+import com.squareup.wire.schema.internal.parser.OptionElement;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.schemaregistry.services.SchemaDefinition;
+import org.apache.nifi.schemaregistry.services.StandardSchemaDefinition;
+import org.apache.nifi.serialization.record.SchemaIdentifier;
+import org.apache.nifi.util.MockComponentLog;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.function.Executable;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestProtobufSchemaCompiler {
+
+ private static final ComponentLog COMPONENT_LOG = new
MockComponentLog("test", TestProtobufSchemaCompiler.class);
+
+ private static final String PACKAGE = "example";
+ private static final String SAMPLE_TYPE = PACKAGE + ".Sample";
+ private static final String WRAPPER_TYPE = PACKAGE + ".Wrapper";
+ private static final String WRAPPER_PAYLOAD_TYPE = WRAPPER_TYPE +
".Payload";
+
+ private static final String SAMPLE_PROTO = "sample.proto";
+ private static final String COMMON_PROTO = "common.proto";
+
+ private static final String FIELD_VALUE = "value";
+ private static final String FIELD_EXTRA = "extra";
+ private static final String ENUM_RED = "RED";
+
+ private final ProtobufSchemaCompiler compiler = new
ProtobufSchemaCompiler("test", COMPONENT_LOG);
+
+ @Test
+ void testCompileSchemaWithCustomOptionWithoutImport() {
+ final String schemaText = """
+ syntax = "proto3";
+ package example;
+
+ message Sample {
+ optional int32 enabled = 1 [(confluent.field_meta) = {
+ params: [
+ {
+ key: "connect.type",
+ value: "int16"
+ }
+ ]
+ }];
+ }
+ """;
+ final SchemaDefinition schemaDefinition =
schemaDefinition(SAMPLE_PROTO, schemaText);
+
+ final Schema schema = compiler.compileOrGetFromCache(schemaDefinition);
+
+ final Field enabled = schema.getField(SAMPLE_TYPE, "enabled");
+
+ assertNotNull(enabled);
+ assertNoParenthesizedOptions(enabled.getOptions());
+ }
+
+ @Test
+ void testCompileNestedSchemaWithRepeatedFieldMetaOptions() throws
IOException {
+ final String schemaText =
readTestResource("/org/apache/nifi/protobuf/test/nested_wrapper_with_field_meta.proto");
+ final SchemaDefinition schemaDefinition =
schemaDefinition(SAMPLE_PROTO, schemaText);
+
+ final Schema schema = compiler.compileOrGetFromCache(schemaDefinition);
+
+ final Field flagA = schema.getField(WRAPPER_PAYLOAD_TYPE, "flag_a");
+ final Field flagB = schema.getField(WRAPPER_PAYLOAD_TYPE, "flag_b");
+ final Field flagC = schema.getField(WRAPPER_PAYLOAD_TYPE, "flag_c");
+
+ assertNotNull(schema.getType(WRAPPER_TYPE));
+ assertNotNull(schema.getType(WRAPPER_TYPE + ".Origin"));
+ assertNotNull(schema.getType(WRAPPER_TYPE + ".chunk"));
+ assertNotNull(flagA);
+ assertNotNull(flagB);
+ assertNotNull(flagC);
+ assertNoParenthesizedOptions(flagA.getOptions());
+ assertNoParenthesizedOptions(flagB.getOptions());
+ assertNoParenthesizedOptions(flagC.getOptions());
+ }
+
+ @Test
+ void testCompileKeepsPackedWhenCustomFieldOptionPresent() {
+ final String schemaText = """
+ syntax = "proto2";
+ package example;
+
+ message Sample {
+ repeated int32 values = 1 [packed = true, (confluent.field_meta)
= {
+ tags: "x"
+ }];
+ }
+ """;
+ final SchemaDefinition schemaDefinition =
schemaDefinition(SAMPLE_PROTO, schemaText);
+
+ final Schema schema = compiler.compileOrGetFromCache(schemaDefinition);
+ final Field values = schema.getField(SAMPLE_TYPE, "values");
+
+ assertNotNull(values);
+ assertTrue(values.isRepeated());
+ assertTrue(values.isPacked());
+ assertNoParenthesizedOptions(values.getOptions());
+ }
+
+ @Test
+ void testCompileKeepsJavaPackageWhenCustomFileOptionPresent() {
+ final String schemaText = """
+ syntax = "proto3";
+ package example;
+
+ option java_package = "com.example";
+ option (custom.file_meta) = "docs";
+
+ message Sample {
+ int32 value = 1;
+ }
+ """;
+ final SchemaDefinition schemaDefinition =
schemaDefinition(SAMPLE_PROTO, schemaText);
+
+ final Schema schema = compiler.compileOrGetFromCache(schemaDefinition);
+ final ProtoFile protoFile = schema.protoFile(SAMPLE_PROTO);
+
+ assertNotNull(schema.getType(SAMPLE_TYPE));
+ assertNotNull(protoFile);
+ assertEquals("com.example", protoFile.javaPackage());
+ assertNoParenthesizedOptions(protoFile.getOptions());
+ }
+
+ @Test
+ void testCompileSchemaWithoutCustomOptions() {
+ final String schemaText = """
+ syntax = "proto3";
+ package example;
+
+ message Sample {
+ int32 value = 1;
+ }
+ """;
+ final SchemaDefinition schemaDefinition =
schemaDefinition(SAMPLE_PROTO, schemaText);
+
+ final Schema schema = compiler.compileOrGetFromCache(schemaDefinition);
+
+ assertNotNull(schema.getField(SAMPLE_TYPE, FIELD_VALUE));
+ }
+
+ @Test
+ void testCompileReferencedSchemaWithCustomOptions() {
+ final String referencedSchemaText = """
+ syntax = "proto3";
+ package example;
+
+ message Common {
+ int32 id = 1 [(custom.field_meta) = "c"];
+ }
+ """;
+ final String rootSchemaText = """
+ syntax = "proto3";
+ package example;
+
+ import "common.proto";
+
+ message Root {
+ Common item = 1 [(custom.field_meta) = "r"];
+ }
+ """;
+ final SchemaDefinition referencedSchema =
schemaDefinition(COMMON_PROTO, referencedSchemaText);
+ final SchemaDefinition rootSchema = schemaDefinition(SAMPLE_PROTO,
rootSchemaText, Map.of(COMMON_PROTO, referencedSchema));
+
+ final Schema schema = compiler.compileOrGetFromCache(rootSchema);
+
+ final Field id = schema.getField(PACKAGE + ".Common", "id");
+ final Field item = schema.getField(PACKAGE + ".Root", "item");
+
+ assertNotNull(id);
+ assertNoParenthesizedOptions(id.getOptions());
+ assertNotNull(item);
+ assertNoParenthesizedOptions(item.getOptions());
+ }
+
+ @Test
+ void testCompileFailsWhenMessageTypeIsUnresolved() {
+ final String schemaText = """
+ syntax = "proto3";
+ package example;
+
+ message Root {
+ Common item = 1;
+ }
+ """;
+ final SchemaDefinition schemaDefinition =
schemaDefinition(SAMPLE_PROTO, schemaText);
+
+ final RuntimeException exception =
assertThrows(RuntimeException.class, () ->
compiler.compileOrGetFromCache(schemaDefinition));
+
+ assertInstanceOf(SchemaException.class, exception.getCause());
+ }
+
+ @Test
+ void testCompileFailsWhenTypeImportIsMissing() {
+ final String schemaText = """
+ syntax = "proto3";
+ package example;
+
+ import "common.proto";
+
+ message Root {
+ Common item = 1;
+ }
+ """;
+ final SchemaDefinition schemaDefinition =
schemaDefinition(SAMPLE_PROTO, schemaText);
+
+ final RuntimeException exception =
assertThrows(RuntimeException.class, () ->
compiler.compileOrGetFromCache(schemaDefinition));
+
+ assertInstanceOf(SchemaException.class, exception.getCause());
+ }
+
+ @Test
+ void testCompileNestedMessageAndEnumWithCustomOptions() {
+ final String schemaText = """
+ syntax = "proto3";
+ package example;
+
+ message Outer {
+ option (custom.message_meta) = true;
+
+ message Inner {
+ int32 value = 1 [(custom.field_meta) = "y"];
+ }
+
+ Inner inner = 1;
+ }
+
+ enum Color {
+ option (custom.enum_meta) = true;
+ RED = 0 [(custom.enum_value_meta) = "r"];
+ GREEN = 1;
+ }
+ """;
+ final SchemaDefinition schemaDefinition =
schemaDefinition(SAMPLE_PROTO, schemaText);
+
+ final Schema schema = compiler.compileOrGetFromCache(schemaDefinition);
+ final MessageType outer = assertInstanceOf(MessageType.class,
schema.getType(PACKAGE + ".Outer"));
+ final EnumType color = assertInstanceOf(EnumType.class,
schema.getType(PACKAGE + ".Color"));
+ final EnumConstant red = color.constant(ENUM_RED);
+
+ assertNotNull(outer.field("inner"));
+ assertNotNull(schema.getField(PACKAGE + ".Outer.Inner", FIELD_VALUE));
+ assertNotNull(red);
+ assertNotNull(color.constant("GREEN"));
+ assertNoParenthesizedOptions(color.getOptions());
+ assertNoParenthesizedOptions(red.getOptions());
+ }
+
+ @Test
+ void testPayloadExtendDeclarationsArePreserved() {
+ final String schemaText = """
+ syntax = "proto2";
+ package example;
+
+ message Base {
+ extensions 100 to 199;
+ optional string name = 1;
+ }
+
+ extend Base {
+ optional int32 extra = 100;
+ }
+ """;
+ final SchemaDefinition schemaDefinition =
schemaDefinition(SAMPLE_PROTO, schemaText);
+
+ final Schema schema = compiler.compileOrGetFromCache(schemaDefinition);
+ final MessageType base = assertInstanceOf(MessageType.class,
schema.getType(PACKAGE + ".Base"));
+ final ProtoFile protoFile = schema.protoFile(SAMPLE_PROTO);
+
+ assertNotNull(base.field("name"));
+ assertNotNull(protoFile);
+ assertEquals(1, protoFile.getExtendList().size());
+ assertEquals(FIELD_EXTRA,
protoFile.getExtendList().get(0).getFields().get(0).getName());
+ }
+
+ @Test
+ void testCompileRemovesCustomOptionsFromAllSyntaxElements() throws
IOException {
+ final String schemaText =
readTestResource("/org/apache/nifi/protobuf/test/custom_options_on_all_elements.proto");
+ final SchemaDefinition schemaDefinition =
schemaDefinition(SAMPLE_PROTO, schemaText);
+
+ final Schema schema = compiler.compileOrGetFromCache(schemaDefinition);
+
+ final ProtoFile protoFile = schema.protoFile(SAMPLE_PROTO);
+ assertNotNull(protoFile);
+ assertEquals("com.example.comprehensive", protoFile.javaPackage());
+ assertNoParenthesizedOptions(protoFile.getOptions());
+
+ final MessageType wrapper = assertInstanceOf(MessageType.class,
schema.getType(WRAPPER_TYPE));
+ assertNoParenthesizedOptions(wrapper.getOptions());
+
+ final Field flag = schema.getField(WRAPPER_TYPE, "flag");
+ assertNotNull(flag);
+ assertNoParenthesizedOptions(flag.getOptions());
+
+ final OneOf kind = wrapper.oneOf("kind");
+ assertNotNull(kind);
+ assertNoParenthesizedOptions(kind.getOptions());
+
+ final Field text = schema.getField(WRAPPER_TYPE, "text");
+ assertNotNull(text);
+ assertNoParenthesizedOptions(text.getOptions());
+
+ assertEquals(1, protoFile.getExtendList().size());
+ final Field extra =
protoFile.getExtendList().get(0).getFields().get(0);
+ assertEquals(FIELD_EXTRA, extra.getName());
+ assertNoParenthesizedOptions(extra.getOptions());
+
+ assertNotNull(schema.getType(WRAPPER_TYPE + ".Nested"));
+ assertEquals(1, wrapper.getNestedExtendList().size());
+ final Field nestedExtra =
wrapper.getNestedExtendList().get(0).getFields().get(0);
+ assertEquals("nested_extra", nestedExtra.getName());
+ assertNoParenthesizedOptions(nestedExtra.getOptions());
+
+ final EnumType color = assertInstanceOf(EnumType.class,
schema.getType(WRAPPER_TYPE + ".Color"));
+ assertNoParenthesizedOptions(color.getOptions());
+
+ final EnumConstant red = color.constant(ENUM_RED);
+ assertNotNull(red);
+ assertNoParenthesizedOptions(red.getOptions());
+
+ final Service catalog = schema.getService(PACKAGE + ".Catalog");
+ assertNotNull(catalog);
+ assertNoParenthesizedOptions(catalog.options());
+
+ final Rpc lookup = catalog.rpc("Lookup");
+ assertNotNull(lookup);
+ assertEquals(PACKAGE + ".Request", lookup.getRequestType().toString());
+ assertEquals(PACKAGE + ".Response",
lookup.getResponseType().toString());
+ assertNoParenthesizedOptions(lookup.getOptions());
+ }
+
+ private void assertNoParenthesizedOptions(final Options options) {
+ final List<OptionElement> elements = options.getElements();
+ final List<Executable> assertions = new ArrayList<>(elements.size());
+ for (final OptionElement option : elements) {
+ assertions.add(() -> assertFalse(option.isParenthesized()));
+ }
+
+ assertAll(assertions);
+ }
+
+ private String readTestResource(final String resourcePath) throws
IOException {
+ try (InputStream inputStream =
getClass().getResourceAsStream(resourcePath)) {
+ assertNotNull(inputStream, "Test resource not found: " +
resourcePath);
+ return new String(inputStream.readAllBytes(),
StandardCharsets.UTF_8);
+ }
+ }
+
+ private SchemaDefinition schemaDefinition(final String name, final String
schemaText) {
+ return schemaDefinition(name, schemaText, Map.of());
+ }
+
+ private SchemaDefinition schemaDefinition(final String name, final String
schemaText, final Map<String, SchemaDefinition> references) {
+ return new StandardSchemaDefinition(
+ SchemaIdentifier.builder().name(name).build(),
+ schemaText,
+ SchemaDefinition.SchemaType.PROTOBUF,
+ references);
+ }
+}
diff --git
a/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/java/org/apache/nifi/services/protobuf/TestStandardProtobufReader.java
b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/java/org/apache/nifi/services/protobuf/TestStandardProtobufReader.java
index 54b36a267b4..edd01f7b5ef 100644
---
a/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/java/org/apache/nifi/services/protobuf/TestStandardProtobufReader.java
+++
b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/java/org/apache/nifi/services/protobuf/TestStandardProtobufReader.java
@@ -16,6 +16,10 @@
*/
package org.apache.nifi.services.protobuf;
+import com.google.protobuf.DescriptorProtos.DescriptorProto;
+import com.google.protobuf.DescriptorProtos.FieldDescriptorProto;
+import com.google.protobuf.DescriptorProtos.FileDescriptorProto;
+import com.google.protobuf.Descriptors.FileDescriptor;
import com.google.protobuf.DynamicMessage;
import org.apache.nifi.schema.access.SchemaNotFoundException;
import org.apache.nifi.schemaregistry.services.SchemaDefinition;
@@ -216,6 +220,80 @@ class TestStandardProtobufReader extends
StandardProtobufReaderTestBase {
}
}
+ @Nested
+ class WithCustomOptions {
+
+ private static final String SCHEMA_FILE_NAME = "sample.proto";
+ private static final String PACKAGE_NAME = "example";
+ private static final String MESSAGE_NAME = "Sample";
+ private static final String MESSAGE_TYPE = PACKAGE_NAME + "." +
MESSAGE_NAME;
+ private static final String ENABLED_FIELD = "enabled";
+ private static final String NAME_FIELD = "name";
+ private static final int ENABLED_VALUE = 42;
+ private static final String NAME_VALUE = "banner";
+ private static final String SCHEMA_TEXT = """
+ syntax = "proto3";
+ package example;
+
+ message Sample {
+ int32 enabled = 1 [(confluent.field_meta) = {
+ params: [
+ {
+ key: "connect.type",
+ value: "int16"
+ }
+ ]
+ }];
+ string name = 2;
+ }
+ """;
+
+ @Test
+ void testCreateRecordReaderWithCustomOptionWithoutImport() throws
Exception {
+ mockSchemaRegistry.returnSchemaText(SCHEMA_TEXT);
+
mockSchemaReferenceReader.returnSchemaIdentifierWithName(SCHEMA_FILE_NAME);
+ mockMessageNameResolver.returnMessageName(MESSAGE_TYPE);
+ enableAllControllerServices();
+
+ final InputStream testDataStream =
generateInputDataForSampleWithCustomOption();
+ final RecordReader recordReader =
standardProtobufReader.createRecordReader(emptyMap(), testDataStream, 0L,
runner.getLogger());
+ final Record record = recordReader.nextRecord();
+
+ assertNotNull(recordReader);
+ assertNotNull(record);
+ assertEquals(ENABLED_VALUE, record.getAsInt(ENABLED_FIELD));
+ assertEquals(NAME_VALUE, record.getAsString(NAME_FIELD));
+ }
+
+ private InputStream generateInputDataForSampleWithCustomOption()
throws DescriptorValidationException {
+ final FileDescriptorProto fileDescriptorProto =
FileDescriptorProto.newBuilder()
+ .setName(SCHEMA_FILE_NAME)
+ .setPackage(PACKAGE_NAME)
+ .setSyntax("proto3")
+ .addMessageType(DescriptorProto.newBuilder()
+ .setName(MESSAGE_NAME)
+ .addField(FieldDescriptorProto.newBuilder()
+ .setName(ENABLED_FIELD)
+ .setNumber(1)
+ .setType(FieldDescriptorProto.Type.TYPE_INT32)
+ .build())
+ .addField(FieldDescriptorProto.newBuilder()
+ .setName(NAME_FIELD)
+ .setNumber(2)
+ .setType(FieldDescriptorProto.Type.TYPE_STRING)
+ .build())
+ .build())
+ .build();
+ final FileDescriptor fileDescriptor =
FileDescriptor.buildFrom(fileDescriptorProto, new FileDescriptor[0]);
+ final Descriptor sampleDescriptor =
fileDescriptor.findMessageTypeByName(MESSAGE_NAME);
+ final DynamicMessage sample =
DynamicMessage.newBuilder(sampleDescriptor)
+ .setField(sampleDescriptor.findFieldByName(ENABLED_FIELD),
ENABLED_VALUE)
+ .setField(sampleDescriptor.findFieldByName(NAME_FIELD),
NAME_VALUE)
+ .build();
+ return sample.toByteString().newInput();
+ }
+ }
+
private SchemaDefinition createSchemaDefinition(final String name) {
return createSchemaDefinition(name, "test schema content", new
HashMap<>());
}
diff --git
a/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/resources/org/apache/nifi/protobuf/test/custom_options_on_all_elements.proto
b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/resources/org/apache/nifi/protobuf/test/custom_options_on_all_elements.proto
new file mode 100644
index 00000000000..f4ccc8c157c
--- /dev/null
+++
b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/resources/org/apache/nifi/protobuf/test/custom_options_on_all_elements.proto
@@ -0,0 +1,66 @@
+/*
+ * 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.
+ */
+syntax = "proto2";
+package example;
+
+option java_package = "com.example.comprehensive";
+option (custom.file_meta) = "docs";
+
+message Wrapper {
+ option (custom.message_meta) = true;
+
+ optional int32 flag = 1 [(custom.field_meta) = "f"];
+
+ oneof kind {
+ option (custom.oneof_meta) = true;
+ string text = 2 [(custom.oneof_field_meta) = "t"];
+ int32 number = 3;
+ }
+
+ extensions 100 to 199 [(custom.extension_range_meta) = true];
+
+ message Nested {
+ extensions 200 to 299;
+ }
+
+ extend Nested {
+ optional int32 nested_extra = 200 [(custom.nested_extend_field_meta) =
"n"];
+ }
+
+ enum Color {
+ option (custom.enum_meta) = true;
+ RED = 0 [(custom.enum_value_meta) = "r"];
+ GREEN = 1;
+ }
+}
+
+extend Wrapper {
+ optional int32 extra = 100 [(custom.extend_field_meta) = "e"];
+}
+
+message Request {
+}
+
+message Response {
+}
+
+service Catalog {
+ option (custom.service_meta) = true;
+ rpc Lookup (Request) returns (Response) {
+ option (custom.method_meta) = true;
+ }
+}
diff --git
a/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/resources/org/apache/nifi/protobuf/test/nested_wrapper_with_field_meta.proto
b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/resources/org/apache/nifi/protobuf/test/nested_wrapper_with_field_meta.proto
new file mode 100644
index 00000000000..c8bab2132f2
--- /dev/null
+++
b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/resources/org/apache/nifi/protobuf/test/nested_wrapper_with_field_meta.proto
@@ -0,0 +1,72 @@
+/*
+ * 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.
+ */
+syntax = "proto3";
+package example;
+
+message Wrapper {
+ optional Payload previous = 1;
+ optional Payload current = 2;
+ Origin origin = 3;
+ string operation = 4;
+ optional int64 event_time = 5;
+ optional chunk grouping = 6;
+
+ message Payload {
+ int64 record_id = 1;
+ string display_name = 2;
+ optional string code = 3;
+ int32 flag_a = 4 [(confluent.field_meta) = {
+ params: [
+ {
+ key: "connect.type",
+ value: "int16"
+ }
+ ]
+ }];
+ optional int32 flag_b = 5 [(confluent.field_meta) = {
+ params: [
+ {
+ key: "connect.type",
+ value: "int16"
+ }
+ ]
+ }];
+ int32 flag_c = 6 [(confluent.field_meta) = {
+ params: [
+ {
+ key: "connect.type",
+ value: "int16"
+ }
+ ]
+ }];
+ string first_seen = 7;
+ optional string last_editor = 8;
+ }
+ message Origin {
+ string schema_version = 1;
+ string producer = 2;
+ int64 origin_time = 3;
+ optional string mode = 4;
+ string catalog = 5;
+ optional string relation = 6;
+ }
+ message chunk {
+ string grouping_id = 1;
+ int64 sequence_a = 2;
+ int64 sequence_b = 3;
+ }
+}
