You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by pv...@apache.org on 2026/09/29 12:27:02 UTC
(nifi) branch main updated: NIFI-14210 Add record writer for Protobuf messages (#11499)
This is an automated email from the ASF dual-hosted git repository.
pvillard31 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new cfe0532888d NIFI-14210 Add record writer for Protobuf messages (#11499)
cfe0532888d is described below
commit cfe0532888da82abc728acf3822d7d79ffd5bbff
Author: Natalia Khodiakova <na...@gmail.com>
AuthorDate: Tue Sep 29 14:26:54 2026 +0200
NIFI-14210 Add record writer for Protobuf messages (#11499)
---
.../nifi-confluent-platform-nar/pom.xml | 5 +
.../apache/nifi/confluent/schema}/VarintUtils.java | 12 +-
.../nifi/confluent/schema}/VarintUtilsTest.java | 10 +-
.../pom.xml | 50 +-
.../ConfluentProtobufMessageIndexWriter.java | 84 +++
.../ProtobufMessageIndexEncoder.java | 113 ++++
.../org.apache.nifi.controller.ControllerService | 3 +-
.../ConfluentProtobufMessageIndexWriterTest.java | 111 ++++
.../ProtobufMessageIndexEncoderTest.java | 223 ++++++++
.../ConfluentProtobufMessageNameResolver.java | 6 +-
.../ConfluentProtobufMessageNameResolverTest.java | 2 +-
.../nifi-confluent-platform-bundle/pom.xml | 1 +
.../nifi-protobuf-services/pom.xml | 19 +
.../services/protobuf/ProtobufSchemaCompiler.java | 46 +-
.../services/protobuf/ProtobufSchemaValidator.java | 54 +-
.../services/protobuf/StandardProtobufWriter.java | 480 +++++++++++++++++
.../WriteProtobufResultWithExternalSchema.java | 121 +++++
.../protobuf/converter/ProtobufDataSerializer.java | 343 ++++++++++++
.../org.apache.nifi.controller.ControllerService | 3 +-
.../protobuf/TestProtobufSchemaCompiler.java | 79 +++
.../protobuf/TestStandardProtobufReader.java | 33 +-
.../protobuf/TestStandardProtobufWriter.java | 586 +++++++++++++++++++++
...stStandardProtobufWriterConfluentRoundTrip.java | 252 +++++++++
.../converter/TestProtobufDataSerializer.java | 334 ++++++++++++
.../services/MessageIndexWriter.java | 54 ++
25 files changed, 2939 insertions(+), 85 deletions(-)
diff --git a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-platform-nar/pom.xml b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-platform-nar/pom.xml
index 37f0136af1f..3ed940f72e7 100644
--- a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-platform-nar/pom.xml
+++ b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-platform-nar/pom.xml
@@ -35,5 +35,10 @@
<artifactId>nifi-confluent-protobuf-message-name-resolver</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-confluent-protobuf-message-index-writer</artifactId>
+ <version>${project.version}</version>
+ </dependency>
</dependencies>
</project>
diff --git a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-name-resolver/src/main/java/org/apache/nifi/confluent/schemaregistry/VarintUtils.java b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-antlr-parser/src/main/java/org/apache/nifi/confluent/schema/VarintUtils.java
similarity index 92%
rename from nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-name-resolver/src/main/java/org/apache/nifi/confluent/schemaregistry/VarintUtils.java
rename to nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-antlr-parser/src/main/java/org/apache/nifi/confluent/schema/VarintUtils.java
index 209a5e9864c..44b47bc4010 100644
--- a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-name-resolver/src/main/java/org/apache/nifi/confluent/schemaregistry/VarintUtils.java
+++ b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-antlr-parser/src/main/java/org/apache/nifi/confluent/schema/VarintUtils.java
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.nifi.confluent.schemaregistry;
+package org.apache.nifi.confluent.schema;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
@@ -38,7 +38,7 @@ import java.io.InputStream;
* <a href="https://developers.google.com/protocol-buffers/docs/encoding#varints">Protocol Buffers Encoding</a></p>
*
*/
-final class VarintUtils {
+public final class VarintUtils {
private VarintUtils() {
}
@@ -50,7 +50,7 @@ final class VarintUtils {
* @return the decoded varint value
* @throws IOException if unable to read from stream or invalid varint format
*/
- static int readVarintFromStream(final InputStream inputStream) throws IOException {
+ public static int readVarintFromStream(final InputStream inputStream) throws IOException {
final int firstByte = inputStream.read();
if (firstByte == -1) {
throw new IOException("Unexpected end of stream while reading varint");
@@ -66,7 +66,7 @@ final class VarintUtils {
* @return the decoded varint value
* @throws IOException if unable to read from stream or invalid varint format
*/
- static int readVarintFromStreamAfterFirstByteConsumed(final InputStream inputStream, final int firstByte) throws IOException {
+ public static int readVarintFromStreamAfterFirstByteConsumed(final InputStream inputStream, final int firstByte) throws IOException {
// accumulated result
int value = 0;
@@ -115,7 +115,7 @@ final class VarintUtils {
* @param encodedValue the zigzag encoded value
* @return the decoded integer value
*/
- static int decodeZigZag(final int encodedValue) {
+ public static int decodeZigZag(final int encodedValue) {
return (encodedValue >>> 1) ^ -(encodedValue & 1);
}
@@ -127,7 +127,7 @@ final class VarintUtils {
* @param value the integer value to encode
* @return byte array containing the zigzag encoded varint
*/
- static byte[] writeZigZagVarint(final int value) {
+ public static byte[] writeZigZagVarint(final int value) {
final ByteArrayOutputStream output = new ByteArrayOutputStream(4);
// Zigzag encode
diff --git a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-name-resolver/src/test/java/org/apache/nifi/confluent/schemaregistry/VarintUtilsTest.java b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-antlr-parser/src/test/java/org/apache/nifi/confluent/schema/VarintUtilsTest.java
similarity index 95%
rename from nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-name-resolver/src/test/java/org/apache/nifi/confluent/schemaregistry/VarintUtilsTest.java
rename to nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-antlr-parser/src/test/java/org/apache/nifi/confluent/schema/VarintUtilsTest.java
index 0c343864e71..71bc97178f9 100644
--- a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-name-resolver/src/test/java/org/apache/nifi/confluent/schemaregistry/VarintUtilsTest.java
+++ b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-antlr-parser/src/test/java/org/apache/nifi/confluent/schema/VarintUtilsTest.java
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.nifi.confluent.schemaregistry;
+package org.apache.nifi.confluent.schema;
import org.junit.jupiter.api.Test;
@@ -23,10 +23,10 @@ import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
-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.schemaregistry.VarintUtils.writeZigZagVarint;
+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;
+import static org.apache.nifi.confluent.schema.VarintUtils.writeZigZagVarint;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
diff --git a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-platform-nar/pom.xml b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-index-writer/pom.xml
similarity index 58%
copy from nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-platform-nar/pom.xml
copy to nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-index-writer/pom.xml
index 37f0136af1f..55f37e1ba6f 100644
--- a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-platform-nar/pom.xml
+++ b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-index-writer/pom.xml
@@ -1,39 +1,57 @@
<?xml version="1.0" encoding="UTF-8"?>
-<!-- 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
+<!-- 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. -->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
+
<parent>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-confluent-platform-bundle</artifactId>
<version>2.13.0-SNAPSHOT</version>
</parent>
- <artifactId>nifi-confluent-platform-nar</artifactId>
- <packaging>nar</packaging>
+
+ <artifactId>nifi-confluent-protobuf-message-index-writer</artifactId>
+ <packaging>jar</packaging>
+ <description>Confluent Protobuf Message Index Writer for NiFi</description>
+
<dependencies>
<dependency>
<groupId>org.apache.nifi</groupId>
- <artifactId>nifi-standard-shared-nar</artifactId>
- <version>${project.version}</version>
- <type>nar</type>
+ <artifactId>nifi-schema-registry-service-api</artifactId>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
- <artifactId>nifi-confluent-schema-registry-service</artifactId>
+ <artifactId>nifi-confluent-platform-schema-api</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
- <artifactId>nifi-confluent-protobuf-message-name-resolver</artifactId>
+ <artifactId>nifi-confluent-protobuf-antlr-parser</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>com.github.ben-manes.caffeine</groupId>
+ <artifactId>caffeine</artifactId>
+ </dependency>
+
+ <!-- Test dependencies -->
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-record</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-index-writer/src/main/java/org/apache/nifi/confluent/schemaregistry/ConfluentProtobufMessageIndexWriter.java b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-index-writer/src/main/java/org/apache/nifi/confluent/schemaregistry/ConfluentProtobufMessageIndexWriter.java
new file mode 100644
index 00000000000..b8777d56dc3
--- /dev/null
+++ b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-index-writer/src/main/java/org/apache/nifi/confluent/schemaregistry/ConfluentProtobufMessageIndexWriter.java
@@ -0,0 +1,84 @@
+/*
+ * 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 com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnDisabled;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.confluent.schema.AntlrProtobufMessageSchemaParser;
+import org.apache.nifi.confluent.schema.ProtobufMessageSchema;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.schemaregistry.services.MessageIndexWriter;
+import org.apache.nifi.schemaregistry.services.MessageName;
+import org.apache.nifi.schemaregistry.services.SchemaDefinition;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+
+@Tags({"confluent", "schema", "registry", "protobuf", "message", "index", "writer"})
+@CapabilityDescription("""
+ Writes Protobuf message index information in Confluent Schema Registry wire format by encoding the declaration-order path to a given message name within the schema definition.
+ For Confluent wire format reference see: https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/index.html#wire-format
+ """)
+public class ConfluentProtobufMessageIndexWriter extends AbstractControllerService implements MessageIndexWriter {
+
+ private static final int MAXIMUM_CACHE_SIZE = 1000;
+ private static final int CACHE_EXPIRE_HOURS = 1;
+
+ private Cache<EncodeMessageIndexArguments, byte[]> messageIndexCache;
+
+ @OnEnabled
+ public void onEnabled(final ConfigurationContext context) {
+ messageIndexCache = Caffeine.newBuilder().maximumSize(MAXIMUM_CACHE_SIZE).expireAfterWrite(Duration.ofHours(CACHE_EXPIRE_HOURS)).build();
+ }
+
+ @OnDisabled
+ public void onDisabled(final ConfigurationContext context) {
+ if (messageIndexCache != null) {
+ messageIndexCache.invalidateAll();
+ messageIndexCache = null;
+ }
+ }
+
+ @Override
+ public void writeMessageIndex(final Map<String, String> variables, final SchemaDefinition schemaDefinition, final MessageName messageName, final OutputStream outputStream) throws IOException {
+ final EncodeMessageIndexArguments encodeMessageIndexArguments = new EncodeMessageIndexArguments(schemaDefinition, messageName);
+ final byte[] encodedMessageIndex = messageIndexCache.get(encodeMessageIndexArguments, this::encodeMessageIndex);
+ outputStream.write(encodedMessageIndex);
+ }
+
+ private byte[] encodeMessageIndex(final EncodeMessageIndexArguments encodeMessageIndexArguments) {
+ try {
+ final String schemaText = encodeMessageIndexArguments.schemaDefinition().getText();
+ final AntlrProtobufMessageSchemaParser parser = new AntlrProtobufMessageSchemaParser();
+ final List<ProtobufMessageSchema> rootMessages = parser.parse(schemaText);
+ return ProtobufMessageIndexEncoder.encode(rootMessages, encodeMessageIndexArguments.messageName());
+ } catch (final Exception e) {
+ throw new IllegalStateException("Failed to generate Protobuf message index", 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 00000000000..581493b1044
--- /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.
+ * <p>
+ * <a href="https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/index.html#wire-format">See the Confluent protobuf wire format.</a>
+ * <p>
+ * 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<ProtobufMessageSchema> rootMessages, final MessageName messageName) {
+ final List<Integer> messageIndexPath = findMessageIndexPath(rootMessages, messageName);
+
+ if (messageIndexPath.size() == 1 && messageIndexPath.getFirst() == 0) {
+ return FIRST_ROOT_MESSAGE_INDEX.clone();
+ }
+
+ 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<Integer> findMessageIndexPath(final List<ProtobufMessageSchema> 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<Integer> 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<Integer> descend(final ProtobufMessageSchema currentMessage, final String currentFullyQualifiedName, final String targetFullyQualifiedName) {
+ if (currentFullyQualifiedName.equals(targetFullyQualifiedName)) {
+ return new ArrayList<>();
+ }
+
+ final List<ProtobufMessageSchema> 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<Integer> messageIndexPath = descend(child, childFullyQualifiedName, targetFullyQualifiedName);
+ if (messageIndexPath != null) {
+ messageIndexPath.addFirst(childIndex);
+ return messageIndexPath;
+ }
+ }
+
+ return null;
+ }
+}
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-confluent-platform-bundle/nifi-confluent-protobuf-message-index-writer/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService
similarity index 88%
copy from nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService
copy to nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-index-writer/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService
index 8ff7783b668..7c9bba67298 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-confluent-platform-bundle/nifi-confluent-protobuf-message-index-writer/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService
@@ -13,5 +13,4 @@
# See the License for the specific language governing permissions and
# 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.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 00000000000..d8de2699ca3
--- /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<Arguments> 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 00000000000..bef1a96412d
--- /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,223 @@
+/*
+ * 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 final String NON_ZERO_NESTED_INDEX_SCHEMA = """
+ syntax = "proto3";
+ package com.example.nested;
+ message Unused {}
+ message Root {
+ message ChildZero {}
+ message ChildOne {}
+ message ChildTwo {
+ message GrandchildZero {}
+ message GrandchildOne {}
+ message GrandchildTwo {}
+ message GrandchildThree {}
+ }
+ }""";
+
+ private static Stream<Arguments> 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}),
+
+ Arguments.of(NON_ZERO_NESTED_INDEX_SCHEMA, new StandardMessageName(Optional.of("com.example.nested"), "Root.ChildTwo.GrandchildThree"), new int[] {1, 2, 3})
+ );
+ }
+
+ @ParameterizedTest
+ @MethodSource("provideMessageIndexTestCases")
+ void testEncode(final String schemaText, final MessageName messageName, final int[] expectedIndexes) throws IOException {
+ final List<ProtobufMessageSchema> 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<ProtobufMessageSchema> 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 testEncodeFirstRootMessageReturnsIndependentArrays() {
+ final List<ProtobufMessageSchema> rootMessages = new AntlrProtobufMessageSchemaParser().parse(DEFAULT_PACKAGE_SCHEMA);
+ final MessageName messageName = new StandardMessageName(Optional.empty(), "User");
+
+ final byte[] first = ProtobufMessageIndexEncoder.encode(rootMessages, messageName);
+ first[0] = 0x7F;
+
+ assertArrayEquals(new byte[] {0x00}, ProtobufMessageIndexEncoder.encode(rootMessages, messageName));
+ }
+
+ @Test
+ void testEncodeMessageNotFoundThrows() {
+ final List<ProtobufMessageSchema> rootMessages = new AntlrProtobufMessageSchemaParser().parse(DEFAULT_PACKAGE_SCHEMA);
+
+ assertThrows(IllegalStateException.class,
+ () -> ProtobufMessageIndexEncoder.encode(rootMessages, new StandardMessageName(Optional.empty(), "DoesNotExist")));
+ }
+
+ @Test
+ void testEncodeNamespaceMismatchThrows() {
+ final List<ProtobufMessageSchema> 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<ProtobufMessageSchema> 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<ProtobufMessageSchema> 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<Integer> 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<Integer> 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 25bb3565aa3..67452f54079 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 java.util.stream.Collectors;
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 1c8aeb67a7d..50bca7695f0 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.nio.ByteBuffer;
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 142ca1bdd2b..0d8d6b41f4c 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 @@
<modules>
<module>nifi-confluent-schema-registry-service</module>
<module>nifi-confluent-protobuf-message-name-resolver</module>
+ <module>nifi-confluent-protobuf-message-index-writer</module>
<module>nifi-confluent-platform-nar</module>
<module>nifi-confluent-protobuf-antlr-parser</module>
<module>nifi-confluent-platform-schema-api</module>
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 9a8c5c0862b..045d7b46fcb 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 @@
<artifactId>nifi-mock</artifactId>
<scope>test</scope>
</dependency>
+ <!-- Confluent components used only for end-to-end write/read round-trip tests of the Confluent wire format -->
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-confluent-schema-registry-service</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-confluent-protobuf-message-index-writer</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-confluent-protobuf-message-name-resolver</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
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 6e264a19292..63423ce6d34 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
@@ -44,7 +44,7 @@ import java.util.concurrent.TimeUnit;
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.
@@ -115,8 +115,8 @@ final class ProtobufSchemaCompiler {
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 {
@@ -183,8 +183,7 @@ final class ProtobufSchemaCompiler {
}
/**
- * 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
@@ -192,14 +191,33 @@ final class ProtobufSchemaCompiler {
*/
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 this method
+ * creates the enclosing directory structure before writing the file.
+ *
+ * @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);
+ }
- final String preparedSchemaDefinition = ProtobufCustomOptionRemover.remove(schemaDefinition.getText(), Location.get(schemaFileName), logger);
- Files.write(schemaFile, preparedSchemaDefinition.getBytes(StandardCharsets.UTF_8), CREATE, WRITE, TRUNCATE_EXISTING);
- logger.debug("Successfully wrote schema to file: {} (string length: {})",
- schemaFile, preparedSchemaDefinition.length());
+ final String preparedSchemaText = ProtobufCustomOptionRemover.remove(schemaText, Location.get(relativePath), logger);
+ Files.write(schemaFile, preparedSchemaText.getBytes(StandardCharsets.UTF_8), CREATE, WRITE, TRUNCATE_EXISTING);
+ logger.debug("Successfully wrote schema to file: {} (string length: {})", schemaFile, preparedSchemaText.length());
}
/**
@@ -231,8 +249,10 @@ final class ProtobufSchemaCompiler {
logger.debug("Processing schema reference [{}] Identifier [{}]",
referenceKey, referencedSchema.getIdentifier());
- // Write referenced schema to appropriate directory
- writeSchemaToTempDirectory(tempDir, referencedSchema);
+ // The reference key is the path used in the import statement of the referencing schema, so the referenced
+ // schema has to be written to that exact path for the compiler to resolve the import. It is unrelated to
+ // the identifier of the referenced schema, which carries the name it is registered under.
+ writeSchemaFile(tempDir, referenceKey, referencedSchema.getText());
// Process nested references recursively
if (!referencedSchema.getReferences().isEmpty()) {
diff --git a/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/ProtobufSchemaValidator.java b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/ProtobufSchemaValidator.java
index cbde4d37ada..9d1a1c71b5f 100644
--- a/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/ProtobufSchemaValidator.java
+++ b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/main/java/org/apache/nifi/services/protobuf/ProtobufSchemaValidator.java
@@ -17,47 +17,45 @@
package org.apache.nifi.services.protobuf;
import org.apache.nifi.schemaregistry.services.SchemaDefinition;
-import org.apache.nifi.serialization.record.SchemaIdentifier;
+
+import java.util.Map;
/**
- * Validates Protocol Buffer SchemaDefinition objects and schema identifiers.
+ * Validates the schema references of Protocol Buffer SchemaDefinition objects.
*/
final class ProtobufSchemaValidator {
+ private static final String PROTO_EXTENSION = ".proto";
+
private ProtobufSchemaValidator() {
}
/**
- * Validates that all SchemaDefinition identifiers end with .proto extension.
- * Performs recursive validation on all referenced schemas.
+ * Validates that every schema reference, at any depth, is keyed by a path ending in the .proto extension.
+ * <p>
+ * A reference is keyed by the path used in the import statement of the referencing schema, for example
+ * {@code airlines/ph/cdm/shared.proto}, and the referenced schema is written to exactly that path so the import
+ * resolves. The extension is required because the compiler only discovers files named {@code *.proto}; without it
+ * the schema would fail to compile later with an unresolved import that does not indicate the cause.
+ * <p>
+ * The identifier of a referenced schema is deliberately not validated. It identifies the schema
+ * in its source registry and is unrelated to the Protobuf import path, so it is not required to
+ * have a .proto suffix.
*
- * @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<String, SchemaDefinition> 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 00000000000..dc39351c6e5
--- /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,480 @@
+/*
+ * 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.MessageType;
+import com.squareup.wire.schema.Schema;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+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.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+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.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.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.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.Collection;
+import java.util.EnumSet;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+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"})
+@SeeAlso(StandardProtobufReader.class)
+@CapabilityDescription("""
+ Serializes NiFi Records into Protocol Buffers binary format. \
+ Supports inline schema text and schema registry lookup for determining the Proto schema. \
+ Optional Schema Reference Writer and Message Index Writer Controller Services can add framing before the \
+ Protobuf payload. When both are configured, schema reference information is written first, followed by message \
+ index information. Selected implementations must be compatible with each other and with the target wire format. \
+ Confluent Protobuf wire format requires both compatible services. \
+ The target Proto message name can be determined statically using the 'Message Name' property or dynamically \
+ using a Message Name Resolver service. \
+ A single record is written per FlowFile because concatenated Protocol Buffers messages cannot be delimited. \
+ The 'google.protobuf.Any' well-known type is not expanded on write; a Record derived from an Any-typed message \
+ is serialized as an ordinary nested message rather than being re-wrapped as an Any.""")
+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("The Controller Service used to write schema reference information before any message index information "
+ + "and the Protobuf payload. The selected implementation must be compatible with the target wire format and any "
+ + "configured Message Index Writer. When not configured, no schema reference information is written.")
+ .required(false)
+ .identifiesControllerService(SchemaReferenceWriter.class)
+ .build();
+
+ public static final PropertyDescriptor MESSAGE_INDEX_WRITER = new PropertyDescriptor.Builder()
+ .name("Message Index Writer")
+ .description("The Controller Service used to write information identifying the selected message within the Protobuf "
+ + "schema. Message index information is written after any schema reference information and before the Protobuf "
+ + "payload. The selected implementation must be compatible with the target wire format and any configured "
+ + "Schema Reference Writer.")
+ .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 volatile ProtobufSchemaCompiler schemaCompiler;
+ private volatile MessageNameResolver messageNameResolver;
+ private volatile SchemaReferenceWriter schemaReferenceWriter;
+ private volatile MessageIndexWriter messageIndexWriter;
+ private volatile SchemaRegistry schemaRegistry;
+ private volatile String schemaAccessStrategyValue;
+ private volatile PropertyValue schemaText;
+ private volatile PropertyValue schemaName;
+ private volatile PropertyValue schemaBranchName;
+ private volatile PropertyValue schemaVersion;
+
+ @OnEnabled
+ public void onEnabled(final ConfigurationContext context) {
+ super.storeSchemaAccessStrategy(context);
+ setupMessageNameResolver(context);
+ schemaAccessStrategyValue = context.getProperty(SCHEMA_ACCESS_STRATEGY).getValue();
+ schemaRegistry = context.getProperty(SCHEMA_REGISTRY).asControllerService(SchemaRegistry.class);
+ schemaReferenceWriter = context.getProperty(SCHEMA_REFERENCE_WRITER).asControllerService(SchemaReferenceWriter.class);
+ messageIndexWriter = context.getProperty(MESSAGE_INDEX_WRITER).asControllerService(MessageIndexWriter.class);
+ schemaName = context.getProperty(SCHEMA_NAME);
+ schemaText = context.getProperty(SCHEMA_TEXT);
+ schemaBranchName = context.getProperty(SCHEMA_BRANCH_NAME);
+ schemaVersion = context.getProperty(SCHEMA_VERSION);
+ }
+
+ @Override
+ protected void init(final ControllerServiceInitializationContext config) throws InitializationException {
+ super.init(config);
+ schemaCompiler = new ProtobufSchemaCompiler(getIdentifier(), getLogger());
+ }
+
+ @Override
+ public RecordSchema getSchema(final Map<String, String> variables, final RecordSchema readSchema) throws SchemaNotFoundException, IOException {
+ return createWriteContext(variables, null).recordSchema();
+ }
+
+ @Override
+ public RecordSetWriter createWriter(final ComponentLog logger, final RecordSchema schema, final OutputStream out, final Map<String, String> variables) throws SchemaNotFoundException, IOException {
+ // Resolve the schema against the identifier of the supplied schema, so that a registry version registered after
+ // getSchema() was called cannot change the schema used for writing
+ final ProtobufWriteContext context = createWriteContext(variables, schema);
+
+ if (schemaReferenceWriter != null) {
+ schemaReferenceWriter.validateSchema(context.recordSchema());
+ }
+ return new WriteProtobufResultWithExternalSchema(context.schema(), context.messageName(), context.recordSchema(),
+ context.schemaDefinition(), schemaReferenceWriter, messageIndexWriter, variables, out);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = new ArrayList<>(super.getSupportedPropertyDescriptors());
+ properties.removeIf(property -> SCHEMA_REFERENCE_READER.getName().equals(property.getName()));
+ properties.add(MESSAGE_NAME_RESOLUTION_STRATEGY);
+ properties.add(MESSAGE_NAME_RESOLVER);
+ properties.add(MESSAGE_NAME);
+ properties.add(SCHEMA_REFERENCE_WRITER);
+ properties.add(MESSAGE_INDEX_WRITER);
+ return properties;
+ }
+
+ @Override
+ protected List<AllowableValue> getSchemaAccessStrategyValues() {
+ return List.of(SCHEMA_NAME_PROPERTY, SCHEMA_TEXT_PROPERTY);
+ }
+
+ @Override
+ protected Collection<ValidationResult> customValidate(final ValidationContext validationContext) {
+ final List<ValidationResult> results = new ArrayList<>(super.customValidate(validationContext));
+
+ validateLiteralSchemaText(validationContext, results);
+ validateSchemaReferenceWriterCompatibility(validationContext, results);
+
+ return results;
+ }
+
+ @Override
+ protected PropertyDescriptor buildSchemaTextProperty() {
+ return PROTOBUF_SCHEMA_TEXT;
+ }
+
+ private ProtobufWriteContext createWriteContext(final Map<String, String> variables, final RecordSchema suppliedSchema) throws SchemaNotFoundException, IOException {
+ final SchemaDefinition schemaDefinition = createSchemaDefinition(variables, suppliedSchema);
+ final Schema schema = schemaCompiler.compileOrGetFromCache(schemaDefinition);
+ final MessageName messageName = messageNameResolver.getMessageName(variables, schemaDefinition, InputStream.nullInputStream());
+
+ final ProtoSchemaParser schemaParser = new ProtoSchemaParser(schema);
+ final RecordSchema parsedSchema = schemaParser.createSchema(messageName.getFullyQualifiedName());
+ // Preserve the schema identifier required by the configured Schema Reference Writer.
+ final RecordSchema recordSchema = new SimpleRecordSchema(parsedSchema.getFields(), schemaDefinition.getIdentifier());
+
+ return new ProtobufWriteContext(schemaDefinition, schema, messageName, recordSchema);
+ }
+
+ private SchemaDefinition createSchemaDefinition(final Map<String, String> variables, final RecordSchema suppliedSchema) throws SchemaNotFoundException, IOException {
+ if (SCHEMA_TEXT_PROPERTY.getValue().equals(schemaAccessStrategyValue)) {
+ return createSchemaDefinitionFromText(variables);
+ } else if (SCHEMA_NAME_PROPERTY.getValue().equals(schemaAccessStrategyValue)) {
+ final SchemaIdentifier suppliedIdentifier = getVersionedIdentifier(suppliedSchema);
+ if (suppliedIdentifier != null) {
+ return schemaRegistry.retrieveSchemaDefinition(suppliedIdentifier);
+ }
+ return createSchemaDefinitionFromRegistry(variables);
+ }
+
+ throw new SchemaNotFoundException("Unsupported schema access strategy: " + schemaAccessStrategyValue);
+ }
+
+ private void validateLiteralSchemaText(final ValidationContext validationContext, final List<ValidationResult> results) {
+ final String schemaAccessStrategy = validationContext.getProperty(SCHEMA_ACCESS_STRATEGY).getValue();
+ if (!SCHEMA_TEXT_PROPERTY.getValue().equals(schemaAccessStrategy)) {
+ return;
+ }
+
+ final String schemaTextValue = validationContext.getProperty(SCHEMA_TEXT).getValue();
+ if (schemaTextValue == null || schemaTextValue.isBlank() || validationContext.isExpressionLanguagePresent(schemaTextValue)) {
+ return;
+ }
+
+ final Schema compiledSchema;
+ try {
+ final SchemaIdentifier schemaIdentifier = SchemaIdentifier.builder()
+ .name(sha256Hex(schemaTextValue) + PROTO_EXTENSION)
+ .build();
+ final SchemaDefinition schemaDefinition = new StandardSchemaDefinition(schemaIdentifier, schemaTextValue, SchemaDefinition.SchemaType.PROTOBUF);
+ compiledSchema = schemaCompiler.compileOrGetFromCache(schemaDefinition);
+ } catch (final RuntimeException e) {
+ // The compiler reports Wire schema errors as SchemaCompilationException and other failures as other
+ // runtime exceptions, all of which indicate that the configured schema text cannot be used
+ results.add(new ValidationResult.Builder()
+ .subject(SCHEMA_TEXT.getDisplayName())
+ .valid(false)
+ .explanation("Invalid Protocol Buffers schema: " + e.getMessage())
+ .build());
+ return;
+ }
+
+ validateLiteralMessageName(validationContext, compiledSchema, results);
+ }
+
+ private void validateLiteralMessageName(final ValidationContext validationContext, final Schema compiledSchema, final List<ValidationResult> results) {
+ final String resolutionStrategy = validationContext.getProperty(MESSAGE_NAME_RESOLUTION_STRATEGY).getValue();
+ if (!MESSAGE_NAME_PROPERTY.getValue().equals(resolutionStrategy)) {
+ return;
+ }
+
+ final String messageNameValue = validationContext.getProperty(MESSAGE_NAME).getValue();
+ if (messageNameValue == null || messageNameValue.isBlank() || validationContext.isExpressionLanguagePresent(messageNameValue)) {
+ return;
+ }
+
+ if (!(compiledSchema.getType(messageNameValue) instanceof MessageType)) {
+ results.add(new ValidationResult.Builder()
+ .subject(MESSAGE_NAME.getDisplayName())
+ .input(messageNameValue)
+ .valid(false)
+ .explanation("Message name '%s' does not identify a message in the configured Protocol Buffers schema".formatted(messageNameValue))
+ .build());
+ }
+ }
+
+ private void validateSchemaReferenceWriterCompatibility(final ValidationContext validationContext, final List<ValidationResult> results) {
+ if (!validationContext.getProperty(SCHEMA_REFERENCE_WRITER).isSet()) {
+ return;
+ }
+
+ final SchemaReferenceWriter referenceWriter = validationContext.getProperty(SCHEMA_REFERENCE_WRITER).asControllerService(SchemaReferenceWriter.class);
+ if (referenceWriter == null) {
+ return;
+ }
+
+ final Set<SchemaField> missingFields = EnumSet.noneOf(SchemaField.class);
+ missingFields.addAll(referenceWriter.getRequiredSchemaFields());
+ missingFields.removeAll(getSuppliedSchemaFields(validationContext));
+
+ if (!missingFields.isEmpty()) {
+ results.add(new ValidationResult.Builder()
+ .subject(SCHEMA_REFERENCE_WRITER.getDisplayName())
+ .valid(false)
+ .explanation("The configured Schema Reference Writer requires schema fields that are not provided "
+ + "by the configured Schema Access Strategy and Schema Registry: " + missingFields)
+ .build());
+ }
+ }
+
+ 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<String, String> 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<String, String> 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);
+ }
+
+ /**
+ * Returns an identifier addressing the exact registry version of the supplied schema, or null when the supplied
+ * schema does not carry both a name and a version.
+ */
+ private SchemaIdentifier getVersionedIdentifier(final RecordSchema suppliedSchema) {
+ if (suppliedSchema == null) {
+ return null;
+ }
+
+ final SchemaIdentifier identifier = suppliedSchema.getIdentifier();
+ if (identifier.getName().isEmpty() || identifier.getVersion().isEmpty()) {
+ return null;
+ }
+
+ final SchemaIdentifier.Builder identifierBuilder = SchemaIdentifier.builder()
+ .name(identifier.getName().get())
+ .version(identifier.getVersion().getAsInt());
+ identifier.getBranch().ifPresent(identifierBuilder::branch);
+ return identifierBuilder.build();
+ }
+
+ 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<String, String> 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 00000000000..1f35aff3e25
--- /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,121 @@
+/*
+ * 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 a single Record as Protocol Buffers binary content. When configured, the
+ * Schema Reference Writer and Message Index Writer write format-specific information
+ * before the Protobuf payload, in that order.
+ * <p>
+ * Only one Record can be written because raw Protocol Buffers messages do not contain
+ * boundaries that allow concatenated messages to be decoded independently.
+ */
+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<String, String> variables;
+ private final ProtobufDataSerializer serializer;
+ private final OutputStream buffered;
+
+ public WriteProtobufResultWithExternalSchema(final Schema schema,
+ final MessageName messageName,
+ final RecordSchema recordSchema,
+ final SchemaDefinition schemaDefinition,
+ final SchemaReferenceWriter schemaReferenceWriter,
+ final MessageIndexWriter messageIndexWriter,
+ final Map<String, String> 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 Map<String, String> onFinishRecordSet() throws IOException {
+ flush();
+ return getSchemaReferenceAttributes();
+ }
+
+ @Override
+ public Map<String, String> 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), so only a single record per
+ // FlowFile can be represented.
+ if (getRecordCount() > 0) {
+ throw new IOException("Protobuf output supports only a single record because concatenated Protobuf messages cannot be delimited");
+ }
+
+ final byte[] payload = serializer.serialize(record);
+ writeFraming(buffered);
+ buffered.write(payload);
+ return getSchemaReferenceAttributes();
+ }
+
+ private Map<String, String> getSchemaReferenceAttributes() {
+ return schemaReferenceWriter == null ? Map.of() : schemaReferenceWriter.getAttributes(recordSchema);
+ }
+
+ private void writeFraming(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 {
+ // Flushes buffered content and closes the underlying stream even if flushing fails; repeated calls have no effect
+ buffered.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 00000000000..0cf5ce17b56
--- /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,343 @@
+/*
+ * 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.Syntax;
+import com.squareup.wire.schema.EnumConstant;
+import com.squareup.wire.schema.EnumType;
+import com.squareup.wire.schema.Field;
+import com.squareup.wire.schema.MessageType;
+import com.squareup.wire.schema.OneOf;
+import com.squareup.wire.schema.Options;
+import com.squareup.wire.schema.ProtoFile;
+import com.squareup.wire.schema.ProtoMember;
+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.nio.ByteBuffer;
+import java.util.Arrays;
+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.
+ * <p>
+ * 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 static final ProtoMember PACKED_OPTION = ProtoMember.get(Options.FIELD_OPTIONS, "packed");
+
+ 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()) {
+ // A decoder keeps only the last field of a oneof, so writing more than one would silently drop values
+ Field fieldWithValue = null;
+ for (final Field field : oneOf.getFields()) {
+ final Object value = record.getValue(field.getName());
+ if (value == null) {
+ continue;
+ }
+ if (fieldWithValue != null) {
+ throw new IOException(String.format("Oneof [%s] has values for more than one field: [%s] and [%s]",
+ oneOf.getName(), fieldWithValue.getName(), field.getName()));
+ }
+ fieldWithValue = field;
+ writeField(codedOutput, field, value);
+ }
+ }
+
+ codedOutput.flush();
+ return output.toByteArray();
+ }
+
+ private void writeField(final CodedOutputStream output, final Field field, final Object value) throws IOException {
+ if (value == null) {
+ // proto2 parsers reject messages without required fields, so fail here instead of producing unreadable content
+ if (field.isRequired()) {
+ throw new IOException(String.format("Required field [%s] has no value", field.getName()));
+ }
+ 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, isPacked(field), value);
+ } else {
+ writeSingleValue(output, tag, protoType, value);
+ }
+ }
+
+ private void writeRepeated(final CodedOutputStream output, final int tag, final ProtoType protoType, final boolean packed, final Object value) throws IOException {
+ final Object[] values = toArray(value);
+
+ if (packed) {
+ // packed repeated scalar and enum fields are written as 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[] packedValues = packedBytes.toByteArray();
+ output.writeUInt32NoTag(packedValues.length);
+ output.writeRawBytes(packedValues);
+ } else {
+ // unpacked repeated fields, including messages, strings and bytes, are written as one entry per element
+ 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()) {
+ if (entry.getKey() == null) {
+ throw new IOException(String.format("Map field with tag [%d] contains a null key", tag));
+ }
+
+ 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 isPacked(final Field field) {
+ if (!isPackable(field.getType())) {
+ return false;
+ }
+ if (field.isPacked()) {
+ return true;
+ }
+
+ // Wire applies the proto3 packed default only to scalar fields, while proto3 packs repeated enum fields by default too
+ if (field.getOptions().get(PACKED_OPTION) != null || !(schema.getType(field.getType()) instanceof EnumType)) {
+ return false;
+ }
+ final ProtoFile protoFile = schema.protoFile(field.getLocation().getPath());
+ return protoFile != null && protoFile.getSyntax() == Syntax.PROTO_3;
+ }
+
+ 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) throws IOException {
+ 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 IOException(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) throws IOException {
+ if (value instanceof final byte[] bytes) {
+ return bytes;
+ }
+ if (value instanceof final ByteBuffer byteBuffer) {
+ final byte[] bytes = new byte[byteBuffer.remaining()];
+ byteBuffer.duplicate().get(bytes);
+ return bytes;
+ }
+ if (value instanceof final Object[] array) {
+ return toByteArray(Arrays.asList(array));
+ }
+ if (value instanceof final List<?> list) {
+ final byte[] bytes = new byte[list.size()];
+ for (int i = 0; i < bytes.length; i++) {
+ if (!(list.get(i) instanceof final Number number)) {
+ throw new IOException(String.format("Unsupported element [%s] in value for bytes field", list.get(i)));
+ }
+ bytes[i] = number.byteValue();
+ }
+ return bytes;
+ }
+ throw new IOException(String.format("Unsupported value type [%s] for bytes field", value.getClass().getName()));
+ }
+}
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 8ff7783b668..4755655bd2a 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
index 64128d24439..e73b3aca656 100644
--- a/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/java/org/apache/nifi/services/protobuf/TestProtobufSchemaCompiler.java
+++ b/nifi-extension-bundles/nifi-protobuf-bundle/nifi-protobuf-services/src/test/java/org/apache/nifi/services/protobuf/TestProtobufSchemaCompiler.java
@@ -67,6 +67,28 @@ class TestProtobufSchemaCompiler {
private static final String FIELD_EXTRA = "extra";
private static final String ENUM_RED = "RED";
+ // Confluent Schema Registry keys each reference by the import path used inside the .proto file, while the
+ // identifier of the referenced schema carries the registry subject, which is not required to look like a file path
+ private static final String ROOT_SUBJECT = "airlines.ph.cdm.reservation.AirlineReservation";
+ private static final String IMPORT_PATH = "airlines/ph/cdm/shared.proto";
+ private static final String REFERENCE_SUBJECT = "airlines.ph.cdm.shared";
+
+ private static final String ROOT_SCHEMA = """
+ syntax = "proto3";
+ package airlines.ph.cdm.reservation;
+ import "airlines/ph/cdm/shared.proto";
+ message AirlineReservation {
+ string reservation_id = 1;
+ airlines.ph.cdm.Status status = 2;
+ }""";
+
+ private static final String REFERENCED_SCHEMA = """
+ syntax = "proto3";
+ package airlines.ph.cdm;
+ message Status {
+ string code = 1;
+ }""";
+
private final ProtobufSchemaCompiler compiler = new ProtobufSchemaCompiler("test", COMPONENT_LOG);
@Test
@@ -372,6 +394,63 @@ class TestProtobufSchemaCompiler {
assertNoParenthesizedOptions(lookup.getOptions());
}
+ @Test
+ void testCompileSchemaWithCrossFileImportFromRegistry() {
+ final SchemaDefinition referencedSchema = new StandardSchemaDefinition(
+ SchemaIdentifier.builder().name(REFERENCE_SUBJECT).id(4L).version(1).build(),
+ REFERENCED_SCHEMA,
+ SchemaDefinition.SchemaType.PROTOBUF);
+
+ final SchemaDefinition rootSchema = new StandardSchemaDefinition(
+ SchemaIdentifier.builder().name(ROOT_SUBJECT).id(3L).version(2).build(),
+ ROOT_SCHEMA,
+ SchemaDefinition.SchemaType.PROTOBUF,
+ Map.of(IMPORT_PATH, referencedSchema));
+
+ final Schema compiled = compiler.compileOrGetFromCache(rootSchema);
+
+ assertNotNull(compiled.getType("airlines.ph.cdm.reservation.AirlineReservation"));
+ assertNotNull(compiled.getType("airlines.ph.cdm.Status"));
+ }
+
+ @Test
+ void testCompileSchemaWithNestedCrossFileImports() {
+ final SchemaDefinition leafSchema = new StandardSchemaDefinition(
+ SchemaIdentifier.builder().name("airlines.ph.cdm.common").id(5L).version(1).build(),
+ """
+ syntax = "proto3";
+ package airlines.ph.cdm.common;
+ message Audit {
+ string created_by = 1;
+ }""",
+ SchemaDefinition.SchemaType.PROTOBUF);
+
+ final SchemaDefinition referencedSchema = new StandardSchemaDefinition(
+ SchemaIdentifier.builder().name(REFERENCE_SUBJECT).id(4L).version(1).build(),
+ """
+ syntax = "proto3";
+ package airlines.ph.cdm;
+ import "airlines/ph/cdm/common/audit.proto";
+ message Status {
+ string code = 1;
+ airlines.ph.cdm.common.Audit audit = 2;
+ }""",
+ SchemaDefinition.SchemaType.PROTOBUF,
+ Map.of("airlines/ph/cdm/common/audit.proto", leafSchema));
+
+ final SchemaDefinition rootSchema = new StandardSchemaDefinition(
+ SchemaIdentifier.builder().name(ROOT_SUBJECT).id(3L).version(2).build(),
+ ROOT_SCHEMA,
+ SchemaDefinition.SchemaType.PROTOBUF,
+ Map.of(IMPORT_PATH, referencedSchema));
+
+ final Schema compiled = compiler.compileOrGetFromCache(rootSchema);
+
+ assertNotNull(compiled.getType("airlines.ph.cdm.reservation.AirlineReservation"));
+ assertNotNull(compiled.getType("airlines.ph.cdm.Status"));
+ assertNotNull(compiled.getType("airlines.ph.cdm.common.Audit"));
+ }
+
private void assertNoParenthesizedOptions(final Options options) {
final List<OptionElement> elements = options.getElements();
final List<Executable> assertions = new ArrayList<>(elements.size());
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 edd01f7b5ef..88ff5545740 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
@@ -48,7 +48,7 @@ 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.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;
@@ -141,7 +141,7 @@ class TestStandardProtobufReader extends StandardProtobufReaderTestBase {
enableAllControllerServices();
- validateSchemaDefinitionIdentifiers(validMainSchema, true);
+ validateSchemaReferencePaths(validMainSchema);
}
@Test
@@ -151,29 +151,42 @@ class TestStandardProtobufReader extends StandardProtobufReaderTestBase {
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 00000000000..191ba6ae7bb
--- /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,586 @@
+/*
+ * 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.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.WriteResult;
+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.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.EnumSet;
+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_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.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.assertNotEquals;
+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 static final Map<String, String> FAKE_ATTRIBUTES = Map.of("schema.name", "proto3");
+
+ private static final String VERSIONED_SUBJECT = "versioned";
+ private static final String VERSIONED_MESSAGE = "Versioned";
+ // Length-delimited field keys: (field number << 3) | wire type 2
+ private static final int FIELD_1_STRING_KEY = 0x0A;
+ private static final int FIELD_2_STRING_KEY = 0x12;
+
+ 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<PropertyDescriptor> descriptors = writer.getPropertyDescriptors();
+ final PropertyDescriptor strategy = descriptors.stream()
+ .filter(descriptor -> SCHEMA_ACCESS_STRATEGY.getName().equals(descriptor.getName()))
+ .findFirst()
+ .orElseThrow();
+
+ final List<String> 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 testEmptyRecordSetWritesNoFraming() throws Exception {
+ enableFakeFramingWriters();
+ 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.finishRecordSet();
+ }
+
+ assertEquals(0, out.size());
+ }
+
+ @Test
+ void testSerializationFailureWritesNoFraming() throws Exception {
+ enableFakeFramingWriters();
+ runner.enableControllerService(writer);
+
+ // A String value for a message field cannot be serialized
+ final MapRecord invalidRecord = new MapRecord(new SimpleRecordSchema(List.of(new RecordField("nestedMessage", RecordFieldType.STRING.getDataType()))),
+ Map.of("nestedMessage", "not a record"));
+
+ final RecordSchema writeSchema = writer.getSchema(emptyMap(), null);
+ final ByteArrayOutputStream out = new ByteArrayOutputStream();
+ try (RecordSetWriter recordSetWriter = writer.createWriter(runner.getLogger(), writeSchema, out, emptyMap())) {
+ recordSetWriter.beginRecordSet();
+ assertThrows(IOException.class, () -> recordSetWriter.write(invalidRecord));
+ }
+
+ assertEquals(0, out.size());
+ }
+
+ @Test
+ void testCloseClosesStreamWhenFlushFails() throws Exception {
+ runner.enableControllerService(writer);
+
+ final RecordSchema writeSchema = writer.getSchema(emptyMap(), null);
+ final FlushFailingOutputStream out = new FlushFailingOutputStream();
+ final RecordSetWriter recordSetWriter = writer.createWriter(runner.getLogger(), writeSchema, out, emptyMap());
+ recordSetWriter.write(buildProto3Record());
+
+ assertThrows(IOException.class, recordSetWriter::close);
+ assertTrue(out.closed);
+ }
+
+ @Test
+ void testCloseTwiceHasNoEffect() throws Exception {
+ runner.enableControllerService(writer);
+
+ final RecordSchema writeSchema = writer.getSchema(emptyMap(), null);
+ final ByteArrayOutputStream out = new ByteArrayOutputStream();
+ final RecordSetWriter recordSetWriter = writer.createWriter(runner.getLogger(), writeSchema, out, emptyMap());
+ recordSetWriter.write(buildProto3Record());
+ recordSetWriter.close();
+ final int size = out.size();
+
+ recordSetWriter.close();
+
+ assertEquals(size, out.size());
+ assertNotEquals(0, size);
+ }
+
+ @Test
+ void testFinishRecordSetReturnsSchemaReferenceAttributes() throws Exception {
+ enableFakeSchemaReferenceWriter();
+ 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());
+ final WriteResult writeResult = recordSetWriter.finishRecordSet();
+
+ assertEquals(FAKE_ATTRIBUTES, writeResult.getAttributes());
+ }
+ }
+
+ @Test
+ void testWriteWithoutActiveRecordSetReturnsSchemaReferenceAttributes() throws Exception {
+ enableFakeSchemaReferenceWriter();
+ 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())) {
+ final WriteResult writeResult = recordSetWriter.write(buildProto3Record());
+
+ assertEquals(FAKE_ATTRIBUTES, writeResult.getAttributes());
+ }
+ }
+
+ @Test
+ void testCreateWriterUsesVersionOfSuppliedSchema() throws Exception {
+ final VersionedSchemaRegistry schemaRegistry = new VersionedSchemaRegistry();
+ runner.addControllerService("versionedRegistry", schemaRegistry);
+ runner.enableControllerService(schemaRegistry);
+
+ final StandardProtobufWriter registryWriter = new StandardProtobufWriter();
+ runner.addControllerService("registryWriter", registryWriter);
+ runner.setProperty(registryWriter, SCHEMA_ACCESS_STRATEGY, SCHEMA_NAME_PROPERTY.getValue());
+ runner.setProperty(registryWriter, SCHEMA_REGISTRY, "versionedRegistry");
+ runner.setProperty(registryWriter, SCHEMA_NAME, VERSIONED_SUBJECT);
+ runner.setProperty(registryWriter, StandardProtobufWriter.MESSAGE_NAME, VERSIONED_MESSAGE);
+ runner.enableControllerService(registryWriter);
+
+ final RecordSchema writeSchema = registryWriter.getSchema(emptyMap(), null);
+ assertEquals(1, writeSchema.getIdentifier().getVersion().getAsInt());
+
+ // A new latest version registered between getSchema() and createWriter() must not change the written content
+ schemaRegistry.latestVersion = 2;
+
+ final MapRecord record = new MapRecord(new SimpleRecordSchema(List.of(new RecordField("name", RecordFieldType.STRING.getDataType()))),
+ Map.of("name", "Alice"));
+ assertEquals(FIELD_1_STRING_KEY, writeFirstByte(registryWriter, writeSchema, record));
+
+ // A schema obtained after the new version was registered uses the new version
+ final RecordSchema latestWriteSchema = registryWriter.getSchema(emptyMap(), null);
+ assertEquals(FIELD_2_STRING_KEY, writeFirstByte(registryWriter, latestWriteSchema, record));
+ }
+
+ private byte writeFirstByte(final StandardProtobufWriter protobufWriter, final RecordSchema writeSchema, final MapRecord record) throws Exception {
+ final ByteArrayOutputStream out = new ByteArrayOutputStream();
+ try (RecordSetWriter recordSetWriter = protobufWriter.createWriter(runner.getLogger(), writeSchema, out, emptyMap())) {
+ recordSetWriter.write(record);
+ }
+ return out.toByteArray()[0];
+ }
+
+ @Test
+ void testInvalidLiteralSchemaText() {
+ runner.setProperty(writer, SCHEMA_TEXT, "message Broken {");
+
+ runner.assertNotValid(writer);
+ }
+
+ @Test
+ void testLiteralMessageNameNotInSchema() {
+ runner.setProperty(writer, StandardProtobufWriter.MESSAGE_NAME, "MissingMessage");
+
+ runner.assertNotValid(writer);
+ }
+
+ @Test
+ void testExpressionLanguageSchemaTextNotCompiledDuringValidation() {
+ runner.setProperty(writer, SCHEMA_TEXT, "${proto.schema}");
+ runner.setProperty(writer, StandardProtobufWriter.MESSAGE_NAME, "MissingMessage");
+
+ runner.assertValid(writer);
+ }
+
+ @Test
+ void testSchemaReferenceWriterRequiringFieldsNotSuppliedIsInvalid() throws Exception {
+ final IdentifierRequiringSchemaReferenceWriter referenceWriter = new IdentifierRequiringSchemaReferenceWriter();
+ runner.addControllerService("identifierReferenceWriter", referenceWriter);
+ runner.enableControllerService(referenceWriter);
+ runner.setProperty(writer, StandardProtobufWriter.SCHEMA_REFERENCE_WRITER, "identifierReferenceWriter");
+
+ runner.assertNotValid(writer);
+ }
+
+ @Test
+ void testCreateWriterValidatesSchemaWithSchemaReferenceWriter() throws Exception {
+ final RejectingSchemaReferenceWriter referenceWriter = new RejectingSchemaReferenceWriter();
+ runner.addControllerService("rejectingReferenceWriter", referenceWriter);
+ runner.enableControllerService(referenceWriter);
+ runner.setProperty(writer, StandardProtobufWriter.SCHEMA_REFERENCE_WRITER, "rejectingReferenceWriter");
+ runner.enableControllerService(writer);
+
+ final RecordSchema writeSchema = writer.getSchema(emptyMap(), null);
+ assertThrows(SchemaNotFoundException.class,
+ () -> writer.createWriter(runner.getLogger(), writeSchema, new ByteArrayOutputStream(), emptyMap()));
+ }
+
+ @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 void enableFakeSchemaReferenceWriter() throws Exception {
+ final FakeSchemaReferenceWriter referenceWriter = new FakeSchemaReferenceWriter();
+ runner.addControllerService("referenceWriter", referenceWriter);
+ runner.enableControllerService(referenceWriter);
+ runner.setProperty(writer, StandardProtobufWriter.SCHEMA_REFERENCE_WRITER, "referenceWriter");
+ }
+
+ private void enableFakeFramingWriters() throws Exception {
+ enableFakeSchemaReferenceWriter();
+ final FakeMessageIndexWriter indexWriter = new FakeMessageIndexWriter();
+ runner.addControllerService("indexWriter", indexWriter);
+ runner.enableControllerService(indexWriter);
+ runner.setProperty(writer, StandardProtobufWriter.MESSAGE_INDEX_WRITER, "indexWriter");
+ }
+
+ 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<String, String> getAttributes(final RecordSchema recordSchema) {
+ return FAKE_ATTRIBUTES;
+ }
+
+ @Override
+ public void validateSchema(final RecordSchema recordSchema) throws SchemaNotFoundException {
+ }
+
+ @Override
+ public Set<SchemaField> getRequiredSchemaFields() {
+ return emptySet();
+ }
+ }
+
+ static class FakeMessageIndexWriter extends AbstractControllerService implements MessageIndexWriter {
+ @Override
+ public void writeMessageIndex(final Map<String, String> variables, final SchemaDefinition schemaDefinition, final MessageName messageName, final OutputStream outputStream) throws IOException {
+ outputStream.write(FAKE_INDEX);
+ }
+ }
+
+ /**
+ * Returns version 1 or 2 of the same subject, where the {@code name} field moves from tag 1 to tag 2, and resolves
+ * lookups without a version to the configurable latest version.
+ */
+ static class VersionedSchemaRegistry extends AbstractControllerService implements SchemaRegistry {
+ private volatile int latestVersion = 1;
+
+ @Override
+ public RecordSchema retrieveSchema(final SchemaIdentifier schemaIdentifier) {
+ throw new UnsupportedOperationException("retrieveSchema is not used in this test");
+ }
+
+ @Override
+ public SchemaDefinition retrieveSchemaDefinition(final SchemaIdentifier schemaIdentifier) {
+ final int version = schemaIdentifier.getVersion().orElse(latestVersion);
+ final String schemaText = """
+ syntax = "proto3";
+ message Versioned {
+ string name = %d;
+ }""".formatted(version);
+ return new StandardSchemaDefinition(
+ SchemaIdentifier.builder().name(VERSIONED_SUBJECT).id((long) version).version(version).build(),
+ schemaText,
+ SchemaDefinition.SchemaType.PROTOBUF);
+ }
+
+ @Override
+ public Set<SchemaField> getSuppliedSchemaFields() {
+ return emptySet();
+ }
+ }
+
+ static class IdentifierRequiringSchemaReferenceWriter extends FakeSchemaReferenceWriter {
+ @Override
+ public Set<SchemaField> getRequiredSchemaFields() {
+ return EnumSet.of(SchemaField.SCHEMA_IDENTIFIER, SchemaField.SCHEMA_VERSION);
+ }
+ }
+
+ static class RejectingSchemaReferenceWriter extends FakeSchemaReferenceWriter {
+ @Override
+ public void validateSchema(final RecordSchema recordSchema) throws SchemaNotFoundException {
+ throw new SchemaNotFoundException("Schema rejected");
+ }
+ }
+
+ static class FlushFailingOutputStream extends ByteArrayOutputStream {
+ private boolean closed;
+
+ @Override
+ public void flush() throws IOException {
+ throw new IOException("Flush failed");
+ }
+
+ @Override
+ public void close() {
+ closed = true;
+ }
+ }
+}
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 00000000000..dd2d5d6b13e
--- /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,252 @@
+/*
+ * 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.EnumSet;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static java.util.Collections.emptyMap;
+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.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_TEXT;
+import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_TEXT_PROPERTY;
+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_VALUE = """
+ 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<String, Object> 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<String, Object> 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"));
+ }
+
+ @Test
+ void testSchemaTextWithConfluentSchemaReferenceWriterIsInvalid() throws Exception {
+ final StandardProtobufWriter writer = new StandardProtobufWriter();
+ runner.addControllerService("schemaTextWriter", writer);
+ runner.setProperty(writer, SCHEMA_ACCESS_STRATEGY, SCHEMA_TEXT_PROPERTY.getValue());
+ runner.setProperty(writer, SCHEMA_TEXT, SCHEMA_TEXT_VALUE);
+ runner.setProperty(writer, StandardProtobufWriter.MESSAGE_NAME, "User");
+ runner.assertValid(writer);
+
+ // Inline schema text has no registry identifier or version for the Confluent header
+ runner.setProperty(writer, StandardProtobufWriter.SCHEMA_REFERENCE_WRITER, "referenceWriter");
+ runner.assertNotValid(writer);
+ }
+
+ 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_VALUE,
+ 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<SchemaField> getSuppliedSchemaFields() {
+ return EnumSet.of(SchemaField.SCHEMA_NAME, SchemaField.SCHEMA_TEXT, SchemaField.SCHEMA_TEXT_FORMAT,
+ SchemaField.SCHEMA_IDENTIFIER, SchemaField.SCHEMA_VERSION);
+ }
+ }
+}
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 00000000000..e8d2e123b61
--- /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,334 @@
+/*
+ * 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.Location;
+import com.squareup.wire.schema.Schema;
+import com.squareup.wire.schema.SchemaLoader;
+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 org.junit.jupiter.api.io.TempDir;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.math.BigInteger;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.nifi.services.protobuf.ProtoTestUtil.generateInputDataForRootMessage;
+import static org.apache.nifi.services.protobuf.ProtoTestUtil.loadProto2TestSchema;
+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;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * 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"));
+ }
+
+ @Test
+ public void testSerializeProto2RequiredFieldPresent() throws IOException {
+ final Schema schema = loadProto2TestSchema();
+ final RecordSchema recordSchema = new ProtoSchemaParser(schema).createSchema("Proto2Message");
+ final MapRecord originalRecord = new MapRecord(recordSchema, Map.of("booleanField", true));
+
+ final byte[] serialized = new ProtobufDataSerializer(schema, "Proto2Message").serialize(originalRecord);
+
+ final MapRecord record = new ProtobufDataConverter(schema, "Proto2Message", recordSchema, false, false)
+ .createRecord(new ByteArrayInputStream(serialized));
+ assertEquals(true, record.getValue("booleanField"));
+ }
+
+ @Test
+ public void testSerializeProto2MissingRequiredFieldFails() {
+ final Schema schema = loadProto2TestSchema();
+ final RecordSchema recordSchema = new ProtoSchemaParser(schema).createSchema("Proto2Message");
+ final MapRecord record = new MapRecord(recordSchema, Map.of("stringField", "Test text"));
+
+ final IOException exception = assertThrows(IOException.class,
+ () -> new ProtobufDataSerializer(schema, "Proto2Message").serialize(record));
+ assertTrue(exception.getMessage().contains("booleanField"));
+ }
+
+ @Test
+ public void testSerializeNullMapKeyFails() throws DescriptorValidationException, IOException {
+ final Schema schema = loadProto3TestSchema();
+ final RecordSchema recordSchema = new ProtoSchemaParser(schema).createSchema("Proto3Message");
+ final MapRecord record = new ProtobufDataConverter(schema, "Proto3Message", recordSchema, false, false)
+ .createRecord(ProtoTestUtil.generateInputDataForProto3());
+
+ final Map<String, Object> mapWithNullKey = new HashMap<>();
+ mapWithNullKey.put(null, 1);
+ final MapRecord nestedRecord = (MapRecord) record.getValue("nestedMessage");
+ final MapRecord nestedRecord2 = (MapRecord) ((Object[]) nestedRecord.getValue("nestedMessage2"))[0];
+ nestedRecord2.setValue("testMap", mapWithNullKey);
+
+ final IOException exception = assertThrows(IOException.class,
+ () -> new ProtobufDataSerializer(schema, "Proto3Message").serialize(record));
+ assertTrue(exception.getMessage().contains("null key"));
+ }
+
+ // Tag bytes: (field number << 3) | wire type, where wire type 0 is varint and 2 is length-delimited
+ private static final byte FIELD_1_VARINT_TAG = 0x08;
+ private static final byte FIELD_1_LENGTH_DELIMITED_TAG = 0x0A;
+
+ @TempDir
+ private Path schemaDirectory;
+
+ @Test
+ public void testSerializeOneofWithMultipleValuesFails() throws IOException {
+ final Schema schema = compileSchema("""
+ syntax = "proto3";
+ message Choice {
+ oneof value {
+ string text = 1;
+ int32 number = 2;
+ }
+ }""");
+ final MapRecord record = new MapRecord(new ProtoSchemaParser(schema).createSchema("Choice"), Map.of("text", "a", "number", 1));
+
+ final IOException exception = assertThrows(IOException.class, () -> new ProtobufDataSerializer(schema, "Choice").serialize(record));
+ assertTrue(exception.getMessage().contains("value"));
+ }
+
+ @Test
+ public void testSerializeOneofWithSingleValue() throws IOException {
+ final Schema schema = compileSchema("""
+ syntax = "proto3";
+ message Choice {
+ oneof value {
+ string text = 1;
+ int32 number = 2;
+ }
+ }""");
+ final RecordSchema recordSchema = new ProtoSchemaParser(schema).createSchema("Choice");
+ final MapRecord originalRecord = new MapRecord(recordSchema, Map.of("number", 7));
+
+ final byte[] serialized = new ProtobufDataSerializer(schema, "Choice").serialize(originalRecord);
+
+ final MapRecord record = new ProtobufDataConverter(schema, "Choice", recordSchema, false, false).createRecord(new ByteArrayInputStream(serialized));
+ assertEquals(7, record.getValue("number"));
+ assertNull(record.getValue("text"));
+ }
+
+ @Test
+ public void testSerializeProto2RepeatedUnpackedByDefault() throws IOException {
+ assertArrayEquals(new byte[] {FIELD_1_VARINT_TAG, 1, FIELD_1_VARINT_TAG, 2}, serializeRepeated("proto2", "int32", ""));
+ }
+
+ @Test
+ public void testSerializeProto2RepeatedPackedOption() throws IOException {
+ assertArrayEquals(new byte[] {FIELD_1_LENGTH_DELIMITED_TAG, 2, 1, 2}, serializeRepeated("proto2", "int32", " [packed = true]"));
+ }
+
+ @Test
+ public void testSerializeProto3RepeatedPackedByDefault() throws IOException {
+ assertArrayEquals(new byte[] {FIELD_1_LENGTH_DELIMITED_TAG, 2, 1, 2}, serializeRepeated("proto3", "int32", ""));
+ }
+
+ @Test
+ public void testSerializeProto3RepeatedUnpackedOption() throws IOException {
+ assertArrayEquals(new byte[] {FIELD_1_VARINT_TAG, 1, FIELD_1_VARINT_TAG, 2}, serializeRepeated("proto3", "int32", " [packed = false]"));
+ }
+
+ @Test
+ public void testSerializeProto3RepeatedEnumPackedByDefault() throws IOException {
+ assertArrayEquals(new byte[] {FIELD_1_LENGTH_DELIMITED_TAG, 2, 1, 2}, serializeRepeated("proto3", "Color", ""));
+ }
+
+ @Test
+ public void testSerializeProto2RepeatedEnumUnpackedByDefault() throws IOException {
+ assertArrayEquals(new byte[] {FIELD_1_VARINT_TAG, 1, FIELD_1_VARINT_TAG, 2}, serializeRepeated("proto2", "Color", ""));
+ }
+
+ @Test
+ public void testSerializeBytesFromByteBuffer() throws IOException {
+ final byte[] expected = "bytes".getBytes(StandardCharsets.UTF_8);
+ assertArrayEquals(expected, serializeAndReadBytes(ByteBuffer.wrap(expected)));
+ }
+
+ @Test
+ public void testSerializeBytesFromUnsupportedValueFails() {
+ assertThrows(IOException.class, () -> serializeAndReadBytes("text"));
+ }
+
+ @Test
+ public void testSerializeBytesWithNonNumericElementFails() {
+ assertThrows(IOException.class, () -> serializeAndReadBytes(new Object[] {1, "two"}));
+ }
+
+ private byte[] serializeRepeated(final String syntax, final String elementType, final String fieldOptions) throws IOException {
+ final Schema schema = compileSchema("""
+ syntax = "%s";
+ enum Color {
+ RED = 0;
+ GREEN = 1;
+ BLUE = 2;
+ }
+ message Values {
+ repeated %s values = 1%s;
+ }""".formatted(syntax, elementType, fieldOptions));
+ final Object[] values = "Color".equals(elementType) ? new Object[] {"GREEN", "BLUE"} : new Object[] {1, 2};
+ final MapRecord record = new MapRecord(new ProtoSchemaParser(schema).createSchema("Values"), Map.of("values", values));
+
+ return new ProtobufDataSerializer(schema, "Values").serialize(record);
+ }
+
+ private byte[] serializeAndReadBytes(final Object value) throws IOException {
+ final Schema schema = compileSchema("""
+ syntax = "proto3";
+ message Binary {
+ bytes data = 1;
+ }""");
+ final RecordSchema recordSchema = new ProtoSchemaParser(schema).createSchema("Binary");
+ final MapRecord originalRecord = new MapRecord(recordSchema, Map.of("data", value));
+
+ final byte[] serialized = new ProtobufDataSerializer(schema, "Binary").serialize(originalRecord);
+
+ final MapRecord record = new ProtobufDataConverter(schema, "Binary", recordSchema, false, false).createRecord(new ByteArrayInputStream(serialized));
+ return (byte[]) record.getValue("data");
+ }
+
+ private Schema compileSchema(final String schemaText) throws IOException {
+ Files.writeString(schemaDirectory.resolve("test.proto"), schemaText);
+ final SchemaLoader schemaLoader = new SchemaLoader(FileSystems.getDefault());
+ schemaLoader.initRoots(List.of(Location.get(schemaDirectory.toString())), List.of());
+ return schemaLoader.loadSchema();
+ }
+}
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 00000000000..3d3d2e83da1
--- /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.
+ * <p>
+ * 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.
+ * </p>
+ */
+public interface MessageIndexWriter extends ControllerService {
+
+ /**
+ * Writes the message index identifying the given message name within the provided schema
+ * definition to the provided output stream.
+ * <p>
+ * 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.
+ * </p>
+ *
+ * @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<String, String> variables, final SchemaDefinition schemaDefinition, final MessageName messageName, final OutputStream outputStream) throws IOException;
+}