This is an automated email from the ASF dual-hosted git repository.
pvillard31 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 d48bd750ab3 NIFI-16427 Sanitize Protobuf custom options before message
name resolution (#11800)
d48bd750ab3 is described below
commit d48bd750ab3b017eb2ae8ff6390760a9b16d21ae
Author: Alaksiej Ščarbaty <[email protected]>
AuthorDate: Mon Oct 12 19:50:19 2026 +0200
NIFI-16427 Sanitize Protobuf custom options before message name resolution
(#11800)
---
.../services/protobuf/ProtobufSchemaCompiler.java | 5 +-
...java => ProtobufSchemaDefinitionSanitizer.java} | 52 ++-
.../services/protobuf/StandardProtobufReader.java | 10 +-
.../services/protobuf/StandardProtobufWriter.java | 7 +-
.../protobuf/TestProtobufSchemaCompiler.java | 261 ---------------
... => TestProtobufSchemaDefinitionSanitizer.java} | 360 ++++++---------------
...stStandardProtobufWriterConfluentRoundTrip.java | 150 ++++++---
7 files changed, 269 insertions(+), 576 deletions(-)
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 63423ce6d34..5db6af45a02 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
@@ -215,9 +215,8 @@ final class ProtobufSchemaCompiler {
Files.createDirectories(parentDirectory);
}
- final String preparedSchemaText =
ProtobufCustomOptionRemover.remove(schemaText, Location.get(relativePath),
logger);
- Files.write(schemaFile,
preparedSchemaText.getBytes(StandardCharsets.UTF_8), CREATE, WRITE,
TRUNCATE_EXISTING);
- logger.debug("Successfully wrote schema to file: {} (string length:
{})", schemaFile, preparedSchemaText.length());
+ Files.write(schemaFile, schemaText.getBytes(StandardCharsets.UTF_8),
CREATE, WRITE, TRUNCATE_EXISTING);
+ logger.debug("Successfully wrote schema to file: {} (string length:
{})", schemaFile, schemaText.length());
}
/**
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/ProtobufSchemaDefinitionSanitizer.java
similarity index 81%
rename from
nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/ProtobufCustomOptionRemover.java
rename to
nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/ProtobufSchemaDefinitionSanitizer.java
index e147d604662..51733448595 100644
---
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/ProtobufSchemaDefinitionSanitizer.java
@@ -16,6 +16,8 @@
*/
package org.apache.nifi.services.protobuf;
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
import com.squareup.wire.schema.Location;
import com.squareup.wire.schema.internal.parser.EnumConstantElement;
import com.squareup.wire.schema.internal.parser.EnumElement;
@@ -32,26 +34,58 @@ 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 org.apache.nifi.schemaregistry.services.SchemaDefinition;
+import org.apache.nifi.schemaregistry.services.StandardSchemaDefinition;
+import org.apache.nifi.serialization.record.SchemaIdentifier;
+import java.time.Duration;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
-final class ProtobufCustomOptionRemover {
+final class ProtobufSchemaDefinitionSanitizer {
- private ProtobufCustomOptionRemover() {
+ private static final Duration SANITIZED_SCHEMAS_CACHE_TTL =
Duration.ofHours(1);
+ private static final int SANITIZED_SCHEMAS_CACHE_SIZE = 200;
+
+ private final Cache<SchemaIdentifier, SchemaDefinition>
sanitizedSchemaCache;
+ private final ComponentLog logger;
+
+ ProtobufSchemaDefinitionSanitizer(final ComponentLog logger) {
+ this.logger = logger;
+ this.sanitizedSchemaCache = Caffeine.newBuilder()
+ .expireAfterAccess(SANITIZED_SCHEMAS_CACHE_TTL)
+ .maximumSize(SANITIZED_SCHEMAS_CACHE_SIZE)
+ .build();
}
/**
- * Removes custom (parenthesized) Protocol Buffer options from schema text.
+ * Removes custom (parenthesized) Protocol Buffer options from a schema
definition and its references.
* 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
+ * @param schemaDefinition original schema definition
+ * @return sanitized schema definition
*/
- static String remove(final String schemaText, final Location location,
final ComponentLog logger) {
+ SchemaDefinition sanitizeOrGetFromCache(final SchemaDefinition
schemaDefinition) {
+ return sanitizedSchemaCache.get(schemaDefinition.getIdentifier(),
+ identifier -> sanitize(schemaDefinition, identifier.toString()));
+ }
+
+ private SchemaDefinition sanitize(final SchemaDefinition schemaDefinition,
final String schemaLocation) {
+ final String sanitizedText =
removeCustomOptions(schemaDefinition.getText(), Location.get(schemaLocation));
+
+ final Map<String, SchemaDefinition> references =
schemaDefinition.getReferences();
+ final Map<String, SchemaDefinition> sanitizedReferences =
HashMap.newHashMap(references.size());
+ for (final Map.Entry<String, SchemaDefinition> reference :
references.entrySet()) {
+ final String referencePath = reference.getKey();
+ sanitizedReferences.put(referencePath,
sanitize(reference.getValue(), referencePath));
+ }
+
+ return new StandardSchemaDefinition(schemaDefinition.getIdentifier(),
sanitizedText, schemaDefinition.getSchemaType(), sanitizedReferences);
+ }
+
+ private String removeCustomOptions(final String schemaText, final Location
location) {
if (schemaText == null || schemaText.isBlank()) {
return schemaText;
}
diff --git
a/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/StandardProtobufReader.java
b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/StandardProtobufReader.java
index 225180f5de8..accad838042 100644
---
a/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/StandardProtobufReader.java
+++
b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/StandardProtobufReader.java
@@ -119,6 +119,7 @@ public class StandardProtobufReader extends
SchemaRegistryService implements Rec
private static final String PROTO_EXTENSION = ".proto";
private volatile ProtobufSchemaCompiler schemaCompiler;
+ private volatile ProtobufSchemaDefinitionSanitizer schemaSanitizer;
private volatile MessageNameResolver messageNameResolver;
private volatile SchemaReferenceReader schemaReferenceReader;
private volatile SchemaRegistry schemaRegistry;
@@ -163,6 +164,7 @@ public class StandardProtobufReader extends
SchemaRegistryService implements Rec
protected void init(final ControllerServiceInitializationContext config)
throws InitializationException {
super.init(config);
schemaCompiler = new ProtobufSchemaCompiler(getIdentifier(),
getLogger());
+ schemaSanitizer = new ProtobufSchemaDefinitionSanitizer(getLogger());
}
@Override
@@ -250,13 +252,15 @@ public class StandardProtobufReader extends
SchemaRegistryService implements Rec
private Schema validateSchemaCompiles(final SchemaIdentifier
schemaIdentifier, final String schemaTextValue) {
final SchemaDefinition schemaDefinition = new
StandardSchemaDefinition(schemaIdentifier, schemaTextValue,
SchemaDefinition.SchemaType.PROTOBUF);
- return schemaCompiler.compileOrGetFromCache(schemaDefinition);
+ final SchemaDefinition sanitizedSchemaDefinition =
schemaSanitizer.sanitizeOrGetFromCache(schemaDefinition);
+ return schemaCompiler.compileOrGetFromCache(sanitizedSchemaDefinition);
}
private RecordReader createProtobufRecordReader(final Map<String, String>
variables, final InputStream in, final SchemaDefinition schemaDefinition)
throws IOException {
- final Schema schema =
schemaCompiler.compileOrGetFromCache(schemaDefinition);
+ final SchemaDefinition sanitizedSchemaDefinition =
schemaSanitizer.sanitizeOrGetFromCache(schemaDefinition);
+ final Schema schema =
schemaCompiler.compileOrGetFromCache(sanitizedSchemaDefinition);
final ProtoSchemaParser schemaParser = new ProtoSchemaParser(schema);
- final MessageName messageName =
messageNameResolver.getMessageName(variables, schemaDefinition, in);
+ final MessageName messageName =
messageNameResolver.getMessageName(variables, sanitizedSchemaDefinition, in);
final RecordSchema recordSchema =
schemaParser.createSchema(messageName.getFullyQualifiedName());
return new ProtobufRecordReader(schema,
messageName.getFullyQualifiedName(), in, recordSchema);
}
diff --git
a/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/StandardProtobufWriter.java
b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/StandardProtobufWriter.java
index dc39351c6e5..7fe588d1e1c 100644
---
a/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/StandardProtobufWriter.java
+++
b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/StandardProtobufWriter.java
@@ -154,6 +154,7 @@ public class StandardProtobufWriter extends
SchemaRegistryService implements Rec
private static final String PROTO_EXTENSION = ".proto";
private volatile ProtobufSchemaCompiler schemaCompiler;
+ private volatile ProtobufSchemaDefinitionSanitizer schemaSanitizer;
private volatile MessageNameResolver messageNameResolver;
private volatile SchemaReferenceWriter schemaReferenceWriter;
private volatile MessageIndexWriter messageIndexWriter;
@@ -182,6 +183,7 @@ public class StandardProtobufWriter extends
SchemaRegistryService implements Rec
protected void init(final ControllerServiceInitializationContext config)
throws InitializationException {
super.init(config);
schemaCompiler = new ProtobufSchemaCompiler(getIdentifier(),
getLogger());
+ schemaSanitizer = new ProtobufSchemaDefinitionSanitizer(getLogger());
}
@Override
@@ -235,7 +237,7 @@ public class StandardProtobufWriter extends
SchemaRegistryService implements Rec
}
private ProtobufWriteContext createWriteContext(final Map<String, String>
variables, final RecordSchema suppliedSchema) throws SchemaNotFoundException,
IOException {
- final SchemaDefinition schemaDefinition =
createSchemaDefinition(variables, suppliedSchema);
+ final SchemaDefinition schemaDefinition =
schemaSanitizer.sanitizeOrGetFromCache(createSchemaDefinition(variables,
suppliedSchema));
final Schema schema =
schemaCompiler.compileOrGetFromCache(schemaDefinition);
final MessageName messageName =
messageNameResolver.getMessageName(variables, schemaDefinition,
InputStream.nullInputStream());
@@ -278,7 +280,8 @@ public class StandardProtobufWriter extends
SchemaRegistryService implements Rec
.name(sha256Hex(schemaTextValue) + PROTO_EXTENSION)
.build();
final SchemaDefinition schemaDefinition = new
StandardSchemaDefinition(schemaIdentifier, schemaTextValue,
SchemaDefinition.SchemaType.PROTOBUF);
- compiledSchema =
schemaCompiler.compileOrGetFromCache(schemaDefinition);
+ final SchemaDefinition sanitizedSchemaDefinition =
schemaSanitizer.sanitizeOrGetFromCache(schemaDefinition);
+ compiledSchema =
schemaCompiler.compileOrGetFromCache(sanitizedSchemaDefinition);
} catch (final RuntimeException e) {
// The compiler reports Wire schema errors as
SchemaCompilationException and other failures as other
// runtime exceptions, all of which indicate that the configured
schema text cannot be used
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
index e73b3aca656..5b26a44d16b 100644
---
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
@@ -16,40 +16,23 @@
*/
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 {
@@ -57,15 +40,11 @@ class TestProtobufSchemaCompiler {
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";
// Confluent Schema Registry keys each reference by the import path used
inside the .proto file, while the
// identifier of the referenced schema carries the registry subject, which
is not required to look like a file path
@@ -91,102 +70,6 @@ class TestProtobufSchemaCompiler {
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 = """
@@ -204,40 +87,6 @@ class TestProtobufSchemaCompiler {
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 = """
@@ -274,43 +123,6 @@ class TestProtobufSchemaCompiler {
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 = """
@@ -338,62 +150,6 @@ class TestProtobufSchemaCompiler {
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());
- }
-
@Test
void testCompileSchemaWithCrossFileImportFromRegistry() {
final SchemaDefinition referencedSchema = new StandardSchemaDefinition(
@@ -451,23 +207,6 @@ class TestProtobufSchemaCompiler {
assertNotNull(compiled.getType("airlines.ph.cdm.common.Audit"));
}
- 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());
}
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/TestProtobufSchemaDefinitionSanitizer.java
similarity index 51%
copy from
nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/java/org/apache/nifi/services/protobuf/TestProtobufSchemaCompiler.java
copy to
nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/java/org/apache/nifi/services/protobuf/TestProtobufSchemaDefinitionSanitizer.java
index e73b3aca656..7cbdb87f8e5 100644
---
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/TestProtobufSchemaDefinitionSanitizer.java
@@ -25,8 +25,8 @@ 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.Type;
import com.squareup.wire.schema.internal.parser.OptionElement;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.schemaregistry.services.SchemaDefinition;
@@ -48,12 +48,12 @@ 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.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
-class TestProtobufSchemaCompiler {
+class TestProtobufSchemaDefinitionSanitizer {
- private static final ComponentLog COMPONENT_LOG = new
MockComponentLog("test", TestProtobufSchemaCompiler.class);
+ private static final ComponentLog COMPONENT_LOG = new
MockComponentLog("test", TestProtobufSchemaDefinitionSanitizer.class);
private static final String PACKAGE = "example";
private static final String SAMPLE_TYPE = PACKAGE + ".Sample";
@@ -63,149 +63,13 @@ class TestProtobufSchemaCompiler {
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";
- // Confluent Schema Registry keys each reference by the import path used
inside the .proto file, while the
- // identifier of the referenced schema carries the registry subject, which
is not required to look like a file path
- private static final String ROOT_SUBJECT =
"airlines.ph.cdm.reservation.AirlineReservation";
- private static final String IMPORT_PATH = "airlines/ph/cdm/shared.proto";
- private static final String REFERENCE_SUBJECT = "airlines.ph.cdm.shared";
-
- private static final String ROOT_SCHEMA = """
- syntax = "proto3";
- package airlines.ph.cdm.reservation;
- import "airlines/ph/cdm/shared.proto";
- message AirlineReservation {
- string reservation_id = 1;
- airlines.ph.cdm.Status status = 2;
- }""";
-
- private static final String REFERENCED_SCHEMA = """
- syntax = "proto3";
- package airlines.ph.cdm;
- message Status {
- string code = 1;
- }""";
-
+ private final ProtobufSchemaDefinitionSanitizer sanitizer = new
ProtobufSchemaDefinitionSanitizer(COMPONENT_LOG);
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() {
+ void testSanitizeRemovesCustomOptionsFromSchemaAndReferences() {
final String referencedSchemaText = """
syntax = "proto3";
package example;
@@ -221,14 +85,22 @@ class TestProtobufSchemaCompiler {
import "common.proto";
message Root {
- Common item = 1 [(custom.field_meta) = "r"];
+ Common item = 1 [(confluent.field_meta) = { params: [ { key:
"k", value: "v" } ] }];
}
""";
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 SchemaDefinition sanitized =
sanitizer.sanitizeOrGetFromCache(rootSchema);
+
+ assertEquals(rootSchema.getIdentifier(), sanitized.getIdentifier());
+ assertEquals(rootSchema.getSchemaType(), sanitized.getSchemaType());
+ final SchemaDefinition sanitizedReference =
sanitized.getReferences().get(COMMON_PROTO);
+ assertNotNull(sanitizedReference);
+ assertEquals(referencedSchema.getIdentifier(),
sanitizedReference.getIdentifier());
+ assertEquals(referencedSchema.getSchemaType(),
sanitizedReference.getSchemaType());
+ final Schema schema = compiler.compileOrGetFromCache(sanitized);
final Field id = schema.getField(PACKAGE + ".Common", "id");
final Field item = schema.getField(PACKAGE + ".Root", "item");
@@ -239,111 +111,110 @@ class TestProtobufSchemaCompiler {
}
@Test
- void testCompileFailsWhenMessageTypeIsUnresolved() {
+ void testSanitizeKeepsSchemaTextWithoutCustomOptions() {
final String schemaText = """
syntax = "proto3";
package example;
- message Root {
- Common item = 1;
+ message Sample {
+ int32 value = 1;
}
""";
- final SchemaDefinition schemaDefinition =
schemaDefinition(SAMPLE_PROTO, schemaText);
- final RuntimeException exception =
assertThrows(RuntimeException.class, () ->
compiler.compileOrGetFromCache(schemaDefinition));
+ final SchemaDefinition sanitized =
sanitizer.sanitizeOrGetFromCache(schemaDefinition(SAMPLE_PROTO, schemaText));
- assertInstanceOf(SchemaException.class, exception.getCause());
+ assertEquals(schemaText, sanitized.getText());
}
@Test
- void testCompileFailsWhenTypeImportIsMissing() {
+ void testSanitizeReturnsCachedDefinitionForSameIdentifier() {
final String schemaText = """
syntax = "proto3";
package example;
- import "common.proto";
-
- message Root {
- Common item = 1;
+ message Sample {
+ int32 value = 1 [(custom.field_meta) = "v"];
}
""";
- final SchemaDefinition schemaDefinition =
schemaDefinition(SAMPLE_PROTO, schemaText);
- final RuntimeException exception =
assertThrows(RuntimeException.class, () ->
compiler.compileOrGetFromCache(schemaDefinition));
+ final SchemaDefinition first =
sanitizer.sanitizeOrGetFromCache(schemaDefinition(SAMPLE_PROTO, schemaText));
+ final SchemaDefinition second =
sanitizer.sanitizeOrGetFromCache(schemaDefinition(SAMPLE_PROTO, schemaText));
- assertInstanceOf(SchemaException.class, exception.getCause());
+ assertSame(first, second);
}
@Test
- void testCompileNestedMessageAndEnumWithCustomOptions() {
+ void testSanitizeSchemaWithCustomOptionWithoutImport() {
final String schemaText = """
syntax = "proto3";
package example;
- message Outer {
- option (custom.message_meta) = true;
+ message Sample {
+ optional int32 enabled = 1 [(confluent.field_meta) = {
+ params: [
+ {
+ key: "connect.type",
+ value: "int16"
+ }
+ ]
+ }];
+ }
+ """;
- message Inner {
- int32 value = 1 [(custom.field_meta) = "y"];
- }
+ final Schema schema = sanitizeAndCompile(schemaText);
+ final Field enabled = schema.getField(SAMPLE_TYPE, "enabled");
- Inner inner = 1;
- }
+ assertNotNull(enabled);
+ assertEquals(1, enabled.getTag());
+ assertNoParenthesizedOptions(enabled.getOptions());
+ }
- enum Color {
- option (custom.enum_meta) = true;
- RED = 0 [(custom.enum_value_meta) = "r"];
- GREEN = 1;
- }
- """;
- final SchemaDefinition schemaDefinition =
schemaDefinition(SAMPLE_PROTO, schemaText);
+ @Test
+ void testSanitizeNestedSchemaWithRepeatedFieldMetaOptions() throws
IOException {
+ final String schemaText =
readTestResource("/org/apache/nifi/protobuf/test/nested_wrapper_with_field_meta.proto");
- 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);
+ final Schema schema = sanitizeAndCompile(schemaText);
+ 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(outer.field("inner"));
- assertNotNull(schema.getField(PACKAGE + ".Outer.Inner", FIELD_VALUE));
- assertNotNull(red);
- assertNotNull(color.constant("GREEN"));
- assertNoParenthesizedOptions(color.getOptions());
- assertNoParenthesizedOptions(red.getOptions());
+ 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 testPayloadExtendDeclarationsArePreserved() {
+ void testSanitizeKeepsPackedWhenCustomFieldOptionPresent() {
final String schemaText = """
syntax = "proto2";
package example;
- message Base {
- extensions 100 to 199;
- optional string name = 1;
- }
-
- extend Base {
- optional int32 extra = 100;
+ 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 MessageType base = assertInstanceOf(MessageType.class,
schema.getType(PACKAGE + ".Base"));
- final ProtoFile protoFile = schema.protoFile(SAMPLE_PROTO);
+ final Schema schema = sanitizeAndCompile(schemaText);
+ final Field values = schema.getField(SAMPLE_TYPE, "values");
- assertNotNull(base.field("name"));
- assertNotNull(protoFile);
- assertEquals(1, protoFile.getExtendList().size());
- assertEquals(FIELD_EXTRA,
protoFile.getExtendList().get(0).getFields().get(0).getName());
+ assertNotNull(values);
+ assertTrue(values.isRepeated());
+ assertTrue(values.isPacked());
+ assertNoParenthesizedOptions(values.getOptions());
}
@Test
- void testCompileRemovesCustomOptionsFromAllSyntaxElements() throws
IOException {
+ void testSanitizeRemovesCustomOptionsFromAllSyntaxElements() 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 Schema schema = sanitizeAndCompile(schemaText);
final ProtoFile protoFile = schema.protoFile(SAMPLE_PROTO);
assertNotNull(protoFile);
@@ -367,7 +238,7 @@ class TestProtobufSchemaCompiler {
assertEquals(1, protoFile.getExtendList().size());
final Field extra =
protoFile.getExtendList().get(0).getFields().get(0);
- assertEquals(FIELD_EXTRA, extra.getName());
+ assertEquals("extra", extra.getName());
assertNoParenthesizedOptions(extra.getOptions());
assertNotNull(schema.getType(WRAPPER_TYPE + ".Nested"));
@@ -395,60 +266,33 @@ class TestProtobufSchemaCompiler {
}
@Test
- void testCompileSchemaWithCrossFileImportFromRegistry() {
- final SchemaDefinition referencedSchema = new StandardSchemaDefinition(
-
SchemaIdentifier.builder().name(REFERENCE_SUBJECT).id(4L).version(1).build(),
- REFERENCED_SCHEMA,
- SchemaDefinition.SchemaType.PROTOBUF);
-
- final SchemaDefinition rootSchema = new StandardSchemaDefinition(
-
SchemaIdentifier.builder().name(ROOT_SUBJECT).id(3L).version(2).build(),
- ROOT_SCHEMA,
- SchemaDefinition.SchemaType.PROTOBUF,
- Map.of(IMPORT_PATH, referencedSchema));
-
- final Schema compiled = compiler.compileOrGetFromCache(rootSchema);
-
-
assertNotNull(compiled.getType("airlines.ph.cdm.reservation.AirlineReservation"));
- assertNotNull(compiled.getType("airlines.ph.cdm.Status"));
+ void testSanitizePreservesTopLevelMessageOrder() {
+ final String schemaText = """
+ syntax = "proto3";
+ package example;
+
+ message A {
+ int32 value = 1 [(custom.field_meta) = "a"];
+ }
+
+ message B {
+ int32 value = 1 [(custom.field_meta) = "b"];
+ }
+
+ message C {
+ int32 value = 1 [(custom.field_meta) = "c"];
+ }
+ """;
+
+ final Schema schema = sanitizeAndCompile(schemaText);
+ final ProtoFile protoFile = schema.protoFile(SAMPLE_PROTO);
+
+ assertNotNull(protoFile);
+ assertEquals(List.of("A", "B", "C"),
protoFile.getTypes().stream().map(Type::getName).toList());
}
- @Test
- void testCompileSchemaWithNestedCrossFileImports() {
- final SchemaDefinition leafSchema = new StandardSchemaDefinition(
-
SchemaIdentifier.builder().name("airlines.ph.cdm.common").id(5L).version(1).build(),
- """
- syntax = "proto3";
- package airlines.ph.cdm.common;
- message Audit {
- string created_by = 1;
- }""",
- SchemaDefinition.SchemaType.PROTOBUF);
-
- final SchemaDefinition referencedSchema = new StandardSchemaDefinition(
-
SchemaIdentifier.builder().name(REFERENCE_SUBJECT).id(4L).version(1).build(),
- """
- syntax = "proto3";
- package airlines.ph.cdm;
- import "airlines/ph/cdm/common/audit.proto";
- message Status {
- string code = 1;
- airlines.ph.cdm.common.Audit audit = 2;
- }""",
- SchemaDefinition.SchemaType.PROTOBUF,
- Map.of("airlines/ph/cdm/common/audit.proto", leafSchema));
-
- final SchemaDefinition rootSchema = new StandardSchemaDefinition(
-
SchemaIdentifier.builder().name(ROOT_SUBJECT).id(3L).version(2).build(),
- ROOT_SCHEMA,
- SchemaDefinition.SchemaType.PROTOBUF,
- Map.of(IMPORT_PATH, referencedSchema));
-
- final Schema compiled = compiler.compileOrGetFromCache(rootSchema);
-
-
assertNotNull(compiled.getType("airlines.ph.cdm.reservation.AirlineReservation"));
- assertNotNull(compiled.getType("airlines.ph.cdm.Status"));
- assertNotNull(compiled.getType("airlines.ph.cdm.common.Audit"));
+ private Schema sanitizeAndCompile(final String schemaText) {
+ return
compiler.compileOrGetFromCache(sanitizer.sanitizeOrGetFromCache(schemaDefinition(SAMPLE_PROTO,
schemaText)));
}
private void assertNoParenthesizedOptions(final Options options) {
@@ -473,10 +317,6 @@ class TestProtobufSchemaCompiler {
}
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);
+ 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/TestStandardProtobufWriterConfluentRoundTrip.java
b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/java/org/apache/nifi/services/protobuf/TestStandardProtobufWriterConfluentRoundTrip.java
index dd2d5d6b13e..92a2747690f 100644
---
a/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/java/org/apache/nifi/services/protobuf/TestStandardProtobufWriterConfluentRoundTrip.java
+++
b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/java/org/apache/nifi/services/protobuf/TestStandardProtobufWriterConfluentRoundTrip.java
@@ -73,9 +73,16 @@ import static org.junit.jupiter.api.Assertions.assertNull;
*/
class TestStandardProtobufWriterConfluentRoundTrip {
- private static final int SCHEMA_ID = 42;
+ private static final long SCHEMA_ID = 42;
// Confluent header: magic byte 0x00 followed by the big-endian schema id
(42).
private static final byte[] EXPECTED_HEADER = {0x00, 0x00, 0x00, 0x00,
0x2A};
+ private static final int HEADER_LENGTH = EXPECTED_HEADER.length;
+
+ private static final String REGISTRY_ID = "registry";
+ private static final String REFERENCE_WRITER_ID = "referenceWriter";
+ private static final String INDEX_WRITER_ID = "indexWriter";
+ private static final String REFERENCE_READER_ID = "referenceReader";
+ private static final String MESSAGE_NAME_RESOLVER_ID =
"messageNameResolver";
private static final String SCHEMA_TEXT_VALUE = """
syntax = "proto3";
@@ -90,6 +97,44 @@ class TestStandardProtobufWriterConfluentRoundTrip {
string name = 1;
}""";
+ private static final String CUSTOM_OPTIONS_SCHEMA_TEXT_VALUE = """
+ syntax = "proto3";
+ message User {
+ int32 id = 1 [(confluent.field_meta) = {
+ params: [
+ {
+ key: "connect.type",
+ value: "int16"
+ }
+ ]
+ }];
+ string name = 2 [(confluent.field_meta) = { doc: "name" }];
+ message Profile {
+ string bio = 1 [(confluent.field_meta) = { params: [ { key: "k",
value: "v" } ] }];
+ }
+ }
+ message Company {
+ string name = 1 [(confluent.field_meta) = { doc: "company" }];
+ }""";
+
+ private static final String USER_MESSAGE = "User";
+ private static final String PROFILE_MESSAGE = "User.Profile";
+ private static final String COMPANY_MESSAGE = "Company";
+
+ // Nested message User.Profile -> declaration path [0, 0] -> length 2 then
indexes 0, 0 (zigzag varints).
+ private static final byte[] NESTED_PROFILE_MESSAGE_INDEX = {0x04, 0x00,
0x00};
+ // Root message Company -> declaration path [1] -> length 1 then index 1
(zigzag varints).
+ private static final byte[] COMPANY_MESSAGE_INDEX = {0x02, 0x02};
+
+ private static final String ID_FIELD = "id";
+ private static final String NAME_FIELD = "name";
+ private static final String BIO_FIELD = "bio";
+
+ private static final int ID_VALUE = 7;
+ private static final String USER_NAME_VALUE = "Alice";
+ private static final String COMPANY_NAME_VALUE = "Acme";
+ private static final String BIO_VALUE = "hello";
+
private TestRunner runner;
private MockSchemaRegistry schemaRegistry;
private ConfluentEncodedSchemaReferenceWriter referenceWriter;
@@ -107,45 +152,72 @@ class TestStandardProtobufWriterConfluentRoundTrip {
referenceReader = new ConfluentEncodedSchemaReferenceReader();
messageNameResolver = new ConfluentProtobufMessageNameResolver();
- enableService("registry", schemaRegistry);
- enableService("referenceWriter", referenceWriter);
- enableService("indexWriter", indexWriter);
- enableService("referenceReader", referenceReader);
- enableService("messageNameResolver", messageNameResolver);
+ enableService(REGISTRY_ID, schemaRegistry);
+ enableService(REFERENCE_WRITER_ID, referenceWriter);
+ enableService(INDEX_WRITER_ID, indexWriter);
+ enableService(REFERENCE_READER_ID, referenceReader);
+ enableService(MESSAGE_NAME_RESOLVER_ID, messageNameResolver);
}
@Test
void testFirstRootMessageSingleByteIndexAndRoundTrip() throws Exception {
final Map<String, Object> values = new HashMap<>();
- values.put("id", 7);
- values.put("name", "Alice");
- final MapRecord userRecord = new MapRecord(recordSchema("id", "name"),
values);
+ values.put(ID_FIELD, ID_VALUE);
+ values.put(NAME_FIELD, USER_NAME_VALUE);
+ final MapRecord userRecord = new MapRecord(recordSchema(ID_FIELD,
NAME_FIELD), values);
- final byte[] output = writeConfluent("User", userRecord);
+ final byte[] output = writeConfluent(USER_MESSAGE, userRecord);
// [header][message index][payload]; the first root message collapses
to the single index byte 0x00.
- assertArrayEquals(EXPECTED_HEADER, slice(output, 0, 5));
- assertEquals(0x00, output[5]);
+ assertArrayEquals(EXPECTED_HEADER, slice(output, 0, HEADER_LENGTH));
+ assertEquals(0x00, output[HEADER_LENGTH]);
final Record readBack = readConfluent(output);
- assertEquals(7, readBack.getValue("id"));
- assertEquals("Alice", readBack.getValue("name"));
+ assertEquals(ID_VALUE, readBack.getValue(ID_FIELD));
+ assertEquals(USER_NAME_VALUE, readBack.getValue(NAME_FIELD));
}
@Test
void testNestedMessageIndexAndRoundTrip() throws Exception {
final Map<String, Object> values = new HashMap<>();
- values.put("bio", "hello");
- final MapRecord profileRecord = new MapRecord(recordSchema("bio"),
values);
+ values.put(BIO_FIELD, BIO_VALUE);
+ final MapRecord profileRecord = new MapRecord(recordSchema(BIO_FIELD),
values);
- final byte[] output = writeConfluent("User.Profile", profileRecord);
+ final byte[] output = writeConfluent(PROFILE_MESSAGE, profileRecord);
- // Nested message User.Profile -> declaration path [0, 0] -> length 2
then indexes 0, 0 (zigzag varints).
- assertArrayEquals(EXPECTED_HEADER, slice(output, 0, 5));
- assertArrayEquals(new byte[] {0x04, 0x00, 0x00}, slice(output, 5, 8));
+ assertArrayEquals(EXPECTED_HEADER, slice(output, 0, HEADER_LENGTH));
+ assertArrayEquals(NESTED_PROFILE_MESSAGE_INDEX, slice(output,
HEADER_LENGTH, NESTED_PROFILE_MESSAGE_INDEX.length));
final Record readBack = readConfluent(output);
- assertEquals("hello", readBack.getValue("bio"));
+ assertEquals(BIO_VALUE, readBack.getValue(BIO_FIELD));
+ }
+
+ @Test
+ void testCustomOptionsNestedMessageRoundTrip() throws Exception {
+ schemaRegistry.returnSchemaText(CUSTOM_OPTIONS_SCHEMA_TEXT_VALUE);
+ final MapRecord profileRecord = new MapRecord(recordSchema(BIO_FIELD),
Map.of(BIO_FIELD, BIO_VALUE));
+
+ final byte[] output = writeConfluent(PROFILE_MESSAGE, profileRecord);
+
+ assertArrayEquals(EXPECTED_HEADER, slice(output, 0, HEADER_LENGTH));
+ assertArrayEquals(NESTED_PROFILE_MESSAGE_INDEX, slice(output,
HEADER_LENGTH, NESTED_PROFILE_MESSAGE_INDEX.length));
+
+ final Record readBack = readConfluent(output);
+ assertEquals(BIO_VALUE, readBack.getValue(BIO_FIELD));
+ }
+
+ @Test
+ void testCustomOptionsSecondRootMessageRoundTrip() throws Exception {
+ schemaRegistry.returnSchemaText(CUSTOM_OPTIONS_SCHEMA_TEXT_VALUE);
+ final MapRecord companyRecord = new
MapRecord(recordSchema(NAME_FIELD), Map.of(NAME_FIELD, COMPANY_NAME_VALUE));
+
+ final byte[] output = writeConfluent(COMPANY_MESSAGE, companyRecord);
+
+ assertArrayEquals(EXPECTED_HEADER, slice(output, 0, HEADER_LENGTH));
+ assertArrayEquals(COMPANY_MESSAGE_INDEX, slice(output, HEADER_LENGTH,
COMPANY_MESSAGE_INDEX.length));
+
+ final Record readBack = readConfluent(output);
+ assertEquals(COMPANY_NAME_VALUE, readBack.getValue(NAME_FIELD));
}
@Test
@@ -154,11 +226,11 @@ class TestStandardProtobufWriterConfluentRoundTrip {
runner.addControllerService("schemaTextWriter", writer);
runner.setProperty(writer, SCHEMA_ACCESS_STRATEGY,
SCHEMA_TEXT_PROPERTY.getValue());
runner.setProperty(writer, SCHEMA_TEXT, SCHEMA_TEXT_VALUE);
- runner.setProperty(writer, StandardProtobufWriter.MESSAGE_NAME,
"User");
+ runner.setProperty(writer, StandardProtobufWriter.MESSAGE_NAME,
USER_MESSAGE);
runner.assertValid(writer);
// Inline schema text has no registry identifier or version for the
Confluent header
- runner.setProperty(writer,
StandardProtobufWriter.SCHEMA_REFERENCE_WRITER, "referenceWriter");
+ runner.setProperty(writer,
StandardProtobufWriter.SCHEMA_REFERENCE_WRITER, REFERENCE_WRITER_ID);
runner.assertNotValid(writer);
}
@@ -166,13 +238,13 @@ class TestStandardProtobufWriterConfluentRoundTrip {
final StandardProtobufWriter writer = new StandardProtobufWriter();
runner.addControllerService("writer-" + messageName, writer);
runner.setProperty(writer, SCHEMA_ACCESS_STRATEGY,
SCHEMA_NAME_PROPERTY.getValue());
- runner.setProperty(writer, SCHEMA_REGISTRY, "registry");
+ runner.setProperty(writer, SCHEMA_REGISTRY, REGISTRY_ID);
runner.setProperty(writer, SCHEMA_NAME, "user");
runner.setProperty(writer,
StandardProtobufWriter.MESSAGE_NAME_RESOLUTION_STRATEGY,
StandardProtobufWriter.MessageNameResolverStrategy.MESSAGE_NAME_PROPERTY.getValue());
runner.setProperty(writer, StandardProtobufWriter.MESSAGE_NAME,
messageName);
- runner.setProperty(writer,
StandardProtobufWriter.SCHEMA_REFERENCE_WRITER, "referenceWriter");
- runner.setProperty(writer,
StandardProtobufWriter.MESSAGE_INDEX_WRITER, "indexWriter");
+ runner.setProperty(writer,
StandardProtobufWriter.SCHEMA_REFERENCE_WRITER, REFERENCE_WRITER_ID);
+ runner.setProperty(writer,
StandardProtobufWriter.MESSAGE_INDEX_WRITER, INDEX_WRITER_ID);
runner.enableControllerService(writer);
final RecordSchema writeSchema = writer.getSchema(emptyMap(), null);
@@ -190,11 +262,11 @@ class TestStandardProtobufWriterConfluentRoundTrip {
final StandardProtobufReader reader = new StandardProtobufReader();
runner.addControllerService("reader", reader);
runner.setProperty(reader, SCHEMA_ACCESS_STRATEGY,
SCHEMA_REFERENCE_READER_PROPERTY.getValue());
- runner.setProperty(reader, SCHEMA_REFERENCE_READER, "referenceReader");
- runner.setProperty(reader, SCHEMA_REGISTRY, "registry");
+ runner.setProperty(reader, SCHEMA_REFERENCE_READER,
REFERENCE_READER_ID);
+ runner.setProperty(reader, SCHEMA_REGISTRY, REGISTRY_ID);
runner.setProperty(reader,
StandardProtobufReader.MESSAGE_NAME_RESOLUTION_STRATEGY,
StandardProtobufReader.MessageNameResolverStrategy.MESSAGE_NAME_RESOLVER.getValue());
- runner.setProperty(reader,
StandardProtobufReader.MESSAGE_NAME_RESOLVER, "messageNameResolver");
+ runner.setProperty(reader,
StandardProtobufReader.MESSAGE_NAME_RESOLVER, MESSAGE_NAME_RESOLVER_ID);
runner.enableControllerService(reader);
final RecordReader recordReader =
reader.createRecordReader(emptyMap(), new ByteArrayInputStream(data),
data.length, runner.getLogger());
@@ -206,7 +278,7 @@ class TestStandardProtobufWriterConfluentRoundTrip {
private RecordSchema recordSchema(final String... fieldNames) {
final RecordField[] fields = new RecordField[fieldNames.length];
for (int i = 0; i < fieldNames.length; i++) {
- final RecordFieldType type = "id".equals(fieldNames[i]) ?
RecordFieldType.INT : RecordFieldType.STRING;
+ final RecordFieldType type = ID_FIELD.equals(fieldNames[i]) ?
RecordFieldType.INT : RecordFieldType.STRING;
fields[i] = new RecordField(fieldNames[i], type.getDataType());
}
return new SimpleRecordSchema(List.of(fields));
@@ -217,9 +289,9 @@ class TestStandardProtobufWriterConfluentRoundTrip {
runner.enableControllerService(service);
}
- private byte[] slice(final byte[] data, final int from, final int to) {
- final byte[] out = new byte[to - from];
- System.arraycopy(data, from, out, 0, to - from);
+ private byte[] slice(final byte[] data, final int from, final int length) {
+ final byte[] out = new byte[length];
+ System.arraycopy(data, from, out, 0, length);
return out;
}
@@ -228,10 +300,11 @@ class TestStandardProtobufWriterConfluentRoundTrip {
* both the writer's name-based lookup and the reader's
Confluent-header-based lookup resolve the same schema.
*/
static class MockSchemaRegistry extends AbstractControllerService
implements SchemaRegistry {
- private final SchemaDefinition schemaDefinition = new
StandardSchemaDefinition(
- SchemaIdentifier.builder().name("user.proto").id((long)
SCHEMA_ID).version(1).build(),
- SCHEMA_TEXT_VALUE,
- SchemaDefinition.SchemaType.PROTOBUF);
+ private volatile String schemaText = SCHEMA_TEXT_VALUE;
+
+ void returnSchemaText(final String schemaText) {
+ this.schemaText = schemaText;
+ }
@Override
public RecordSchema retrieveSchema(final SchemaIdentifier
schemaIdentifier) {
@@ -240,7 +313,8 @@ class TestStandardProtobufWriterConfluentRoundTrip {
@Override
public SchemaDefinition retrieveSchemaDefinition(final
SchemaIdentifier schemaIdentifier) throws SchemaNotFoundException {
- return schemaDefinition;
+ final SchemaIdentifier identifier =
SchemaIdentifier.builder().name("user.proto").id(SCHEMA_ID).version(1).build();
+ return new StandardSchemaDefinition(identifier, schemaText,
SchemaDefinition.SchemaType.PROTOBUF);
}
@Override
