apache · prankstrisse · Aug 3, 2026
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
@@ -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/...confluent/schemaregistry/VarintUtils.java → ...he/nifi/confluent/schema/VarintUtils.java b/...confluent/schemaregistry/VarintUtils.java → ...he/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 @@
  * <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 @@ private 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 @@ static int readVarintFromStream(final InputStream inputStream) throws IOExceptio
      * @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 @@ static int readVarintFromStreamAfterFirstByteConsumed(final InputStream inputStr
      * @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 @@ static int decodeZigZag(final int encodedValue) {
      * @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/...luent/schemaregistry/VarintUtilsTest.java → ...ifi/confluent/schema/VarintUtilsTest.java b/...luent/schemaregistry/VarintUtilsTest.java → ...ifi/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.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/...ndles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-index-writer/pom.xml b/...ndles/nifi-confluent-platform-bundle/nifi-confluent-protobuf-message-index-writer/pom.xml
@@ -0,0 +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
+    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.12.0-SNAPSHOT</version>
+    </parent>
+
+    <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-schema-registry-service-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-confluent-platform-schema-api</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <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/...in/java/org/apache/nifi/confluent/schemaregistry/ConfluentProtobufMessageIndexWriter.java b/...in/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 parse protobuf schema", e);
+        }
+    }
+
+    private record EncodeMessageIndexArguments(SchemaDefinition schemaDefinition, MessageName messageName) {
+    }
+}
diff --git a/...r/src/main/java/org/apache/nifi/confluent/schemaregistry/ProtobufMessageIndexEncoder.java b/...r/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;
+        }
+
+        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/...-writer/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService b/...-writer/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService
@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+org.apache.nifi.confluent.schemaregistry.ConfluentProtobufMessageIndexWriter