rootMessages = parser.parse(schemaText);
+ return ProtobufMessageIndexEncoder.encode(rootMessages, encodeMessageIndexArguments.messageName());
+ } catch (final Exception e) {
+ throw new IllegalStateException("Failed to parse protobuf schema", e);
+ }
+ }
+
+ private record EncodeMessageIndexArguments(SchemaDefinition schemaDefinition, MessageName messageName) {
+ }
+}
diff --git a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-index-writer/src/main/java/org/apache/nifi/confluent/schemaregistry/ProtobufMessageIndexEncoder.java b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-index-writer/src/main/java/org/apache/nifi/confluent/schemaregistry/ProtobufMessageIndexEncoder.java
new file mode 100644
index 000000000000..1f29ecdb6c53
--- /dev/null
+++ b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-index-writer/src/main/java/org/apache/nifi/confluent/schemaregistry/ProtobufMessageIndexEncoder.java
@@ -0,0 +1,113 @@
+/*
+ * 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.confluent.schemaregistry;
+
+import org.apache.nifi.confluent.schema.ProtobufMessageSchema;
+import org.apache.nifi.confluent.schema.VarintUtils;
+import org.apache.nifi.schemaregistry.services.MessageName;
+
+import java.io.ByteArrayOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+import static java.lang.String.format;
+
+/**
+ * Computes the Confluent wire format message index path for a target message within a parsed
+ * Protobuf schema, and encodes it to bytes. This is the inverse of the message index decoding
+ * performed by {@code ConfluentProtobufMessageNameResolver}: given a fully qualified message
+ * name, it locates the path of declaration-order indexes leading to that message and encodes it
+ * as zigzag varints, applying the single-byte {@code 0x00} optimization for the common case of
+ * the first root message.
+ *
+ * See the Confluent protobuf wire format.
+ *
+ * This class has no dependency on the NiFi framework and can be exercised directly with plain
+ * unit tests.
+ */
+final class ProtobufMessageIndexEncoder {
+
+ private static final byte[] FIRST_ROOT_MESSAGE_INDEX = {0x00};
+
+ private ProtobufMessageIndexEncoder() {
+ }
+
+ /**
+ * Encodes the message index path for the given message name within the given schema.
+ *
+ * @param rootMessages the root messages of the parsed Protobuf schema, in declaration order
+ * @param messageName the target message name to locate
+ * @return the encoded message index bytes
+ * @throws IllegalStateException if the message name cannot be located within the schema
+ */
+ static byte[] encode(final List rootMessages, final MessageName messageName) {
+ final List messageIndexPath = findMessageIndexPath(rootMessages, messageName);
+
+ if (messageIndexPath.size() == 1 && messageIndexPath.getFirst() == 0) {
+ return FIRST_ROOT_MESSAGE_INDEX;
+ }
+
+ final ByteArrayOutputStream output = new ByteArrayOutputStream();
+ output.writeBytes(VarintUtils.writeZigZagVarint(messageIndexPath.size()));
+ for (final int index : messageIndexPath) {
+ output.writeBytes(VarintUtils.writeZigZagVarint(index));
+ }
+ return output.toByteArray();
+ }
+
+ private static List findMessageIndexPath(final List rootMessages, final MessageName messageName) {
+ // Match on the fully qualified name across the message tree rather than on the namespace/name split, so the
+ // encoder is robust to how the MessageName was constructed (a statically configured "package.Outer.Nested"
+ // and a resolver-produced name both reduce to the same fully qualified name).
+ final String targetFullyQualifiedName = messageName.getFullyQualifiedName();
+
+ for (int rootIndex = 0; rootIndex < rootMessages.size(); rootIndex++) {
+ final ProtobufMessageSchema rootMessage = rootMessages.get(rootIndex);
+ final String rootFullyQualifiedName = rootMessage.getPackageName()
+ .map(packageName -> packageName + "." + rootMessage.getName())
+ .orElseGet(rootMessage::getName);
+
+ final List messageIndexPath = descend(rootMessage, rootFullyQualifiedName, targetFullyQualifiedName);
+ if (messageIndexPath != null) {
+ messageIndexPath.addFirst(rootIndex);
+ return messageIndexPath;
+ }
+ }
+
+ throw new IllegalStateException(format("Message not found in schema definition: %s", targetFullyQualifiedName));
+ }
+
+ private static List descend(final ProtobufMessageSchema currentMessage, final String currentFullyQualifiedName, final String targetFullyQualifiedName) {
+ if (currentFullyQualifiedName.equals(targetFullyQualifiedName)) {
+ return new ArrayList<>();
+ }
+
+ final List children = currentMessage.getChildMessageSchemas();
+ for (int childIndex = 0; childIndex < children.size(); childIndex++) {
+ final ProtobufMessageSchema child = children.get(childIndex);
+ final String childFullyQualifiedName = currentFullyQualifiedName + "." + child.getName();
+
+ final List messageIndexPath = descend(child, childFullyQualifiedName, targetFullyQualifiedName);
+ if (messageIndexPath != null) {
+ messageIndexPath.addFirst(childIndex);
+ return messageIndexPath;
+ }
+ }
+
+ return null;
+ }
+}
diff --git a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-index-writer/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-index-writer/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService
new file mode 100644
index 000000000000..7c9bba672984
--- /dev/null
+++ b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-index-writer/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService
@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.nifi.confluent.schemaregistry.ConfluentProtobufMessageIndexWriter
diff --git a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-index-writer/src/test/java/org/apache/nifi/confluent/schemaregistry/ConfluentProtobufMessageIndexWriterTest.java b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-index-writer/src/test/java/org/apache/nifi/confluent/schemaregistry/ConfluentProtobufMessageIndexWriterTest.java
new file mode 100644
index 000000000000..d8de2699ca32
--- /dev/null
+++ b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-index-writer/src/test/java/org/apache/nifi/confluent/schemaregistry/ConfluentProtobufMessageIndexWriterTest.java
@@ -0,0 +1,111 @@
+/*
+ * 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.confluent.schemaregistry;
+
+import org.apache.nifi.schemaregistry.services.MessageName;
+import org.apache.nifi.schemaregistry.services.SchemaDefinition;
+import org.apache.nifi.schemaregistry.services.StandardMessageName;
+import org.apache.nifi.schemaregistry.services.StandardSchemaDefinition;
+import org.apache.nifi.serialization.record.SchemaIdentifier;
+import org.apache.nifi.util.NoOpProcessor;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Stream;
+
+import static org.apache.nifi.schemaregistry.services.SchemaDefinition.SchemaType.PROTOBUF;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class ConfluentProtobufMessageIndexWriterTest {
+
+ // Schema without package (default package)
+ private static final String DEFAULT_PACKAGE_SCHEMA = """
+ syntax = "proto3";
+ message User {
+ int32 id = 1;
+ string name = 2;
+ Address address = 3;
+ message Profile {
+ string bio = 1;
+ }
+ }
+ message Company {
+ string name = 1;
+ }
+ message Address {
+ string street = 1;
+ }""";
+
+ private static final SchemaIdentifier ID = SchemaIdentifier.builder()
+ .id(1L)
+ .build();
+
+ private static final SchemaDefinition SCHEMA_WITH_DEFAULT_PACKAGE = new StandardSchemaDefinition(
+ ID,
+ DEFAULT_PACKAGE_SCHEMA,
+ PROTOBUF,
+ Map.of());
+
+ private ConfluentProtobufMessageIndexWriter writer;
+
+ private static Stream provideMessageIndexTestCases() {
+ return Stream.of(
+ // the format is [target message name], [expected wire-format bytes for the message index]
+ Arguments.of(new StandardMessageName(Optional.empty(), "User"), new byte[] {0x00}),
+ Arguments.of(new StandardMessageName(Optional.empty(), "Company"), new byte[] {0x02, 0x02}),
+ Arguments.of(new StandardMessageName(Optional.empty(), "Address"), new byte[] {0x02, 0x04}),
+ Arguments.of(new StandardMessageName(Optional.empty(), "User.Profile"), new byte[] {0x04, 0x00, 0x00})
+ );
+ }
+
+ @BeforeEach
+ void setUp() throws Exception {
+ writer = new ConfluentProtobufMessageIndexWriter();
+ final TestRunner testRunner = TestRunners.newTestRunner(NoOpProcessor.class);
+ testRunner.addControllerService("messageIndexWriter", writer);
+ testRunner.enableControllerService(writer);
+ }
+
+ @ParameterizedTest
+ @MethodSource("provideMessageIndexTestCases")
+ void testWriteMessageIndex(final MessageName messageName, final byte[] expectedBytes) throws IOException {
+ final ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+
+ writer.writeMessageIndex(Map.of(), SCHEMA_WITH_DEFAULT_PACKAGE, messageName, outputStream);
+
+ assertArrayEquals(expectedBytes, outputStream.toByteArray());
+ }
+
+ @Test
+ void testWriteMessageIndexUnknownMessageThrows() {
+ final ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ final MessageName unknownMessageName = new StandardMessageName(Optional.empty(), "DoesNotExist");
+
+ assertThrows(IllegalStateException.class,
+ () -> writer.writeMessageIndex(Map.of(), SCHEMA_WITH_DEFAULT_PACKAGE, unknownMessageName, outputStream));
+ }
+}
diff --git a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-index-writer/src/test/java/org/apache/nifi/confluent/schemaregistry/ProtobufMessageIndexEncoderTest.java b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-index-writer/src/test/java/org/apache/nifi/confluent/schemaregistry/ProtobufMessageIndexEncoderTest.java
new file mode 100644
index 000000000000..12ddde4e9eee
--- /dev/null
+++ b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-index-writer/src/test/java/org/apache/nifi/confluent/schemaregistry/ProtobufMessageIndexEncoderTest.java
@@ -0,0 +1,195 @@
+/*
+ * 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.confluent.schemaregistry;
+
+import org.apache.nifi.confluent.schema.AntlrProtobufMessageSchemaParser;
+import org.apache.nifi.confluent.schema.ProtobufMessageSchema;
+import org.apache.nifi.confluent.schema.VarintUtils;
+import org.apache.nifi.schemaregistry.services.MessageName;
+import org.apache.nifi.schemaregistry.services.StandardMessageName;
+import org.apache.nifi.schemaregistry.services.StandardMessageNameFactory;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/**
+ * This class contains no NiFi framework dependencies: it exercises {@link ProtobufMessageIndexEncoder}
+ * directly, without a TestRunner or any controller-service context.
+ */
+class ProtobufMessageIndexEncoderTest {
+
+ // Schema without package (default package)
+ private static final String DEFAULT_PACKAGE_SCHEMA = """
+ syntax = "proto3";
+ message User {
+ int32 id = 1;
+ string name = 2;
+ Address address = 3;
+ message Profile {
+ string bio = 1;
+ Settings settings = 2;
+ message Settings {
+ bool notifications = 1;
+ string theme = 2;
+ }
+ }
+ }
+ message Company {
+ string name = 1;
+ Address address = 2;
+ }
+ message Address {
+ string street = 1;
+ string city = 2;
+ }""";
+
+ // Schema with explicit package
+ private static final String EXPLICIT_PACKAGE_SCHEMA = """
+ syntax = "proto3";
+ package com.example.proto;
+ message User {
+ int32 id = 1;
+ string name = 2;
+ Address address = 3;
+ message Profile {
+ string bio = 1;
+ Settings settings = 2;
+ message Settings {
+ bool notifications = 1;
+ string theme = 2;
+ }
+ }
+ }
+ message Company {
+ string name = 1;
+ Address address = 2;
+ }
+ message Address {
+ string street = 1;
+ string city = 2;
+ }""";
+
+ private static Stream provideMessageIndexTestCases() {
+ return Stream.of(
+ // the format is [input schema], [target message name], [expected message indexes]
+ Arguments.of(DEFAULT_PACKAGE_SCHEMA, new StandardMessageName(Optional.empty(), "User"), new int[] {0}),
+ Arguments.of(DEFAULT_PACKAGE_SCHEMA, new StandardMessageName(Optional.empty(), "Company"), new int[] {1}),
+ Arguments.of(DEFAULT_PACKAGE_SCHEMA, new StandardMessageName(Optional.empty(), "Address"), new int[] {2}),
+ Arguments.of(DEFAULT_PACKAGE_SCHEMA, new StandardMessageName(Optional.empty(), "User.Profile"), new int[] {0, 0}),
+ Arguments.of(DEFAULT_PACKAGE_SCHEMA, new StandardMessageName(Optional.empty(), "User.Profile.Settings"), new int[] {0, 0, 0}),
+
+ Arguments.of(EXPLICIT_PACKAGE_SCHEMA, new StandardMessageName(Optional.of("com.example.proto"), "User"), new int[] {0}),
+ Arguments.of(EXPLICIT_PACKAGE_SCHEMA, new StandardMessageName(Optional.of("com.example.proto"), "Company"), new int[] {1}),
+ Arguments.of(EXPLICIT_PACKAGE_SCHEMA, new StandardMessageName(Optional.of("com.example.proto"), "Address"), new int[] {2}),
+ Arguments.of(EXPLICIT_PACKAGE_SCHEMA, new StandardMessageName(Optional.of("com.example.proto"), "User.Profile"), new int[] {0, 0}),
+ Arguments.of(EXPLICIT_PACKAGE_SCHEMA, new StandardMessageName(Optional.of("com.example.proto"), "User.Profile.Settings"), new int[] {0, 0, 0})
+ );
+ }
+
+ @ParameterizedTest
+ @MethodSource("provideMessageIndexTestCases")
+ void testEncode(final String schemaText, final MessageName messageName, final int[] expectedIndexes) throws IOException {
+ final List rootMessages = new AntlrProtobufMessageSchemaParser().parse(schemaText);
+
+ final byte[] encoded = ProtobufMessageIndexEncoder.encode(rootMessages, messageName);
+
+ assertEquals(IntStream.of(expectedIndexes).boxed().toList(), decodeIndexes(encoded));
+ }
+
+ @Test
+ void testEncodeFirstRootMessageUsesSingleByteOptimization() {
+ final List rootMessages = new AntlrProtobufMessageSchemaParser().parse(DEFAULT_PACKAGE_SCHEMA);
+
+ final byte[] encoded = ProtobufMessageIndexEncoder.encode(rootMessages, new StandardMessageName(Optional.empty(), "User"));
+
+ assertArrayEquals(new byte[] {0x00}, encoded);
+ }
+
+ @Test
+ void testEncodeMessageNotFoundThrows() {
+ final List rootMessages = new AntlrProtobufMessageSchemaParser().parse(DEFAULT_PACKAGE_SCHEMA);
+
+ assertThrows(IllegalStateException.class,
+ () -> ProtobufMessageIndexEncoder.encode(rootMessages, new StandardMessageName(Optional.empty(), "DoesNotExist")));
+ }
+
+ @Test
+ void testEncodeNamespaceMismatchThrows() {
+ final List rootMessages = new AntlrProtobufMessageSchemaParser().parse(EXPLICIT_PACKAGE_SCHEMA);
+
+ assertThrows(IllegalStateException.class,
+ () -> ProtobufMessageIndexEncoder.encode(rootMessages, new StandardMessageName(Optional.empty(), "User")));
+ }
+
+ @Test
+ void testEncodeNestedMessageFromFactorySplitName() throws IOException {
+ // StandardMessageNameFactory.fromName is what the writer's static Message Name property uses. For a nested
+ // message it splits "User.Profile" into namespace="User"/name="Profile" - the encoder must still resolve it
+ // to the nested index [0, 0] by matching on the fully qualified name.
+ final List rootMessages = new AntlrProtobufMessageSchemaParser().parse(DEFAULT_PACKAGE_SCHEMA);
+
+ final byte[] encoded = ProtobufMessageIndexEncoder.encode(rootMessages, StandardMessageNameFactory.fromName("User.Profile"));
+
+ assertEquals(List.of(0, 0), decodeIndexes(encoded));
+ }
+
+ @Test
+ void testEncodeNestedMessageWithPackageFromFactorySplitName() throws IOException {
+ // With a package, fromName splits "com.example.proto.User.Profile" into namespace="com.example.proto.User"/
+ // name="Profile"; the fully qualified name still resolves to the nested index [0, 0].
+ final List rootMessages = new AntlrProtobufMessageSchemaParser().parse(EXPLICIT_PACKAGE_SCHEMA);
+
+ final byte[] encoded = ProtobufMessageIndexEncoder.encode(rootMessages, StandardMessageNameFactory.fromName("com.example.proto.User.Profile"));
+
+ assertEquals(List.of(0, 0), decodeIndexes(encoded));
+ }
+
+ /**
+ * Decodes message indexes according to Confluent wire format, mirroring the read side, to
+ * verify what {@link ProtobufMessageIndexEncoder} produced can be decoded back correctly.
+ */
+ private List decodeIndexes(final byte[] encoded) throws IOException {
+ final ByteArrayInputStream inputStream = new ByteArrayInputStream(encoded);
+ final int firstByte = inputStream.read();
+ if (firstByte == 0) {
+ return List.of(0);
+ }
+
+ int arrayLength = VarintUtils.readVarintFromStreamAfterFirstByteConsumed(inputStream, firstByte);
+ arrayLength = VarintUtils.decodeZigZag(arrayLength);
+
+ final List indexes = new ArrayList<>();
+ for (int i = 0; i < arrayLength; i++) {
+ final int rawIndex = VarintUtils.readVarintFromStream(inputStream);
+ indexes.add(VarintUtils.decodeZigZag(rawIndex));
+ }
+ return indexes;
+ }
+}
diff --git a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-name-resolver/src/main/java/org/apache/nifi/confluent/schemaregistry/ConfluentProtobufMessageNameResolver.java b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-name-resolver/src/main/java/org/apache/nifi/confluent/schemaregistry/ConfluentProtobufMessageNameResolver.java
index 25bb3565aa3d..67452f540791 100644
--- a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-name-resolver/src/main/java/org/apache/nifi/confluent/schemaregistry/ConfluentProtobufMessageNameResolver.java
+++ b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-name-resolver/src/main/java/org/apache/nifi/confluent/schemaregistry/ConfluentProtobufMessageNameResolver.java
@@ -42,9 +42,9 @@
import static java.lang.String.format;
import static java.util.Collections.singletonList;
-import static org.apache.nifi.confluent.schemaregistry.VarintUtils.decodeZigZag;
-import static org.apache.nifi.confluent.schemaregistry.VarintUtils.readVarintFromStream;
-import static org.apache.nifi.confluent.schemaregistry.VarintUtils.readVarintFromStreamAfterFirstByteConsumed;
+import static org.apache.nifi.confluent.schema.VarintUtils.decodeZigZag;
+import static org.apache.nifi.confluent.schema.VarintUtils.readVarintFromStream;
+import static org.apache.nifi.confluent.schema.VarintUtils.readVarintFromStreamAfterFirstByteConsumed;
@Tags({"confluent", "schema", "registry", "protobuf", "message", "name", "resolver"})
@CapabilityDescription("""
diff --git a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-name-resolver/src/test/java/org/apache/nifi/confluent/schemaregistry/ConfluentProtobufMessageNameResolverTest.java b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-name-resolver/src/test/java/org/apache/nifi/confluent/schemaregistry/ConfluentProtobufMessageNameResolverTest.java
index 1c8aeb67a7d7..50bca7695f06 100644
--- a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-name-resolver/src/test/java/org/apache/nifi/confluent/schemaregistry/ConfluentProtobufMessageNameResolverTest.java
+++ b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-name-resolver/src/test/java/org/apache/nifi/confluent/schemaregistry/ConfluentProtobufMessageNameResolverTest.java
@@ -39,7 +39,7 @@
import java.util.Map;
import java.util.stream.Stream;
-import static org.apache.nifi.confluent.schemaregistry.VarintUtils.writeZigZagVarint;
+import static org.apache.nifi.confluent.schema.VarintUtils.writeZigZagVarint;
import static org.apache.nifi.schemaregistry.services.SchemaDefinition.SchemaType.PROTOBUF;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
diff --git a/nifi-extension-bundles/nifi-confluent-platform-bundle/pom.xml b/nifi-extension-bundles/nifi-confluent-platform-bundle/pom.xml
index e6c3d4d12c96..2f915ef53f68 100644
--- a/nifi-extension-bundles/nifi-confluent-platform-bundle/pom.xml
+++ b/nifi-extension-bundles/nifi-confluent-platform-bundle/pom.xml
@@ -24,6 +24,7 @@
nifi-confluent-schema-registry-service
nifi-confluent-protobuf-message-name-resolver
+ nifi-confluent-protobuf-message-index-writer
nifi-confluent-platform-nar
nifi-confluent-protobuf-antlr-parser
nifi-confluent-platform-schema-api
diff --git a/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/pom.xml b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/pom.xml
index 96f3cce314b5..25aba10922db 100644
--- a/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/pom.xml
+++ b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/pom.xml
@@ -75,6 +75,25 @@
nifi-mock
test
+
+
+ org.apache.nifi
+ nifi-confluent-schema-registry-service
+ ${project.version}
+ test
+
+
+ org.apache.nifi
+ nifi-confluent-protobuf-message-index-writer
+ ${project.version}
+ test
+
+
+ org.apache.nifi
+ nifi-confluent-protobuf-message-name-resolver
+ ${project.version}
+ test
+
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 da4f20eac556..b5346016f067 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 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;
@@ -43,7 +44,7 @@
import static java.nio.file.StandardOpenOption.CREATE;
import static java.nio.file.StandardOpenOption.TRUNCATE_EXISTING;
import static java.nio.file.StandardOpenOption.WRITE;
-import static org.apache.nifi.services.protobuf.ProtobufSchemaValidator.validateSchemaDefinitionIdentifiers;
+import static org.apache.nifi.services.protobuf.ProtobufSchemaValidator.validateSchemaReferencePaths;
/**
* Handles Protocol Buffer schema compilation, caching, and temporary directory operations.
@@ -114,8 +115,8 @@ public Schema compileOrGetFromCache(final SchemaDefinition schemaDefinition) {
private Schema compileSchemaDefinition(final SchemaDefinition schemaDefinition) throws IOException {
logger.debug("Starting schema compilation for identifier: {}", schemaDefinition.getIdentifier());
- // Validate that all schema identifiers end with .proto extension
- validateSchemaDefinitionIdentifiers(schemaDefinition, true);
+ // Validate that every schema reference is keyed by an import path ending in .proto
+ validateSchemaReferencePaths(schemaDefinition);
return executeWithTemporaryDirectory(tempDir -> {
try {
@@ -182,8 +183,7 @@ private void safeDeleteDirectory(final Path directory) {
}
/**
- * Writes a schema definition to the temporary directory structure.
- * If package name is present, creates the appropriate directory structure.
+ * Writes a schema definition to the temporary directory structure using the name of its identifier.
*
* @param tempDir the temporary directory root
* @param schemaDefinition the schema definition to write
@@ -191,14 +191,32 @@ private void safeDeleteDirectory(final Path directory) {
*/
private void writeSchemaToTempDirectory(final Path tempDir, final SchemaDefinition schemaDefinition) throws IOException {
logger.debug("Writing schema definition to temporary directory. Identifier: {}", schemaDefinition.getIdentifier());
+ writeSchemaFile(tempDir, generateSchemaFileName(schemaDefinition), schemaDefinition.getText());
+ }
+
+ /**
+ * Writes schema text to a path relative to the temporary directory, creating any parent directories.
+ * Import paths may contain directories, such as {@code airlines/ph/cdm/shared.proto}, so the enclosing
+ * directory structure has to exist before the file is written.
+ *
+ * @param tempDir the temporary directory root
+ * @param relativePath the path of the schema file relative to the temporary directory
+ * @param schemaText the schema text to write
+ * @throws IOException if unable to create directories or write files
+ */
+ private void writeSchemaFile(final Path tempDir, final String relativePath, final String schemaText) throws IOException {
+ final Path schemaFile = tempDir.resolve(relativePath).normalize();
+ if (!schemaFile.startsWith(tempDir)) {
+ throw new IOException("Schema file path is not contained in the schema directory: " + relativePath);
+ }
- final String schemaFileName = generateSchemaFileName(schemaDefinition);
- final Path schemaFile = tempDir.resolve(schemaFileName);
+ final Path parentDirectory = schemaFile.getParent();
+ if (parentDirectory != null) {
+ Files.createDirectories(parentDirectory);
+ }
- // Write schema text to file
- Files.write(schemaFile, schemaDefinition.getText().getBytes(), CREATE, WRITE, TRUNCATE_EXISTING);
- logger.debug("Successfully wrote schema to file: {} (string length: {})",
- schemaFile, schemaDefinition.getText().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());
}
/**
@@ -230,8 +248,10 @@ private void processSchemaReferences(final Path tempDir, final Map
+ * A reference is keyed by the path used in the import statement of the referencing schema, for example
+ * {@code airlines/ph/cdm/shared.proto}, and the referenced schema is written to exactly that path so the import
+ * resolves. The extension is required because the compiler only discovers files named {@code *.proto}; without it
+ * the schema would fail to compile later with an unresolved import that does not indicate the cause.
+ *
+ * The identifier of a referenced schema is deliberately not validated. It carries the subject the schema is
+ * registered under, which is unrelated to the import path and legitimately has no .proto suffix. Under the
+ * Confluent RecordNameStrategy, for instance, a subject is a fully qualified record name.
*
- * @param schemaDefinition the schema definition to validate
- * @param isRootSchemaDefinition set to true if schema definition is a root definition, false otherwise
- * @throws IllegalArgumentException if any identifier does not end with .proto extension
+ * @param schemaDefinition the schema definition whose references should be validated
+ * @throws IllegalArgumentException if any reference is keyed by a path that does not end in .proto
*/
- static void validateSchemaDefinitionIdentifiers(final SchemaDefinition schemaDefinition, final boolean isRootSchemaDefinition) {
- // do not validate schema identifier names for root schema definitions. They might be coming from sources like text fields,
- // flow file attributes and other sources that do not support naming.
- if (!isRootSchemaDefinition) {
- validateSchemaIdentifier(schemaDefinition.getIdentifier());
- }
-
- // Recursively validate all referenced schemas
- // schema references have to end with .proto extension.
- for (final SchemaDefinition referencedSchema : schemaDefinition.getReferences().values()) {
- validateSchemaDefinitionIdentifiers(referencedSchema, false);
+ static void validateSchemaReferencePaths(final SchemaDefinition schemaDefinition) {
+ for (final Map.Entry reference : schemaDefinition.getReferences().entrySet()) {
+ validateReferencePath(reference.getKey());
+ validateSchemaReferencePaths(reference.getValue());
}
}
- /**
- * Validates that a single SchemaIdentifier has a name ending with .proto extension.
- *
- * @param schemaIdentifier the schema identifier to validate
- * @throws IllegalArgumentException if the identifier name does not end with .proto extension
- */
- private static void validateSchemaIdentifier(final SchemaIdentifier schemaIdentifier) {
- schemaIdentifier.getName()
- .filter(name -> name.endsWith(".proto"))
- .orElseThrow(() -> new IllegalArgumentException("Schema identifier must have a name that ends with .proto extension. Schema identifier: " + schemaIdentifier));
+ private static void validateReferencePath(final String referencePath) {
+ if (referencePath == null || !referencePath.endsWith(PROTO_EXTENSION)) {
+ throw new IllegalArgumentException(
+ "Schema reference must be keyed by the import path of the referenced schema, ending with the .proto extension. Schema reference: " + referencePath);
+ }
}
}
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
new file mode 100644
index 000000000000..2344f565b588
--- /dev/null
+++ b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/StandardProtobufWriter.java
@@ -0,0 +1,356 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.services.protobuf;
+
+import com.squareup.wire.schema.Schema;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.DescribedValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.PropertyValue;
+import org.apache.nifi.context.PropertyContext;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.controller.ControllerServiceInitializationContext;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.reporting.InitializationException;
+import org.apache.nifi.schema.access.SchemaNotFoundException;
+import org.apache.nifi.schemaregistry.services.MessageIndexWriter;
+import org.apache.nifi.schemaregistry.services.MessageName;
+import org.apache.nifi.schemaregistry.services.MessageNameResolver;
+import org.apache.nifi.schemaregistry.services.SchemaDefinition;
+import org.apache.nifi.schemaregistry.services.SchemaReferenceWriter;
+import org.apache.nifi.schemaregistry.services.SchemaRegistry;
+import org.apache.nifi.schemaregistry.services.StandardMessageNameFactory;
+import org.apache.nifi.schemaregistry.services.StandardSchemaDefinition;
+import org.apache.nifi.serialization.RecordSetWriter;
+import org.apache.nifi.serialization.RecordSetWriterFactory;
+import org.apache.nifi.serialization.SchemaRegistryService;
+import org.apache.nifi.serialization.SimpleRecordSchema;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.serialization.record.SchemaIdentifier;
+import org.apache.nifi.services.protobuf.schema.ProtoSchemaParser;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.nifi.expression.ExpressionLanguageScope.FLOWFILE_ATTRIBUTES;
+import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_ACCESS_STRATEGY;
+import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_BRANCH_NAME;
+import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_NAME;
+import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_NAME_PROPERTY;
+import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_REFERENCE_READER;
+import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_REGISTRY;
+import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_TEXT;
+import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_TEXT_PROPERTY;
+import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_VERSION;
+import static org.apache.nifi.services.protobuf.StandardProtobufWriter.MessageNameResolverStrategy.MESSAGE_NAME_PROPERTY;
+
+@Tags({"protobuf", "record", "writer", "serializer", "confluent"})
+@CapabilityDescription("""
+ Serializes NiFi Records into Protocol Buffers binary format. \
+ Supports inline schema text and schema registry lookup for determining the Proto schema. \
+ When a Schema Reference Writer is configured, a Confluent wire-format header is written; when a \
+ Message Index Writer is also configured, the Confluent message index array is written after the header. \
+ The target Proto message name can be determined statically using the 'Message Name' property, \
+ or dynamically using a Message Name Resolver service.
+ A single record is written per FlowFile, since concatenated Protocol Buffers messages cannot be delimited. \
+ The 'google.protobuf.Any' well-known type is not expanded on write; a Record derived from an Any-typed message \
+ is serialized as an ordinary nested message rather than being re-wrapped as an Any.""")
+public class StandardProtobufWriter extends SchemaRegistryService implements RecordSetWriterFactory {
+
+ public static final PropertyDescriptor MESSAGE_NAME_RESOLUTION_STRATEGY = new PropertyDescriptor.Builder()
+ .name("Message Name Resolution Strategy")
+ .description("Strategy for determining the Protocol Buffers message name for serialization")
+ .required(true)
+ .allowableValues(MESSAGE_NAME_PROPERTY, MessageNameResolverStrategy.MESSAGE_NAME_RESOLVER)
+ .defaultValue(MESSAGE_NAME_PROPERTY)
+ .build();
+
+ public static final PropertyDescriptor MESSAGE_NAME = new PropertyDescriptor.Builder()
+ .name("Message Name")
+ .description("Fully qualified name of the Protocol Buffers message including its package (eg. mypackage.MyMessage).")
+ .required(true)
+ .expressionLanguageSupported(FLOWFILE_ATTRIBUTES)
+ .dependsOn(MESSAGE_NAME_RESOLUTION_STRATEGY, MESSAGE_NAME_PROPERTY)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor MESSAGE_NAME_RESOLVER = new PropertyDescriptor.Builder()
+ .name("Message Name Resolver")
+ .description("Service that dynamically resolves Protocol Buffer message names from FlowFile attributes. "
+ + "On the write side the resolver is invoked with an empty content stream, so only resolvers that derive the "
+ + "message name from attributes are supported; resolvers that read the message name from message content "
+ + "(such as the Confluent wire-format resolver used on the read side) are not applicable here.")
+ .required(true)
+ .identifiesControllerService(MessageNameResolver.class)
+ .dependsOn(MESSAGE_NAME_RESOLUTION_STRATEGY, MessageNameResolverStrategy.MESSAGE_NAME_RESOLVER)
+ .build();
+
+ public static final PropertyDescriptor SCHEMA_REFERENCE_WRITER = new PropertyDescriptor.Builder()
+ .name("Schema Reference Writer")
+ .description("Service used to write schema reference information, such as a Confluent wire-format header, before the serialized Protobuf content. "
+ + "When not configured, plain Protobuf content is written without any header.")
+ .required(false)
+ .identifiesControllerService(SchemaReferenceWriter.class)
+ .build();
+
+ public static final PropertyDescriptor MESSAGE_INDEX_WRITER = new PropertyDescriptor.Builder()
+ .name("Message Index Writer")
+ .description("Service used to write the Confluent message index array identifying the target message within the schema, written after the Schema Reference Writer header. "
+ + "Applicable only when producing Confluent wire-format content.")
+ .required(false)
+ .identifiesControllerService(MessageIndexWriter.class)
+ .build();
+
+ private static final PropertyDescriptor PROTOBUF_SCHEMA_TEXT = new PropertyDescriptor.Builder()
+ .fromPropertyDescriptor(SCHEMA_TEXT)
+ .required(true)
+ .clearValidators()
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .defaultValue("${proto.schema}")
+ .description("The text of a Proto 3 formatted Schema")
+ .build();
+
+ private static final String PROTO_EXTENSION = ".proto";
+
+ private static final InputStream EMPTY_INPUT_STREAM = new ByteArrayInputStream(new byte[0]);
+
+ private volatile ProtobufSchemaCompiler schemaCompiler;
+ private volatile MessageNameResolver messageNameResolver;
+ private volatile SchemaReferenceWriter schemaReferenceWriter;
+ private volatile MessageIndexWriter messageIndexWriter;
+ private volatile SchemaRegistry schemaRegistry;
+ private volatile String schemaAccessStrategyValue;
+ private volatile PropertyValue schemaText;
+ private volatile PropertyValue schemaName;
+ private volatile PropertyValue schemaBranchName;
+ private volatile PropertyValue schemaVersion;
+
+ @OnEnabled
+ public void onEnabled(final ConfigurationContext context) {
+ super.storeSchemaAccessStrategy(context);
+ setupMessageNameResolver(context);
+ schemaAccessStrategyValue = context.getProperty(SCHEMA_ACCESS_STRATEGY).getValue();
+ schemaRegistry = context.getProperty(SCHEMA_REGISTRY).asControllerService(SchemaRegistry.class);
+ schemaReferenceWriter = context.getProperty(SCHEMA_REFERENCE_WRITER).asControllerService(SchemaReferenceWriter.class);
+ messageIndexWriter = context.getProperty(MESSAGE_INDEX_WRITER).asControllerService(MessageIndexWriter.class);
+ schemaName = context.getProperty(SCHEMA_NAME);
+ schemaText = context.getProperty(SCHEMA_TEXT);
+ schemaBranchName = context.getProperty(SCHEMA_BRANCH_NAME);
+ schemaVersion = context.getProperty(SCHEMA_VERSION);
+ }
+
+ @Override
+ protected void init(final ControllerServiceInitializationContext config) throws InitializationException {
+ super.init(config);
+ schemaCompiler = new ProtobufSchemaCompiler(getIdentifier(), getLogger());
+ }
+
+ @Override
+ public RecordSchema getSchema(final Map variables, final RecordSchema readSchema) throws SchemaNotFoundException, IOException {
+ return createWriteContext(variables).recordSchema();
+ }
+
+ @Override
+ public RecordSetWriter createWriter(final ComponentLog logger, final RecordSchema schema, final OutputStream out, final Map variables) throws SchemaNotFoundException, IOException {
+ final ProtobufWriteContext context = createWriteContext(variables);
+ return new WriteProtobufResultWithExternalSchema(context.schema(), context.messageName(), context.recordSchema(),
+ context.schemaDefinition(), schemaReferenceWriter, messageIndexWriter, variables, out);
+ }
+
+ @Override
+ protected List getSupportedPropertyDescriptors() {
+ final List properties = new ArrayList<>(super.getSupportedPropertyDescriptors());
+ // The Schema Reference Reader is a read-side concern: a writer determines its schema from the configured
+ // access strategy and writes references through the Schema Reference Writer instead.
+ properties.removeIf(property -> SCHEMA_REFERENCE_READER.getName().equals(property.getName()));
+ properties.add(MESSAGE_NAME_RESOLUTION_STRATEGY);
+ properties.add(MESSAGE_NAME_RESOLVER);
+ properties.add(MESSAGE_NAME);
+ properties.add(SCHEMA_REFERENCE_WRITER);
+ properties.add(MESSAGE_INDEX_WRITER);
+ return properties;
+ }
+
+ @Override
+ protected List getSchemaAccessStrategyValues() {
+ // Only the strategies that createSchemaDefinition supports are offered; the inherited list also contains
+ // the Schema Reference Reader strategy, which cannot be used to obtain a schema for writing.
+ return List.of(SCHEMA_NAME_PROPERTY, SCHEMA_TEXT_PROPERTY);
+ }
+
+ @Override
+ protected PropertyDescriptor buildSchemaTextProperty() {
+ return PROTOBUF_SCHEMA_TEXT;
+ }
+
+ private ProtobufWriteContext createWriteContext(final Map variables) throws SchemaNotFoundException, IOException {
+ final SchemaDefinition schemaDefinition = createSchemaDefinition(variables);
+ final Schema schema = schemaCompiler.compileOrGetFromCache(schemaDefinition);
+ final MessageName messageName = messageNameResolver.getMessageName(variables, schemaDefinition, EMPTY_INPUT_STREAM);
+
+ final ProtoSchemaParser schemaParser = new ProtoSchemaParser(schema);
+ final RecordSchema parsedSchema = schemaParser.createSchema(messageName.getFullyQualifiedName());
+ // Preserve the schema identifier from the SchemaDefinition so the configured Schema Reference Writer can
+ // write the correct schema id in the Confluent header.
+ final RecordSchema recordSchema = new SimpleRecordSchema(parsedSchema.getFields(), schemaDefinition.getIdentifier());
+
+ return new ProtobufWriteContext(schemaDefinition, schema, messageName, recordSchema);
+ }
+
+ private SchemaDefinition createSchemaDefinition(final Map variables) throws SchemaNotFoundException, IOException {
+ if (SCHEMA_TEXT_PROPERTY.getValue().equals(schemaAccessStrategyValue)) {
+ return createSchemaDefinitionFromText(variables);
+ } else if (SCHEMA_NAME_PROPERTY.getValue().equals(schemaAccessStrategyValue)) {
+ return createSchemaDefinitionFromRegistry(variables);
+ }
+
+ throw new SchemaNotFoundException("Unsupported schema access strategy: " + schemaAccessStrategyValue);
+ }
+
+ private void setupMessageNameResolver(final ConfigurationContext context) {
+ final MessageNameResolverStrategy messageNameResolverStrategy = context.getProperty(MESSAGE_NAME_RESOLUTION_STRATEGY).asAllowableValue(MessageNameResolverStrategy.class);
+ messageNameResolver = switch (messageNameResolverStrategy) {
+ case MESSAGE_NAME_PROPERTY -> new PropertyMessageNameResolver(context);
+ case MESSAGE_NAME_RESOLVER -> context.getProperty(MESSAGE_NAME_RESOLVER).asControllerService(MessageNameResolver.class);
+ };
+ }
+
+ private SchemaDefinition createSchemaDefinitionFromText(final Map variables) throws SchemaNotFoundException {
+ final String schemaTextString = schemaText.evaluateAttributeExpressions(variables).getValue();
+ validateSchemaText(schemaTextString);
+
+ final String hash = sha256Hex(schemaTextString);
+ final SchemaIdentifier schemaIdentifier = SchemaIdentifier.builder()
+ .name(hash + PROTO_EXTENSION)
+ .build();
+
+ return new StandardSchemaDefinition(schemaIdentifier, schemaTextString, SchemaDefinition.SchemaType.PROTOBUF);
+ }
+
+ private SchemaDefinition createSchemaDefinitionFromRegistry(final Map variables) throws SchemaNotFoundException, IOException {
+ final String schemaNameValue = schemaName.evaluateAttributeExpressions(variables).getValue();
+ validateSchemaName(schemaNameValue);
+
+ final String schemaBranchNameValue = schemaBranchName.evaluateAttributeExpressions(variables).getValue();
+ final String schemaVersionValue = schemaVersion.evaluateAttributeExpressions(variables).getValue();
+
+ final SchemaIdentifier schemaIdentifier = buildSchemaIdentifier(schemaNameValue, schemaBranchNameValue, schemaVersionValue);
+ return schemaRegistry.retrieveSchemaDefinition(schemaIdentifier);
+ }
+
+ private SchemaIdentifier buildSchemaIdentifier(final String schemaNameValue, final String schemaBranchNameValue, final String schemaVersionValue) throws SchemaNotFoundException {
+ final SchemaIdentifier.Builder identifierBuilder = SchemaIdentifier.builder().name(schemaNameValue);
+
+ if (schemaBranchNameValue != null && !schemaBranchNameValue.isBlank()) {
+ identifierBuilder.branch(schemaBranchNameValue);
+ }
+
+ if (schemaVersionValue != null && !schemaVersionValue.isBlank()) {
+ try {
+ identifierBuilder.version(Integer.valueOf(schemaVersionValue));
+ } catch (final NumberFormatException nfe) {
+ throw new SchemaNotFoundException("Could not retrieve schema with name '%s' because a non-numeric version was supplied '%s'"
+ .formatted(schemaNameValue, schemaVersionValue), nfe);
+ }
+ }
+
+ return identifierBuilder.build();
+ }
+
+ private String sha256Hex(final String input) {
+ final MessageDigest digest;
+ try {
+ digest = MessageDigest.getInstance("SHA-256");
+ } catch (final NoSuchAlgorithmException e) {
+ throw new IllegalStateException(e);
+ }
+ final byte[] hash = digest.digest(input.getBytes(StandardCharsets.UTF_8));
+ return HexFormat.of().formatHex(hash);
+ }
+
+ private void validateSchemaText(final String schemaTextString) throws SchemaNotFoundException {
+ if (schemaTextString == null || schemaTextString.isBlank()) {
+ throw new SchemaNotFoundException("Schema text not found");
+ }
+ }
+
+ private void validateSchemaName(final String schemaNameValue) throws SchemaNotFoundException {
+ if (schemaNameValue == null || schemaNameValue.isBlank()) {
+ throw new SchemaNotFoundException("Schema name not provided or is blank");
+ }
+ }
+
+ enum MessageNameResolverStrategy implements DescribedValue {
+
+ MESSAGE_NAME_PROPERTY("Message Name Property", "Use the 'Message Name' property value to determine the message name"),
+ MESSAGE_NAME_RESOLVER("Message Name Resolver", "Use a 'Message Name Resolver' service to dynamically determine the message name");
+
+ private final String displayName;
+ private final String description;
+
+ MessageNameResolverStrategy(final String displayName, final String description) {
+ this.displayName = displayName;
+ this.description = description;
+ }
+
+ @Override
+ public String getValue() {
+ return name();
+ }
+
+ @Override
+ public String getDisplayName() {
+ return displayName;
+ }
+
+ @Override
+ public String getDescription() {
+ return description;
+ }
+ }
+
+ static class PropertyMessageNameResolver extends AbstractControllerService implements MessageNameResolver {
+ private final PropertyContext context;
+
+ PropertyMessageNameResolver(final PropertyContext context) {
+ this.context = context;
+ }
+
+ @Override
+ public MessageName getMessageName(final Map variables, final SchemaDefinition schemaDefinition, final InputStream in) {
+ final String messageName = context.getProperty(MESSAGE_NAME).evaluateAttributeExpressions(variables).getValue();
+ return StandardMessageNameFactory.fromName(messageName);
+ }
+ }
+
+ private record ProtobufWriteContext(SchemaDefinition schemaDefinition, Schema schema, MessageName messageName, RecordSchema recordSchema) {
+ }
+}
diff --git a/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/WriteProtobufResultWithExternalSchema.java b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/WriteProtobufResultWithExternalSchema.java
new file mode 100644
index 000000000000..40a35fb061dd
--- /dev/null
+++ b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/WriteProtobufResultWithExternalSchema.java
@@ -0,0 +1,137 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.services.protobuf;
+
+import com.squareup.wire.schema.Schema;
+import org.apache.nifi.schemaregistry.services.MessageIndexWriter;
+import org.apache.nifi.schemaregistry.services.MessageName;
+import org.apache.nifi.schemaregistry.services.SchemaDefinition;
+import org.apache.nifi.schemaregistry.services.SchemaReferenceWriter;
+import org.apache.nifi.serialization.AbstractRecordSetWriter;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.services.protobuf.converter.ProtobufDataSerializer;
+
+import java.io.BufferedOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Map;
+
+/**
+ * Writes Records as Protocol Buffers binary content. When a {@link SchemaReferenceWriter} is
+ * configured, a Confluent wire-format header (magic byte and schema identifier) is written first;
+ * when a {@link MessageIndexWriter} is configured, the Confluent message index array follows the
+ * header. The serialized Protobuf payload is written last.
+ *
+ * The Confluent framing (header and message index) is written once at the beginning of the record
+ * set, mirroring {@code WriteAvroResultWithExternalSchema}; the typical Confluent use case writes a
+ * single message per FlowFile.
+ */
+public class WriteProtobufResultWithExternalSchema extends AbstractRecordSetWriter {
+
+ private final RecordSchema recordSchema;
+ private final SchemaDefinition schemaDefinition;
+ private final MessageName messageName;
+ private final SchemaReferenceWriter schemaReferenceWriter;
+ private final MessageIndexWriter messageIndexWriter;
+ private final Map variables;
+ private final ProtobufDataSerializer serializer;
+ private final OutputStream buffered;
+ private boolean closed = false;
+
+ public WriteProtobufResultWithExternalSchema(final Schema schema,
+ final MessageName messageName,
+ final RecordSchema recordSchema,
+ final SchemaDefinition schemaDefinition,
+ final SchemaReferenceWriter schemaReferenceWriter,
+ final MessageIndexWriter messageIndexWriter,
+ final Map variables,
+ final OutputStream out) {
+ super(out);
+ this.recordSchema = recordSchema;
+ this.schemaDefinition = schemaDefinition;
+ this.messageName = messageName;
+ this.schemaReferenceWriter = schemaReferenceWriter;
+ this.messageIndexWriter = messageIndexWriter;
+ this.variables = variables;
+ this.buffered = new BufferedOutputStream(out);
+ this.serializer = new ProtobufDataSerializer(schema, messageName.getFullyQualifiedName());
+ }
+
+ @Override
+ protected void onBeginRecordSet() throws IOException {
+ writeConfluentFraming(buffered);
+ }
+
+ @Override
+ protected Map onFinishRecordSet() throws IOException {
+ flush();
+ return Map.of();
+ }
+
+ @Override
+ public Map writeRecord(final Record record) throws IOException {
+ // Concatenated top-level Protobuf messages cannot be delimited: a standard decoder would merge them into a
+ // single message (repeated fields accumulate, singular fields take last-wins). Only a single record per
+ // FlowFile can be represented, which also matches the Confluent one-message-per-record convention.
+ if (getRecordCount() > 0) {
+ throw new IOException("Protobuf output supports only a single record because concatenated Protobuf messages cannot be delimited");
+ }
+
+ // If we are not writing an active record set, then we need to ensure that the Confluent framing is written.
+ if (!isActiveRecordSet()) {
+ flush();
+ writeConfluentFraming(buffered);
+ }
+
+ final byte[] payload = serializer.serialize(record);
+ buffered.write(payload);
+ return Map.of();
+ }
+
+ private void writeConfluentFraming(final OutputStream out) throws IOException {
+ if (schemaReferenceWriter != null) {
+ schemaReferenceWriter.writeHeader(recordSchema, out);
+ }
+ if (messageIndexWriter != null) {
+ messageIndexWriter.writeMessageIndex(variables, schemaDefinition, messageName, out);
+ }
+ }
+
+ @Override
+ public void flush() throws IOException {
+ buffered.flush();
+ }
+
+ @Override
+ public String getMimeType() {
+ return "application/octet-stream";
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (closed) {
+ return;
+ }
+ closed = true;
+
+ // Ensure buffered content is flushed to the underlying stream before it is closed, including the
+ // write-without-active-record-set path where onFinishRecordSet is never invoked.
+ flush();
+ super.close();
+ }
+}
diff --git a/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/converter/ProtobufDataSerializer.java b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/converter/ProtobufDataSerializer.java
new file mode 100644
index 000000000000..70f198ded1b6
--- /dev/null
+++ b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/converter/ProtobufDataSerializer.java
@@ -0,0 +1,300 @@
+/*
+ * 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.converter;
+
+import com.google.protobuf.CodedOutputStream;
+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.ProtoType;
+import com.squareup.wire.schema.Schema;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.serialization.record.util.DataTypeUtils;
+import org.apache.nifi.services.protobuf.FieldType;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Serializes a NiFi {@link Record} into Protocol Buffers binary payload using a Square Wire
+ * {@link Schema}. This is the write-side inverse of {@code ProtobufDataConverter}: it walks the
+ * declared fields of the target message type and encodes each present Record value to a
+ * {@link CodedOutputStream} following the Protocol Buffers wire format.
+ *
+ * This class has no dependency on the NiFi framework and can be exercised directly with plain
+ * unit tests.
+ */
+public class ProtobufDataSerializer {
+
+ private static final int WIRETYPE_VARINT = 0;
+ private static final int WIRETYPE_FIXED64 = 1;
+ private static final int WIRETYPE_LENGTH_DELIMITED = 2;
+ private static final int WIRETYPE_FIXED32 = 5;
+
+ private static final int MAP_KEY_TAG = 1;
+ private static final int MAP_VALUE_TAG = 2;
+
+ private final Schema schema;
+ private final String rootMessageType;
+
+ public ProtobufDataSerializer(final Schema schema, final String rootMessageType) {
+ this.schema = schema;
+ this.rootMessageType = rootMessageType;
+ }
+
+ /**
+ * Serializes the provided Record into Protocol Buffers binary format for the configured root
+ * message type.
+ *
+ * @param record the record to serialize
+ * @return the serialized protobuf payload
+ * @throws IOException if the record cannot be encoded
+ */
+ public byte[] serialize(final Record record) throws IOException {
+ final MessageType messageType = (MessageType) schema.getType(rootMessageType);
+ Objects.requireNonNull(messageType, String.format("Message with name [%s] not found in the provided proto files", rootMessageType));
+
+ return serializeMessage(messageType, record);
+ }
+
+ private byte[] serializeMessage(final MessageType messageType, final Record record) throws IOException {
+ final ByteArrayOutputStream output = new ByteArrayOutputStream();
+ final CodedOutputStream codedOutput = CodedOutputStream.newInstance(output);
+
+ for (final Field field : messageType.getDeclaredFields()) {
+ writeField(codedOutput, field, record.getValue(field.getName()));
+ }
+ for (final Field field : messageType.getExtensionFields()) {
+ writeField(codedOutput, field, record.getValue(field.getName()));
+ }
+ for (final OneOf oneOf : messageType.getOneOfs()) {
+ for (final Field field : oneOf.getFields()) {
+ writeField(codedOutput, field, record.getValue(field.getName()));
+ }
+ }
+
+ codedOutput.flush();
+ return output.toByteArray();
+ }
+
+ private void writeField(final CodedOutputStream output, final Field field, final Object value) throws IOException {
+ if (value == null) {
+ return;
+ }
+
+ final int tag = field.getTag();
+ final ProtoType protoType = field.getType();
+
+ if (protoType.isMap()) {
+ writeMap(output, tag, protoType, value);
+ } else if (field.isRepeated()) {
+ writeRepeated(output, tag, protoType, value);
+ } else {
+ writeSingleValue(output, tag, protoType, value);
+ }
+ }
+
+ private void writeRepeated(final CodedOutputStream output, final int tag, final ProtoType protoType, final Object value) throws IOException {
+ final Object[] values = toArray(value);
+
+ if (isPackable(protoType)) {
+ // proto3 packs repeated scalar and enum fields into a single length-delimited entry
+ final ByteArrayOutputStream packedBytes = new ByteArrayOutputStream();
+ final CodedOutputStream packedOutput = CodedOutputStream.newInstance(packedBytes);
+ for (final Object element : values) {
+ writeScalarOrEnumValueNoTag(packedOutput, protoType, element);
+ }
+ packedOutput.flush();
+
+ output.writeTag(tag, WIRETYPE_LENGTH_DELIMITED);
+ final byte[] packed = packedBytes.toByteArray();
+ output.writeUInt32NoTag(packed.length);
+ output.writeRawBytes(packed);
+ } else {
+ // repeated messages, strings and bytes are written as separate length-delimited entries
+ for (final Object element : values) {
+ writeSingleValue(output, tag, protoType, element);
+ }
+ }
+ }
+
+ private void writeSingleValue(final CodedOutputStream output, final int tag, final ProtoType protoType, final Object value) throws IOException {
+ if (protoType.isScalar()) {
+ final FieldType fieldType = FieldType.findValue(protoType.getSimpleName());
+ output.writeTag(tag, wireTypeFor(fieldType));
+ writeScalarValueNoTag(output, fieldType, value);
+ return;
+ }
+
+ if (schema.getType(protoType) instanceof EnumType) {
+ output.writeTag(tag, WIRETYPE_VARINT);
+ output.writeEnumNoTag(enumTag(protoType, value));
+ return;
+ }
+
+ // nested message
+ final MessageType messageType = (MessageType) schema.getType(protoType);
+ Objects.requireNonNull(messageType, String.format("Message type with name [%s] not found in the provided proto files", protoType));
+ final byte[] nested = serializeMessage(messageType, toRecord(value, protoType));
+ output.writeTag(tag, WIRETYPE_LENGTH_DELIMITED);
+ output.writeUInt32NoTag(nested.length);
+ output.writeRawBytes(nested);
+ }
+
+ private void writeMap(final CodedOutputStream output, final int tag, final ProtoType protoType, final Object value) throws IOException {
+ if (!(value instanceof final Map map)) {
+ throw new IOException(String.format("Expected a Map value for map field but received [%s]", value.getClass()));
+ }
+
+ final ProtoType keyType = protoType.getKeyType();
+ final ProtoType valueType = protoType.getValueType();
+
+ for (final Map.Entry entry : map.entrySet()) {
+ final ByteArrayOutputStream entryBytes = new ByteArrayOutputStream();
+ final CodedOutputStream entryOutput = CodedOutputStream.newInstance(entryBytes);
+
+ writeSingleValue(entryOutput, MAP_KEY_TAG, keyType, entry.getKey());
+ if (entry.getValue() != null) {
+ writeSingleValue(entryOutput, MAP_VALUE_TAG, valueType, entry.getValue());
+ }
+ entryOutput.flush();
+
+ output.writeTag(tag, WIRETYPE_LENGTH_DELIMITED);
+ final byte[] entryEncoded = entryBytes.toByteArray();
+ output.writeUInt32NoTag(entryEncoded.length);
+ output.writeRawBytes(entryEncoded);
+ }
+ }
+
+ private void writeScalarOrEnumValueNoTag(final CodedOutputStream output, final ProtoType protoType, final Object value) throws IOException {
+ if (protoType.isScalar()) {
+ writeScalarValueNoTag(output, FieldType.findValue(protoType.getSimpleName()), value);
+ } else {
+ output.writeEnumNoTag(enumTag(protoType, value));
+ }
+ }
+
+ private void writeScalarValueNoTag(final CodedOutputStream output, final FieldType fieldType, final Object value) throws IOException {
+ // 32-bit varint types (uint32, sint32, fixed32) map to a Record LONG in ProtoSchemaParser, so they are
+ // coerced through long and narrowed to int; the unsigned/zigzag encoding is handled by CodedOutputStream.
+ switch (fieldType) {
+ case BOOL -> output.writeBoolNoTag(DataTypeUtils.toBoolean(value, null));
+ case INT32 -> output.writeInt32NoTag((int) DataTypeUtils.toLong(value, null).longValue());
+ case SFIXED32 -> output.writeSFixed32NoTag((int) DataTypeUtils.toLong(value, null).longValue());
+ case UINT32 -> output.writeUInt32NoTag((int) DataTypeUtils.toLong(value, null).longValue());
+ case SINT32 -> output.writeSInt32NoTag((int) DataTypeUtils.toLong(value, null).longValue());
+ case FIXED32 -> output.writeFixed32NoTag((int) DataTypeUtils.toLong(value, null).longValue());
+ case INT64 -> output.writeInt64NoTag(DataTypeUtils.toLong(value, null));
+ case SFIXED64 -> output.writeSFixed64NoTag(DataTypeUtils.toLong(value, null));
+ case SINT64 -> output.writeSInt64NoTag(DataTypeUtils.toLong(value, null));
+ case UINT64 -> output.writeUInt64NoTag(toBigInteger(value).longValue());
+ case FIXED64 -> output.writeFixed64NoTag(toBigInteger(value).longValue());
+ case FLOAT -> output.writeFloatNoTag(DataTypeUtils.toFloat(value, null));
+ case DOUBLE -> output.writeDoubleNoTag(DataTypeUtils.toDouble(value, null));
+ case STRING -> output.writeStringNoTag(DataTypeUtils.toString(value, (String) null));
+ case BYTES -> output.writeByteArrayNoTag(toByteArray(value));
+ }
+ }
+
+ private int wireTypeFor(final FieldType fieldType) {
+ return switch (fieldType) {
+ case BOOL, INT32, INT64, UINT32, UINT64, SINT32, SINT64 -> WIRETYPE_VARINT;
+ case FIXED32, SFIXED32, FLOAT -> WIRETYPE_FIXED32;
+ case FIXED64, SFIXED64, DOUBLE -> WIRETYPE_FIXED64;
+ case STRING, BYTES -> WIRETYPE_LENGTH_DELIMITED;
+ };
+ }
+
+ private boolean isPackable(final ProtoType protoType) {
+ if (protoType.isScalar()) {
+ final FieldType fieldType = FieldType.findValue(protoType.getSimpleName());
+ return fieldType != FieldType.STRING && fieldType != FieldType.BYTES;
+ }
+ // enums are packable; messages are not
+ return schema.getType(protoType) instanceof EnumType;
+ }
+
+ private int enumTag(final ProtoType protoType, final Object value) {
+ final EnumType enumType = (EnumType) schema.getType(protoType);
+ Objects.requireNonNull(enumType, String.format("Enum with name [%s] not found in the provided proto files", protoType));
+
+ final String constantName = String.valueOf(value);
+ final EnumConstant constant = enumType.constant(constantName);
+ if (constant == null) {
+ throw new IllegalStateException(String.format("Enum constant [%s] not found in enum [%s]", constantName, protoType));
+ }
+ return constant.getTag();
+ }
+
+ private Record toRecord(final Object value, final ProtoType protoType) throws IOException {
+ if (value instanceof final Record record) {
+ return record;
+ }
+ throw new IOException(String.format("Expected a Record value for message field [%s] but received [%s]", protoType, value.getClass()));
+ }
+
+ private Object[] toArray(final Object value) {
+ if (value instanceof final Object[] array) {
+ return array;
+ }
+ if (value instanceof final List list) {
+ return list.toArray();
+ }
+ return new Object[] {value};
+ }
+
+ private BigInteger toBigInteger(final Object value) {
+ if (value instanceof final BigInteger bigInteger) {
+ return bigInteger;
+ }
+ if (value instanceof final Number number) {
+ return BigInteger.valueOf(number.longValue());
+ }
+ return new BigInteger(String.valueOf(value));
+ }
+
+ private byte[] toByteArray(final Object value) {
+ if (value instanceof final byte[] bytes) {
+ return bytes;
+ }
+ if (value instanceof final Object[] array) {
+ final byte[] bytes = new byte[array.length];
+ for (int i = 0; i < array.length; i++) {
+ bytes[i] = ((Number) array[i]).byteValue();
+ }
+ return bytes;
+ }
+ final List collected = new ArrayList<>();
+ if (value instanceof final List list) {
+ for (final Object element : list) {
+ collected.add(((Number) element).byteValue());
+ }
+ }
+ final byte[] bytes = new byte[collected.size()];
+ for (int i = 0; i < collected.size(); i++) {
+ bytes[i] = collected.get(i);
+ }
+ return bytes;
+ }
+}
diff --git a/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService
index 8ff7783b668d..4755655bd2a1 100644
--- a/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService
+++ b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService
@@ -14,4 +14,5 @@
# limitations under the License.
org.apache.nifi.services.protobuf.ProtobufReader
-org.apache.nifi.services.protobuf.StandardProtobufReader
\ No newline at end of file
+org.apache.nifi.services.protobuf.StandardProtobufReader
+org.apache.nifi.services.protobuf.StandardProtobufWriter
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 000000000000..1fd1b51b2121
--- /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,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.services.protobuf;
+
+import com.squareup.wire.schema.Schema;
+import org.apache.nifi.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 java.util.Map;
+
+import static org.apache.nifi.schemaregistry.services.SchemaDefinition.SchemaType.PROTOBUF;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+/**
+ * Verifies schema compilation when a schema imports another file. Confluent Schema Registry keys each
+ * reference by the import path used inside the {@code .proto} (for example {@code airlines/ph/cdm/shared.proto}),
+ * while the referenced schema's identifier carries the registry subject, which is a different value and is not
+ * required to look like a file path.
+ */
+class TestProtobufSchemaCompiler {
+
+ 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;
+ }""";
+
+ @Test
+ void testCompileSchemaWithCrossFileImportFromRegistry() {
+ final SchemaDefinition referencedSchema = new StandardSchemaDefinition(
+ SchemaIdentifier.builder().name(REFERENCE_SUBJECT).id(4L).version(1).build(),
+ REFERENCED_SCHEMA,
+ PROTOBUF);
+
+ final SchemaDefinition rootSchema = new StandardSchemaDefinition(
+ SchemaIdentifier.builder().name(ROOT_SUBJECT).id(3L).version(2).build(),
+ ROOT_SCHEMA,
+ PROTOBUF,
+ Map.of(IMPORT_PATH, referencedSchema));
+
+ final ProtobufSchemaCompiler compiler = new ProtobufSchemaCompiler("test", new MockComponentLog("test", new Object()));
+ final Schema compiled = compiler.compileOrGetFromCache(rootSchema);
+
+ assertNotNull(compiled.getType("airlines.ph.cdm.reservation.AirlineReservation"));
+ assertNotNull(compiled.getType("airlines.ph.cdm.Status"));
+ }
+
+ @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;
+ }""",
+ 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;
+ }""",
+ 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,
+ PROTOBUF,
+ Map.of(IMPORT_PATH, referencedSchema));
+
+ final ProtobufSchemaCompiler compiler = new ProtobufSchemaCompiler("test", new MockComponentLog("test", new Object()));
+ 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"));
+ }
+}
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 54b36a267b45..c99a40f04f3c 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
@@ -44,7 +44,7 @@
import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_TEXT;
import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_TEXT_PROPERTY;
import static org.apache.nifi.services.protobuf.ProtoTestUtil.generateInputDataForProto3;
-import static org.apache.nifi.services.protobuf.ProtobufSchemaValidator.validateSchemaDefinitionIdentifiers;
+import static org.apache.nifi.services.protobuf.ProtobufSchemaValidator.validateSchemaReferencePaths;
import static org.apache.nifi.services.protobuf.StandardProtobufReader.MESSAGE_NAME;
import static org.apache.nifi.services.protobuf.StandardProtobufReader.MESSAGE_NAME_RESOLUTION_STRATEGY;
import static org.apache.nifi.services.protobuf.StandardProtobufReader.MESSAGE_NAME_RESOLVER;
@@ -137,7 +137,7 @@ void testValidSchemaDefinitionWithProtoExtension() {
enableAllControllerServices();
- validateSchemaDefinitionIdentifiers(validMainSchema, true);
+ validateSchemaReferencePaths(validMainSchema);
}
@Test
@@ -147,29 +147,42 @@ void testRootSchemaIsAllowedToHaveInvalidName() {
enableAllControllerServices();
- validateSchemaDefinitionIdentifiers(validSchema, true);
+ validateSchemaReferencePaths(validSchema);
}
@Test
- void testValidMainSchemaWithInvalidReferencedSchema() {
- final SchemaDefinition invalidReferencedSchema = createSchemaDefinition("user_profile.invalid");
+ void testReferenceKeyedByPathWithoutProtoExtensionIsRejected() {
+ // The key is the import path, and the compiler only discovers files named *.proto.
+ final SchemaDefinition invalidReferencedSchema = createSchemaDefinition("user_profile.proto");
final SchemaDefinition mixedSchema = createSchemaDefinition("user_settings.proto",
Map.of("user_profile.invalid", invalidReferencedSchema));
enableAllControllerServices();
- final IllegalArgumentException referencedException = assertThrows(IllegalArgumentException.class, () -> {
- validateSchemaDefinitionIdentifiers(mixedSchema, true);
- });
+ final IllegalArgumentException referencedException = assertThrows(IllegalArgumentException.class,
+ () -> validateSchemaReferencePaths(mixedSchema));
- assertTrue(referencedException.getMessage().contains("ends with .proto extension"));
+ assertTrue(referencedException.getMessage().contains(".proto extension"));
+ }
+
+ @Test
+ void testReferencedSubjectWithoutProtoExtensionIsAccepted() {
+ // A registry subject is unrelated to the import path and legitimately has no .proto suffix; under the
+ // Confluent RecordNameStrategy it is a fully qualified record name. Only the key has to look like a path.
+ final SchemaDefinition referencedSchema = createSchemaDefinition("airlines.ph.cdm.shared");
+ final SchemaDefinition mainSchema = createSchemaDefinition("airlines.ph.cdm.reservation.AirlineReservation",
+ Map.of("airlines/ph/cdm/shared.proto", referencedSchema));
+
+ enableAllControllerServices();
+
+ validateSchemaReferencePaths(mainSchema);
}
@Test
void testSchemaDefinitionWithMissingName() {
final SchemaDefinition schemaWithoutName = createSchemaDefinitionWithoutName();
enableAllControllerServices();
- validateSchemaDefinitionIdentifiers(schemaWithoutName, true);
+ validateSchemaReferencePaths(schemaWithoutName);
}
@Nested
diff --git a/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/java/org/apache/nifi/services/protobuf/TestStandardProtobufWriter.java b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/java/org/apache/nifi/services/protobuf/TestStandardProtobufWriter.java
new file mode 100644
index 000000000000..596912bd9710
--- /dev/null
+++ b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/java/org/apache/nifi/services/protobuf/TestStandardProtobufWriter.java
@@ -0,0 +1,317 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.services.protobuf;
+
+import com.squareup.wire.schema.Schema;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.schema.access.SchemaField;
+import org.apache.nifi.schema.access.SchemaNotFoundException;
+import org.apache.nifi.schemaregistry.services.MessageIndexWriter;
+import org.apache.nifi.schemaregistry.services.MessageName;
+import org.apache.nifi.schemaregistry.services.SchemaDefinition;
+import org.apache.nifi.schemaregistry.services.SchemaReferenceWriter;
+import org.apache.nifi.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordSetWriter;
+import org.apache.nifi.serialization.record.MapRecord;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.services.protobuf.converter.ProtobufDataConverter;
+import org.apache.nifi.services.protobuf.schema.ProtoSchemaParser;
+import org.apache.nifi.util.NoOpProcessor;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static java.util.Collections.emptyMap;
+import static java.util.Collections.emptySet;
+import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_ACCESS_STRATEGY;
+import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_NAME_PROPERTY;
+import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_REFERENCE_READER;
+import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_TEXT;
+import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_TEXT_PROPERTY;
+import static org.apache.nifi.services.protobuf.ProtoTestUtil.generateInputDataForProto3;
+import static org.apache.nifi.services.protobuf.ProtoTestUtil.generateInputDataForRepeatedProto3;
+import static org.apache.nifi.services.protobuf.ProtoTestUtil.loadProto3TestSchema;
+import static org.apache.nifi.services.protobuf.ProtoTestUtil.loadRepeatedProto3TestSchema;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestStandardProtobufWriter {
+
+ private static final String PROTO_3_MESSAGE = "Proto3Message";
+ private static final byte[] FAKE_HEADER = {0x00, 0x00, 0x00, 0x00, 0x2A};
+ private static final byte[] FAKE_INDEX = {0x00};
+
+ private TestRunner runner;
+ private StandardProtobufWriter writer;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ runner = TestRunners.newTestRunner(NoOpProcessor.class);
+ writer = new StandardProtobufWriter();
+ runner.addControllerService("writer", writer);
+ runner.setProperty(writer, SCHEMA_ACCESS_STRATEGY, SCHEMA_TEXT_PROPERTY.getValue());
+ runner.setProperty(writer, SCHEMA_TEXT, getTestProto3File());
+ runner.setProperty(writer, StandardProtobufWriter.MESSAGE_NAME_RESOLUTION_STRATEGY,
+ StandardProtobufWriter.MessageNameResolverStrategy.MESSAGE_NAME_PROPERTY.getValue());
+ runner.setProperty(writer, StandardProtobufWriter.MESSAGE_NAME, PROTO_3_MESSAGE);
+ }
+
+ @Test
+ void testWritePlainProtobufRoundTrip() throws Exception {
+ runner.enableControllerService(writer);
+
+ final byte[] output = writeSingleRecord(buildProto3Record());
+
+ assertProto3Message(new ByteArrayInputStream(output));
+ }
+
+ @Test
+ void testOnlySupportedSchemaAccessStrategiesAreOffered() {
+ // The inherited strategy list also contains the Schema Reference Reader strategy, which the writer cannot
+ // use to obtain a schema; offering it would let a user configure a service that always fails at runtime.
+ final List descriptors = writer.getPropertyDescriptors();
+ final PropertyDescriptor strategy = descriptors.stream()
+ .filter(descriptor -> SCHEMA_ACCESS_STRATEGY.getName().equals(descriptor.getName()))
+ .findFirst()
+ .orElseThrow();
+
+ final List allowed = strategy.getAllowableValues().stream().map(AllowableValue::getValue).toList();
+ assertEquals(List.of(SCHEMA_NAME_PROPERTY.getValue(), SCHEMA_TEXT_PROPERTY.getValue()), allowed);
+
+ // The read-side Schema Reference Reader property is not applicable to a writer.
+ assertTrue(descriptors.stream().noneMatch(descriptor -> SCHEMA_REFERENCE_READER.getName().equals(descriptor.getName())));
+ }
+
+ @Test
+ void testWriteWithoutActiveRecordSetFlushesOnClose() throws Exception {
+ runner.enableControllerService(writer);
+
+ final RecordSchema writeSchema = writer.getSchema(emptyMap(), null);
+ final ByteArrayOutputStream out = new ByteArrayOutputStream();
+ // No beginRecordSet/finishRecordSet: content is written outside an active record set and must survive close().
+ try (RecordSetWriter recordSetWriter = writer.createWriter(runner.getLogger(), writeSchema, out, emptyMap())) {
+ recordSetWriter.write(buildProto3Record());
+ }
+
+ assertProto3Message(new ByteArrayInputStream(out.toByteArray()));
+ }
+
+ @Test
+ void testWriteMultipleRecordsFailsFast() throws Exception {
+ runner.enableControllerService(writer);
+
+ final RecordSchema writeSchema = writer.getSchema(emptyMap(), null);
+ final ByteArrayOutputStream out = new ByteArrayOutputStream();
+ try (RecordSetWriter recordSetWriter = writer.createWriter(runner.getLogger(), writeSchema, out, emptyMap())) {
+ recordSetWriter.beginRecordSet();
+ recordSetWriter.write(buildProto3Record());
+ assertThrows(IOException.class, () -> recordSetWriter.write(buildProto3Record()));
+ }
+ }
+
+ @Test
+ void testWriteConfluentFramingOrder() throws Exception {
+ final FakeSchemaReferenceWriter referenceWriter = new FakeSchemaReferenceWriter();
+ final FakeMessageIndexWriter indexWriter = new FakeMessageIndexWriter();
+ runner.addControllerService("referenceWriter", referenceWriter);
+ runner.addControllerService("indexWriter", indexWriter);
+ runner.enableControllerService(referenceWriter);
+ runner.enableControllerService(indexWriter);
+ runner.setProperty(writer, StandardProtobufWriter.SCHEMA_REFERENCE_WRITER, "referenceWriter");
+ runner.setProperty(writer, StandardProtobufWriter.MESSAGE_INDEX_WRITER, "indexWriter");
+ runner.enableControllerService(writer);
+
+ final byte[] output = writeSingleRecord(buildProto3Record());
+
+ // Confluent framing: [header][message index][protobuf payload]
+ final byte[] header = Arrays.copyOfRange(output, 0, FAKE_HEADER.length);
+ final byte[] index = Arrays.copyOfRange(output, FAKE_HEADER.length, FAKE_HEADER.length + FAKE_INDEX.length);
+ assertArrayEquals(FAKE_HEADER, header);
+ assertArrayEquals(FAKE_INDEX, index);
+
+ final byte[] payload = Arrays.copyOfRange(output, FAKE_HEADER.length + FAKE_INDEX.length, output.length);
+ assertProto3Message(new ByteArrayInputStream(payload));
+ }
+
+ @Test
+ void testRoundTripThroughStandardProtobufReaderProto3() throws Exception {
+ runner.enableControllerService(writer);
+ final byte[] output = writeSingleRecord(buildProto3Record());
+
+ final StandardProtobufReader reader = createAndEnableReader("proto3reader", getTestProto3File(), PROTO_3_MESSAGE);
+ final RecordReader recordReader = reader.createRecordReader(emptyMap(), new ByteArrayInputStream(output), output.length, runner.getLogger());
+
+ final Record record = recordReader.nextRecord();
+ assertEquals(true, record.getValue("booleanField"));
+ assertEquals("Test text", record.getValue("stringField"));
+ assertEquals(Integer.MAX_VALUE, record.getValue("int32Field"));
+ assertEquals(Long.MAX_VALUE, record.getValue("int64Field"));
+
+ final Record nested = (Record) record.getValue("nestedMessage");
+ assertEquals("ENUM_VALUE_3", nested.getValue("testEnum"));
+ final Object[] nested2 = (Object[]) nested.getValue("nestedMessage2");
+ final Record oneOfHolder = (Record) nested2[0];
+ assertEquals(3, oneOfHolder.getValue("int32Option"));
+
+ assertNull(recordReader.nextRecord());
+ }
+
+ @Test
+ void testRoundTripThroughStandardProtobufReaderRepeated() throws Exception {
+ final String repeatedSchema = getSchemaFile("test_repeated_proto3.proto");
+ final StandardProtobufWriter repeatedWriter = createAndEnableWriter("repeatedwriter", repeatedSchema, "RootMessage");
+ final byte[] output = writeSingleRecordWith(repeatedWriter, buildRepeatedRecord());
+
+ final StandardProtobufReader reader = createAndEnableReader("repeatedreader", repeatedSchema, "RootMessage");
+ final RecordReader recordReader = reader.createRecordReader(emptyMap(), new ByteArrayInputStream(output), output.length, runner.getLogger());
+
+ final Record record = recordReader.nextRecord();
+ final Object[] repeatedMessage = (Object[]) record.getValue("repeatedMessage");
+ final Record first = (Record) repeatedMessage[0];
+ assertArrayEquals(new Object[]{true, false}, (Object[]) first.getValue("booleanField"));
+ assertArrayEquals(new Object[]{"Test text1", "Test text2"}, (Object[]) first.getValue("stringField"));
+ assertArrayEquals(new Object[]{"ENUM_VALUE_2", "ENUM_VALUE_3"}, (Object[]) first.getValue("testEnum"));
+
+ assertNull(recordReader.nextRecord());
+ }
+
+ private MapRecord buildRepeatedRecord() throws Exception {
+ final Schema schema = loadRepeatedProto3TestSchema();
+ final RecordSchema recordSchema = new ProtoSchemaParser(schema).createSchema("RootMessage");
+ return new ProtobufDataConverter(schema, "RootMessage", recordSchema, false, false)
+ .createRecord(generateInputDataForRepeatedProto3());
+ }
+
+ private StandardProtobufReader createAndEnableReader(final String id, final String schemaText, final String messageName) throws Exception {
+ final StandardProtobufReader reader = new StandardProtobufReader();
+ runner.addControllerService(id, reader);
+ runner.setProperty(reader, SCHEMA_ACCESS_STRATEGY, SCHEMA_TEXT_PROPERTY.getValue());
+ runner.setProperty(reader, SCHEMA_TEXT, schemaText);
+ runner.setProperty(reader, StandardProtobufReader.MESSAGE_NAME_RESOLUTION_STRATEGY,
+ StandardProtobufReader.MessageNameResolverStrategy.MESSAGE_NAME_PROPERTY.getValue());
+ runner.setProperty(reader, StandardProtobufReader.MESSAGE_NAME, messageName);
+ runner.enableControllerService(reader);
+ return reader;
+ }
+
+ private StandardProtobufWriter createAndEnableWriter(final String id, final String schemaText, final String messageName) throws Exception {
+ final StandardProtobufWriter protobufWriter = new StandardProtobufWriter();
+ runner.addControllerService(id, protobufWriter);
+ runner.setProperty(protobufWriter, SCHEMA_ACCESS_STRATEGY, SCHEMA_TEXT_PROPERTY.getValue());
+ runner.setProperty(protobufWriter, SCHEMA_TEXT, schemaText);
+ runner.setProperty(protobufWriter, StandardProtobufWriter.MESSAGE_NAME_RESOLUTION_STRATEGY,
+ StandardProtobufWriter.MessageNameResolverStrategy.MESSAGE_NAME_PROPERTY.getValue());
+ runner.setProperty(protobufWriter, StandardProtobufWriter.MESSAGE_NAME, messageName);
+ runner.enableControllerService(protobufWriter);
+ return protobufWriter;
+ }
+
+ private byte[] writeSingleRecordWith(final StandardProtobufWriter protobufWriter, final MapRecord record) throws IOException, SchemaNotFoundException {
+ final RecordSchema writeSchema = protobufWriter.getSchema(emptyMap(), null);
+ final ByteArrayOutputStream out = new ByteArrayOutputStream();
+ try (RecordSetWriter recordSetWriter = protobufWriter.createWriter(runner.getLogger(), writeSchema, out, emptyMap())) {
+ recordSetWriter.beginRecordSet();
+ recordSetWriter.write(record);
+ recordSetWriter.finishRecordSet();
+ recordSetWriter.flush();
+ }
+ return out.toByteArray();
+ }
+
+ private MapRecord buildProto3Record() throws Exception {
+ final Schema schema = loadProto3TestSchema();
+ final RecordSchema recordSchema = new ProtoSchemaParser(schema).createSchema(PROTO_3_MESSAGE);
+ return new ProtobufDataConverter(schema, PROTO_3_MESSAGE, recordSchema, false, false)
+ .createRecord(generateInputDataForProto3());
+ }
+
+ private byte[] writeSingleRecord(final MapRecord record) throws IOException, SchemaNotFoundException {
+ return writeSingleRecordWith(writer, record);
+ }
+
+ private void assertProto3Message(final ByteArrayInputStream payload) throws IOException {
+ final Schema schema = loadProto3TestSchema();
+ final RecordSchema recordSchema = new ProtoSchemaParser(schema).createSchema(PROTO_3_MESSAGE);
+ final MapRecord record = new ProtobufDataConverter(schema, PROTO_3_MESSAGE, recordSchema, false, false)
+ .createRecord(payload);
+
+ assertEquals(true, record.getValue("booleanField"));
+ assertEquals("Test text", record.getValue("stringField"));
+ assertEquals(Integer.MAX_VALUE, record.getValue("int32Field"));
+ assertEquals(Long.MAX_VALUE, record.getValue("int64Field"));
+ assertArrayEquals("Test bytes".getBytes(), (byte[]) record.getValue("bytesField"));
+
+ final MapRecord nestedRecord = (MapRecord) record.getValue("nestedMessage");
+ assertEquals("ENUM_VALUE_3", nestedRecord.getValue("testEnum"));
+ }
+
+ private String getTestProto3File() {
+ return getSchemaFile("test_proto3.proto");
+ }
+
+ private String getSchemaFile(final String resourceName) {
+ try {
+ return new String(getClass().getClassLoader().getResourceAsStream(resourceName).readAllBytes());
+ } catch (final Exception e) {
+ throw new RuntimeException("Failed to read " + resourceName + " from resources", e);
+ }
+ }
+
+ static class FakeSchemaReferenceWriter extends AbstractControllerService implements SchemaReferenceWriter {
+ @Override
+ public void writeHeader(final RecordSchema recordSchema, final OutputStream outputStream) throws IOException {
+ outputStream.write(FAKE_HEADER);
+ }
+
+ @Override
+ public Map getAttributes(final RecordSchema recordSchema) {
+ return Map.of();
+ }
+
+ @Override
+ public void validateSchema(final RecordSchema recordSchema) {
+ }
+
+ @Override
+ public Set getRequiredSchemaFields() {
+ return emptySet();
+ }
+ }
+
+ static class FakeMessageIndexWriter extends AbstractControllerService implements MessageIndexWriter {
+ @Override
+ public void writeMessageIndex(final Map variables, final SchemaDefinition schemaDefinition, final MessageName messageName, final OutputStream outputStream) throws IOException {
+ outputStream.write(FAKE_INDEX);
+ }
+ }
+}
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
new file mode 100644
index 000000000000..56925f282116
--- /dev/null
+++ b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/java/org/apache/nifi/services/protobuf/TestStandardProtobufWriterConfluentRoundTrip.java
@@ -0,0 +1,235 @@
+/*
+ * 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 org.apache.nifi.confluent.schemaregistry.ConfluentEncodedSchemaReferenceReader;
+import org.apache.nifi.confluent.schemaregistry.ConfluentEncodedSchemaReferenceWriter;
+import org.apache.nifi.confluent.schemaregistry.ConfluentProtobufMessageIndexWriter;
+import org.apache.nifi.confluent.schemaregistry.ConfluentProtobufMessageNameResolver;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ControllerService;
+import org.apache.nifi.schema.access.SchemaField;
+import org.apache.nifi.schema.access.SchemaNotFoundException;
+import org.apache.nifi.schemaregistry.services.SchemaDefinition;
+import org.apache.nifi.schemaregistry.services.SchemaRegistry;
+import org.apache.nifi.schemaregistry.services.StandardSchemaDefinition;
+import org.apache.nifi.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordSetWriter;
+import org.apache.nifi.serialization.SimpleRecordSchema;
+import org.apache.nifi.serialization.record.MapRecord;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.serialization.record.RecordField;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.serialization.record.SchemaIdentifier;
+import org.apache.nifi.util.NoOpProcessor;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static java.util.Collections.emptyMap;
+import static java.util.Collections.emptySet;
+import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_ACCESS_STRATEGY;
+import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_NAME;
+import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_NAME_PROPERTY;
+import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_REFERENCE_READER;
+import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_REFERENCE_READER_PROPERTY;
+import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_REGISTRY;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * End-to-end Confluent wire-format tests: {@link StandardProtobufWriter} configured with the real
+ * {@link ConfluentEncodedSchemaReferenceWriter} and {@link ConfluentProtobufMessageIndexWriter} produces
+ * [header][message index][payload], with exact byte assertions on the header and index (including the
+ * single-byte [0] optimization and a nested-message index), and the content round-trips back through
+ * {@link StandardProtobufReader} using the matching {@link ConfluentEncodedSchemaReferenceReader} and
+ * {@link ConfluentProtobufMessageNameResolver}.
+ */
+class TestStandardProtobufWriterConfluentRoundTrip {
+
+ private static final int 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 String SCHEMA_TEXT = """
+ syntax = "proto3";
+ message User {
+ int32 id = 1;
+ string name = 2;
+ message Profile {
+ string bio = 1;
+ }
+ }
+ message Company {
+ string name = 1;
+ }""";
+
+ private TestRunner runner;
+ private MockSchemaRegistry schemaRegistry;
+ private ConfluentEncodedSchemaReferenceWriter referenceWriter;
+ private ConfluentProtobufMessageIndexWriter indexWriter;
+ private ConfluentEncodedSchemaReferenceReader referenceReader;
+ private ConfluentProtobufMessageNameResolver messageNameResolver;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ runner = TestRunners.newTestRunner(NoOpProcessor.class);
+
+ schemaRegistry = new MockSchemaRegistry();
+ referenceWriter = new ConfluentEncodedSchemaReferenceWriter();
+ indexWriter = new ConfluentProtobufMessageIndexWriter();
+ referenceReader = new ConfluentEncodedSchemaReferenceReader();
+ messageNameResolver = new ConfluentProtobufMessageNameResolver();
+
+ enableService("registry", schemaRegistry);
+ enableService("referenceWriter", referenceWriter);
+ enableService("indexWriter", indexWriter);
+ enableService("referenceReader", referenceReader);
+ enableService("messageNameResolver", messageNameResolver);
+ }
+
+ @Test
+ void testFirstRootMessageSingleByteIndexAndRoundTrip() throws Exception {
+ final Map values = new HashMap<>();
+ values.put("id", 7);
+ values.put("name", "Alice");
+ final MapRecord userRecord = new MapRecord(recordSchema("id", "name"), values);
+
+ final byte[] output = writeConfluent("User", 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]);
+
+ final Record readBack = readConfluent(output);
+ assertEquals(7, readBack.getValue("id"));
+ assertEquals("Alice", readBack.getValue("name"));
+ }
+
+ @Test
+ void testNestedMessageIndexAndRoundTrip() throws Exception {
+ final Map values = new HashMap<>();
+ values.put("bio", "hello");
+ final MapRecord profileRecord = new MapRecord(recordSchema("bio"), values);
+
+ final byte[] output = writeConfluent("User.Profile", 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));
+
+ final Record readBack = readConfluent(output);
+ assertEquals("hello", readBack.getValue("bio"));
+ }
+
+ private byte[] writeConfluent(final String messageName, final MapRecord record) throws Exception {
+ 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_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.enableControllerService(writer);
+
+ final RecordSchema writeSchema = writer.getSchema(emptyMap(), null);
+ final ByteArrayOutputStream out = new ByteArrayOutputStream();
+ try (RecordSetWriter recordSetWriter = writer.createWriter(runner.getLogger(), writeSchema, out, emptyMap())) {
+ recordSetWriter.beginRecordSet();
+ recordSetWriter.write(record);
+ recordSetWriter.finishRecordSet();
+ recordSetWriter.flush();
+ }
+ return out.toByteArray();
+ }
+
+ private Record readConfluent(final byte[] data) throws Exception {
+ 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, StandardProtobufReader.MESSAGE_NAME_RESOLUTION_STRATEGY,
+ StandardProtobufReader.MessageNameResolverStrategy.MESSAGE_NAME_RESOLVER.getValue());
+ runner.setProperty(reader, StandardProtobufReader.MESSAGE_NAME_RESOLVER, "messageNameResolver");
+ runner.enableControllerService(reader);
+
+ final RecordReader recordReader = reader.createRecordReader(emptyMap(), new ByteArrayInputStream(data), data.length, runner.getLogger());
+ final Record record = recordReader.nextRecord();
+ assertNull(recordReader.nextRecord());
+ return record;
+ }
+
+ 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;
+ fields[i] = new RecordField(fieldNames[i], type.getDataType());
+ }
+ return new SimpleRecordSchema(List.of(fields));
+ }
+
+ private void enableService(final String id, final ControllerService service) throws Exception {
+ runner.addControllerService(id, service);
+ 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);
+ return out;
+ }
+
+ /**
+ * Returns a fixed Protobuf SchemaDefinition (with a numeric schema id and version) for any identifier, so that
+ * 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,
+ SchemaDefinition.SchemaType.PROTOBUF);
+
+ @Override
+ public RecordSchema retrieveSchema(final SchemaIdentifier schemaIdentifier) {
+ throw new UnsupportedOperationException("retrieveSchema is not used in this test");
+ }
+
+ @Override
+ public SchemaDefinition retrieveSchemaDefinition(final SchemaIdentifier schemaIdentifier) throws SchemaNotFoundException {
+ return schemaDefinition;
+ }
+
+ @Override
+ public Set getSuppliedSchemaFields() {
+ return emptySet();
+ }
+ }
+}
diff --git a/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/java/org/apache/nifi/services/protobuf/converter/TestProtobufDataSerializer.java b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/java/org/apache/nifi/services/protobuf/converter/TestProtobufDataSerializer.java
new file mode 100644
index 000000000000..24d8e36fe405
--- /dev/null
+++ b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/java/org/apache/nifi/services/protobuf/converter/TestProtobufDataSerializer.java
@@ -0,0 +1,151 @@
+/*
+ * 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.converter;
+
+import com.google.protobuf.Descriptors.DescriptorValidationException;
+import com.squareup.wire.schema.Schema;
+import org.apache.nifi.serialization.record.MapRecord;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.services.protobuf.ProtoTestUtil;
+import org.apache.nifi.services.protobuf.schema.ProtoSchemaParser;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.math.BigInteger;
+import java.util.Map;
+
+import static org.apache.nifi.services.protobuf.ProtoTestUtil.generateInputDataForRootMessage;
+import static org.apache.nifi.services.protobuf.ProtoTestUtil.loadProto3TestSchema;
+import static org.apache.nifi.services.protobuf.ProtoTestUtil.loadRepeatedProto3TestSchema;
+import static org.apache.nifi.services.protobuf.ProtoTestUtil.loadRootMessageSchema;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * Validates {@link ProtobufDataSerializer} by round-tripping through the reader-side
+ * {@link ProtobufDataConverter}: a Record is produced from known protobuf bytes, re-serialized by
+ * the serializer, and parsed back, asserting the values survive the round trip losslessly. Runs
+ * without the NiFi framework (no TestRunner).
+ */
+public class TestProtobufDataSerializer {
+
+ @Test
+ public void testSerializeProto3RoundTrip() throws DescriptorValidationException, IOException {
+ final Schema schema = loadProto3TestSchema();
+ final RecordSchema recordSchema = new ProtoSchemaParser(schema).createSchema("Proto3Message");
+
+ final MapRecord originalRecord = new ProtobufDataConverter(schema, "Proto3Message", recordSchema, false, false)
+ .createRecord(ProtoTestUtil.generateInputDataForProto3());
+
+ final byte[] serialized = new ProtobufDataSerializer(schema, "Proto3Message").serialize(originalRecord);
+
+ final MapRecord record = new ProtobufDataConverter(schema, "Proto3Message", recordSchema, false, false)
+ .createRecord(new ByteArrayInputStream(serialized));
+
+ assertEquals(true, record.getValue("booleanField"));
+ assertEquals("Test text", record.getValue("stringField"));
+ assertEquals(Integer.MAX_VALUE, record.getValue("int32Field"));
+ assertEquals(4294967295L, record.getValue("uint32Field"));
+ assertEquals(Integer.MIN_VALUE, record.getValue("sint32Field"));
+ assertEquals(4294967294L, record.getValue("fixed32Field"));
+ assertEquals(Integer.MAX_VALUE, record.getValue("sfixed32Field"));
+ assertEquals(Double.MAX_VALUE, record.getValue("doubleField"));
+ assertEquals(Float.MAX_VALUE, record.getValue("floatField"));
+ assertArrayEquals("Test bytes".getBytes(), (byte[]) record.getValue("bytesField"));
+ assertEquals(Long.MAX_VALUE, record.getValue("int64Field"));
+ assertEquals(new BigInteger("18446744073709551615"), record.getValue("uint64Field"));
+ assertEquals(Long.MIN_VALUE, record.getValue("sint64Field"));
+ assertEquals(new BigInteger("18446744073709551614"), record.getValue("fixed64Field"));
+ assertEquals(Long.MAX_VALUE, record.getValue("sfixed64Field"));
+
+ final MapRecord nestedRecord = (MapRecord) record.getValue("nestedMessage");
+ assertEquals("ENUM_VALUE_3", nestedRecord.getValue("testEnum"));
+
+ final Object[] recordList = (Object[]) nestedRecord.getValue("nestedMessage2");
+ assertEquals(1, recordList.length);
+
+ final MapRecord nestedRecord2 = (MapRecord) recordList[0];
+ assertEquals(Map.of("test_key_entry1", 101, "test_key_entry2", 202), nestedRecord2.getValue("testMap"));
+
+ // Only one field is set in the OneOf field
+ assertNull(nestedRecord2.getValue("stringOption"));
+ assertNull(nestedRecord2.getValue("booleanOption"));
+ assertEquals(3, nestedRecord2.getValue("int32Option"));
+ }
+
+ @Test
+ public void testSerializeNestedMessageRoundTrip() throws DescriptorValidationException, IOException {
+ final Schema schema = loadRootMessageSchema();
+ final String messageName = "org.apache.nifi.protobuf.test.RootMessage";
+ final RecordSchema recordSchema = new ProtoSchemaParser(schema).createSchema(messageName);
+
+ final MapRecord originalRecord = new ProtobufDataConverter(schema, messageName, recordSchema, false, false)
+ .createRecord(generateInputDataForRootMessage());
+
+ final byte[] serialized = new ProtobufDataSerializer(schema, messageName).serialize(originalRecord);
+
+ final MapRecord record = new ProtobufDataConverter(schema, messageName, recordSchema, false, false)
+ .createRecord(new ByteArrayInputStream(serialized));
+
+ final MapRecord nestedRecord = (MapRecord) record.getValue("nestedMessage");
+ assertEquals("ENUM_VALUE_3", nestedRecord.getValue("testEnum"));
+ }
+
+ @Test
+ public void testSerializeRepeatedProto3RoundTrip() throws DescriptorValidationException, IOException {
+ final Schema schema = loadRepeatedProto3TestSchema();
+ final RecordSchema recordSchema = new ProtoSchemaParser(schema).createSchema("RootMessage");
+
+ final MapRecord originalRecord = new ProtobufDataConverter(schema, "RootMessage", recordSchema, false, false)
+ .createRecord(ProtoTestUtil.generateInputDataForRepeatedProto3());
+
+ final byte[] serialized = new ProtobufDataSerializer(schema, "RootMessage").serialize(originalRecord);
+
+ final MapRecord record = new ProtobufDataConverter(schema, "RootMessage", recordSchema, false, false)
+ .createRecord(new ByteArrayInputStream(serialized));
+
+ final Object[] repeatedMessage = (Object[]) record.getValue("repeatedMessage");
+ final MapRecord record1 = (MapRecord) repeatedMessage[0];
+
+ assertArrayEquals(new Object[]{true, false}, (Object[]) record1.getValue("booleanField"));
+ assertArrayEquals(new Object[]{"Test text1", "Test text2"}, (Object[]) record1.getValue("stringField"));
+ assertArrayEquals(new Object[]{Integer.MAX_VALUE, Integer.MAX_VALUE - 1}, (Object[]) record1.getValue("int32Field"));
+ assertArrayEquals(new Object[]{4294967295L, 4294967294L}, (Object[]) record1.getValue("uint32Field"));
+ assertArrayEquals(new Object[]{Integer.MIN_VALUE, Integer.MIN_VALUE + 1}, (Object[]) record1.getValue("sint32Field"));
+ assertArrayEquals(new Object[]{4294967294L, 4294967293L}, (Object[]) record1.getValue("fixed32Field"));
+ assertArrayEquals(new Object[]{Integer.MAX_VALUE, Integer.MAX_VALUE - 1}, (Object[]) record1.getValue("sfixed32Field"));
+ assertArrayEquals(new Object[]{Double.MAX_VALUE, Double.MAX_VALUE - 1}, (Object[]) record1.getValue("doubleField"));
+ assertArrayEquals(new Object[]{Float.MAX_VALUE, Float.MAX_VALUE - 1}, (Object[]) record1.getValue("floatField"));
+ assertArrayEquals(new Object[]{Long.MAX_VALUE, Long.MAX_VALUE - 1}, (Object[]) record1.getValue("int64Field"));
+ assertArrayEquals(new Object[]{Long.MIN_VALUE, Long.MIN_VALUE + 1}, (Object[]) record1.getValue("sint64Field"));
+ assertArrayEquals(new Object[]{Long.MAX_VALUE, Long.MAX_VALUE - 1}, (Object[]) record1.getValue("sfixed64Field"));
+ assertArrayEquals(new Object[]{"ENUM_VALUE_2", "ENUM_VALUE_3"}, (Object[]) record1.getValue("testEnum"));
+
+ final Object[] uint64FieldValues = (Object[]) record1.getValue("uint64Field");
+ assertEquals(new BigInteger("18446744073709551615"), uint64FieldValues[0]);
+ assertEquals(new BigInteger("18446744073709551614"), uint64FieldValues[1]);
+
+ final Object[] bytesFieldValues = (Object[]) record1.getValue("bytesField");
+ assertArrayEquals("Test bytes1".getBytes(), (byte[]) bytesFieldValues[0]);
+ assertArrayEquals("Test bytes2".getBytes(), (byte[]) bytesFieldValues[1]);
+
+ final MapRecord record2 = (MapRecord) repeatedMessage[1];
+ assertArrayEquals(new Object[]{true}, (Object[]) record2.getValue("booleanField"));
+ }
+}
diff --git a/nifi-extension-bundles/nifi-standard-services/nifi-schema-registry-service-api/src/main/java/org/apache/nifi/schemaregistry/services/MessageIndexWriter.java b/nifi-extension-bundles/nifi-standard-services/nifi-schema-registry-service-api/src/main/java/org/apache/nifi/schemaregistry/services/MessageIndexWriter.java
new file mode 100644
index 000000000000..3d3d2e83da12
--- /dev/null
+++ b/nifi-extension-bundles/nifi-standard-services/nifi-schema-registry-service-api/src/main/java/org/apache/nifi/schemaregistry/services/MessageIndexWriter.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.schemaregistry.services;
+
+import org.apache.nifi.controller.ControllerService;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Map;
+
+/**
+ * An interface for writing message index information that locates a specific message within
+ * a schema definition containing multiple message declarations.
+ *
+ * This is the write-side counterpart to {@link MessageNameResolver}: given a message name that
+ * has already been determined, it encodes the path to that message within the schema definition
+ * and writes it to an output stream, enabling a reader to later resolve the same message name
+ * from the encoded path.
+ *
+ */
+public interface MessageIndexWriter extends ControllerService {
+
+ /**
+ * Writes the message index identifying the given message name within the provided schema
+ * definition to the provided output stream.
+ *
+ * This method analyzes the given schema definition to determine the path of the target
+ * message and encodes that path to the output stream. The encoding strategy depends on the
+ * specific implementation and may involve navigating nested message declarations or
+ * consulting schema metadata.
+ *
+ *
+ * @param variables additional variables that may influence the encoding process, such as context-specific information
+ * @param schemaDefinition the schema definition containing schema information and metadata
+ * @param messageName the name of the message whose index should be written
+ * @param outputStream the output stream to which the message index should be written
+ * @throws IOException if an I/O error occurs while writing to the output stream or processing the schema
+ */
+ void writeMessageIndex(final Map variables, final SchemaDefinition schemaDefinition, final MessageName messageName, final OutputStream outputStream) throws IOException;
+}