NIFI-14210 Add record writer for Protobuf messages by prankstrisse · Pull Request #11499 · apache/nifi

prankstrisse · 2026-08-03T09:35:14Z

Summary

Changes

This PR adds a new Protobuf writer to NiFi, so it can now write Protobuf messages (before, it could only read them). As part of that work, it also fixes schema resolution in Schema Registry: schemas that reference other schemas now work correctly, whether they're in the same package or a different one.

StandardProtobufWriter (nifi-protobuf-services) — new SchemaRegistryService + RecordSetWriterFactory. Resolves the target Proto message either statically via a Message Name property or dynamically via a MessageNameResolver service, compiles the schema with the existing ProtobufSchemaCompiler, and serializes each Record with a new ProtobufDataSerializer. A single message is written per FlowFile, since concatenated Protobuf messages cannot be delimited on read.
ProtobufDataSerializer — maps NiFi Record fields onto Wire Schema/MessageType descriptors and encodes the binary payload. google.protobuf.Any-typed fields are written as plain nested messages rather than re-wrapped as Any.
WriteProtobufResultWithExternalSchema — orchestrates header + message-index + payload sequencing, mirroring the existing WriteAvroResultWithExternalSchema pattern: invokes SchemaReferenceWriter.writeHeader(...) when a Schema Reference Writer is configured, then an optional MessageIndexWriter.writeMessageIndex(...) step, then the Protobuf payload.
MessageIndexWriter (nifi-schema-registry-service-api) — new shared interface, the write-side inverse of MessageNameResolver: given a resolved message name and schema definition, encodes the path to that message and writes it to the output stream so a reader can resolve the same name back from the encoded path.
ConfluentProtobufMessageIndexWriter + ProtobufMessageIndexEncoder (nifi-confluent-protobuf-message-index-writer, new module) — Confluent implementation of MessageIndexWriter. Walks nested message declarations to find the target message's declaration-order path and zigzag-varint encodes it (including the [0] single-message optimization), mirroring the decoding done by ConfluentProtobufMessageNameResolver. Together with the unchanged, already-generic ConfluentEncodedSchemaReferenceWriter, this reproduces the Confluent wire format (5-byte header + message-index array + payload) on write.
VarintUtils — relocated from the message-name-resolver module to org.apache.nifi.confluent.schema and made non-package-private so it can be shared between the resolver (read) and index writer (write) modules.
Minor visibility/refactoring changes to ProtobufSchemaCompiler and ProtobufSchemaValidator to support reuse from the writer.

Tracking

Please complete the following tracking steps prior to pull request creation.

Issue Tracking

Apache NiFi Jira issue created

Pull Request Tracking

Pull Request title starts with Apache NiFi Jira issue number, such as NIFI-00000
Pull Request commit message starts with Apache NiFi Jira issue number, as such NIFI-00000
Pull request contains commits signed with a registered key indicating Verified status

Pull Request Formatting

Pull Request based on current revision of the main branch
Pull Request refers to a feature branch with one commit containing changes

Verification

Please indicate the verification steps performed prior to pull request creation.

Build

Build completed using ./mvnw clean install -P contrib-check
- JDK 21
- JDK 25

Licensing

New dependencies are compatible with the Apache License 2.0 according to the License Policy
New dependencies are documented in applicable LICENSE and NOTICE files

Documentation

Documentation formatting appears as expected in rendered files

prankstrisse · 2026-09-24T12:01:15Z

@tpalfy thanks for the interest in this PR. There were some merge conflicts after merging #11710 . So the commit I pushed today is the resolution of them

pvillard31 · 2026-09-24T12:39:31Z

+    }
+
+    @Override
+    protected Map<String, String> onFinishRecordSet() throws IOException {


Should this return schemaReferenceWriter.getAttributes(recordSchema) when a Schema Reference Writer is configured, so attribute-based schema references are not discarded?

yes, fixed. It now returns the attributes both in onFinishRecordSet() and when a record is written without an active record set

pvillard31 · 2026-09-24T12:39:31Z

+                writeField(codedOutput, field, record.getValue(field.getName()));
+            }
+        }
+


Should a missing proto2 required field fail serialization, or should the writer reject proto2 schemas if only proto3 output is supported?

I've seen in the tests for the read part - it handles proto2, so I kept proto2 support in the writer for consistency and made a missing required field fail with an IOException.

pvillard31 · 2026-09-24T12:39:31Z

+    }
+
+    @Override
+    public RecordSetWriter createWriter(final ComponentLog logger, final RecordSchema schema, final OutputStream out, final Map<String, String> variables) throws SchemaNotFoundException, IOException {


Should createWriter() use the supplied schema instead of resolving the registry schema again, so a moving latest version cannot change between getSchema() and writer creation?

Yes, totally agree.
I can't use the supplied RecordSchema on its own, because writing also needs the compiled Protobuf schema and the original schema text (for the message index and imports). They cannot be rebuilt from a RecordSchema. So createWriter() now fetches the registry schema using the name and version of the supplied schema. If the supplied schema has no name and version, it falls back to the previous lookup.

pvillard31 · 2026-09-24T12:39:31Z

+            final ByteArrayOutputStream entryBytes = new ByteArrayOutputStream();
+            final CodedOutputStream entryOutput = CodedOutputStream.newInstance(entryBytes);
+
+            writeSingleValue(entryOutput, MAP_KEY_TAG, keyType, entry.getKey());


Should null map keys be rejected with a clear IOException before scalar conversion?

Yes, fixed. I also changed unknown enum values to throw IOException instead of IllegalStateException, so that the errors are reported in the same way in all places

pvillard31 · 2026-09-24T12:39:31Z

+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;


Should @Seealso(StandardProtobufReader.class) be added so users can discover the companion reader?

pvillard31 · 2026-09-24T12:39:31Z

+        final List<Integer> messageIndexPath = findMessageIndexPath(rootMessages, messageName);
+
+        if (messageIndexPath.size() == 1 && messageIndexPath.getFirst() == 0) {
+            return FIRST_ROOT_MESSAGE_INDEX;


Should this return a copy of FIRST_ROOT_MESSAGE_INDEX so callers cannot mutate shared cached data?

Yes, done. Side note: ConfluentProtobufMessageIndexWriter also caches the encoded arrays and returns the same array each time. That's safe because callers only write those bytes to the output stream and never change them, so I left the cache as it is

- Return Schema Reference Writer attributes from the record set writer - Resolve the registry schema using the version of the supplied schema in createWriter - Fail serialization when a proto2 required field has no value - Reject null map keys and unknown enum constants with IOException - Return a copy of the first root message index - Add SeeAlso reference to StandardProtobufReader

prankstrisse · 2026-09-24T14:02:22Z

Thank you very much for the review @pvillard31 , I pushed a commit with the fixes

tpalfy · 2026-09-22T13:56:09Z

+     * The identifier of a referenced schema is deliberately not validated. It carries the subject the schema is
+     * registered under, which is unrelated to the import path and legitimately has no .proto suffix. Under the
+     * Confluent RecordNameStrategy, for instance, a subject is a fully qualified record name.


I would avoid referencing Confluent-specific details.

Suggested change

* The identifier of a referenced schema is deliberately not validated. It carries the subject the schema is

* registered under, which is unrelated to the import path and legitimately has no .proto suffix. Under the

* Confluent RecordNameStrategy, for instance, a subject is a fully qualified record name.

* 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.

thanks, done

tpalfy · 2026-09-22T14:40:59Z

+     * Import paths may contain directories, such as {@code airlines/ph/cdm/shared.proto}, so the enclosing
+     * directory structure has to exist before the file is written.


Suggested change

* Import paths may contain directories, such as {@code airlines/ph/cdm/shared.proto}, so the enclosing

* directory structure has to exist before the file is written.

* 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.

thanks, done

tpalfy · 2026-09-22T15:58:58Z

+        // The Schema Reference Reader is a read-side concern: a writer determines its schema from the configured
+        // access strategy and writes references through the Schema Reference Writer instead.


Unnecessary (and a bit confusing) comment.

Suggested change

// The Schema Reference Reader is a read-side concern: a writer determines its schema from the configured

// access strategy and writes references through the Schema Reference Writer instead.

thanks, done

tpalfy · 2026-09-22T15:59:57Z

+        // Only the strategies that createSchemaDefinition supports are offered; the inherited list also contains
+        // the Schema Reference Reader strategy, which cannot be used to obtain a schema for writing.


Unnecessary comments.

Suggested change

// Only the strategies that createSchemaDefinition supports are offered; the inherited list also contains

// the Schema Reference Reader strategy, which cannot be used to obtain a schema for writing.

thanks, done

tpalfy · 2026-09-22T16:01:37Z

+        // Preserve the schema identifier from the SchemaDefinition so the configured Schema Reference Writer can
+        // write the correct schema id in the Confluent header.


Useful comment, but better avoid referencing Confluent-specific details.

Suggested change

// Preserve the schema identifier from the SchemaDefinition so the configured Schema Reference Writer can

// write the correct schema id in the Confluent header.

// Preserve the schema identifier required by the configured Schema Reference Writer.

thanks, done

tpalfy · 2026-09-22T16:15:29Z

+        .description("Service used to write the Confluent message index array identifying the target message within the schema, written after the Schema Reference Writer header. "
+            + "Applicable only when producing Confluent wire-format content.")


Better avoid Confluent specific details.

Suggested change

.description("Service used to write the Confluent message index array identifying the target message within the schema, written after the Schema Reference Writer header. "

+ "Applicable only when producing Confluent wire-format content.")

.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.")

thanks, done

tpalfy · 2026-09-22T16:26:43Z

+    private ProtobufWriteContext createWriteContext(final Map<String, String> variables) throws SchemaNotFoundException, IOException {
+        final SchemaDefinition schemaDefinition = createSchemaDefinition(variables);
+        final Schema schema = schemaCompiler.compileOrGetFromCache(schemaDefinition);
+        final MessageName messageName = messageNameResolver.getMessageName(variables, schemaDefinition, EMPTY_INPUT_STREAM);


While technically correct because the 0-length byte array behaves like an immutable object and can be reused safely, it is surprising and gives the code-reader a pause. That being said, I'm okay keeping it like this - but let me offer an alternative:

Suggested change

final MessageName messageName = messageNameResolver.getMessageName(variables, schemaDefinition, EMPTY_INPUT_STREAM);

final MessageName messageName = messageNameResolver.getMessageName(variables, schemaDefinition, InputStream.nullInputStream());

yes, totally agree, changed it

tpalfy · 2026-09-23T13:37:18Z

+@CapabilityDescription("""
+    Serializes NiFi Records into Protocol Buffers binary format. \
+    Supports inline schema text and schema registry lookup for determining the Proto schema. \
+    When a Schema Reference Writer is configured, a Confluent wire-format header is written; when a \
+    Message Index Writer is also configured, the Confluent message index array is written after the header. \
+    The target Proto message name can be determined statically using the 'Message Name' property, \
+    or dynamically using a Message Name Resolver service.
+    A single record is written per FlowFile, since concatenated Protocol Buffers messages cannot be delimited. \
+    The 'google.protobuf.Any' well-known type is not expanded on write; a Record derived from an Any-typed message \
+    is serialized as an ordinary nested message rather than being re-wrapped as an Any.""")


Avoid referencing Confluent as part of the generic capability description. A reminded-type mentioning of the need of both services for Confluent is okay here though.

Suggested change

@CapabilityDescription("""

Serializes NiFi Records into Protocol Buffers binary format. \

Supports inline schema text and schema registry lookup for determining the Proto schema. \

When a Schema Reference Writer is configured, a Confluent wire-format header is written; when a \

Message Index Writer is also configured, the Confluent message index array is written after the header. \

The target Proto message name can be determined statically using the 'Message Name' property, \

or dynamically using a Message Name Resolver service.

A single record is written per FlowFile, since concatenated Protocol Buffers messages cannot be delimited. \

The 'google.protobuf.Any' well-known type is not expanded on write; a Record derived from an Any-typed message \

is serialized as an ordinary nested message rather than being re-wrapped as an Any.""")

@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.""")

done, thanks

tpalfy · 2026-09-23T14:19:33Z

+        schemaBranchName = context.getProperty(SCHEMA_BRANCH_NAME);
+        schemaVersion = context.getProperty(SCHEMA_VERSION);
+    }
+


An inline schema via Schema Text doesn't have a Confluent-compatible identifier.
Not sure if this is intended, but if it is, we can add a custom validate to make sure that when the ConfluentEncodedSchemaReferenceWriter is set, the Schema Name strategy is selected as well.

Also we can add validation for literal Schema Text values.

Suggested change

@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;

}

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;

}

try {

final SchemaIdentifier schemaIdentifier = SchemaIdentifier.builder()

.name(sha256Hex(schemaTextValue) + PROTO_EXTENSION)

.build();

final SchemaDefinition schemaDefinition = new StandardSchemaDefinition(

schemaIdentifier, schemaTextValue, SchemaDefinition.SchemaType.PROTOBUF);

final Schema compiledSchema = schemaCompiler.compileOrGetFromCache(schemaDefinition);

validateLiteralMessageName(validationContext, compiledSchema, results);

} catch (final SchemaCompilationException e) {

results.add(new ValidationResult.Builder()

.subject(SCHEMA_TEXT.getDisplayName())

.valid(false)

.explanation("Invalid Protocol Buffers schema: " + e.getMessage())

.build());

}

}

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());

}

}

Yes, that's intended: inline Schema Text has no registry identifier, so a Confluent header can't be written. Added your validation, with one small change: the catch is RuntimeException instead of only SchemaCompilationException, because the compiler can also throw other runtime exceptions and those should become a validation error too. Also added tests for an invalid schema text, an unknown message name, and Schema Text combined with the Confluent Schema Reference Writer

tpalfy · 2026-09-23T14:36:05Z

+
+    @Override
+    public RecordSetWriter createWriter(final ComponentLog logger, final RecordSchema schema, final OutputStream out, final Map<String, String> variables) throws SchemaNotFoundException, IOException {
+        final ProtobufWriteContext context = createWriteContext(variables);


Suggested change

final ProtobufWriteContext context = createWriteContext(variables);

final ProtobufWriteContext context = createWriteContext(variables);

if (schemaReferenceWriter != null) {

schemaReferenceWriter.validateSchema(context.recordSchema());

}

thanks, added

…d writer - Remove Confluent-specific details from generic descriptions and comments - Clarify Schema Reference Writer and Message Index Writer descriptions - Validate literal Schema Text and Message Name values - Validate that the Schema Access Strategy supplies the fields required by the Schema Reference Writer - Validate the schema with the Schema Reference Writer when creating a writer - Use InputStream.nullInputStream() for message name resolution Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tpalfy · 2026-09-24T14:35:49Z

+    @Override
+    public void close() throws IOException {
+        if (closed) {
+            return;
+        }
+        closed = true;
+
+        // Ensure buffered content is flushed to the underlying stream before it is closed, including the
+        // write-without-active-record-set path where onFinishRecordSet is never invoked.
+        flush();
+        super.close();
+    }


If flush() throws an exception, the super.close() won't get called. We can simply call close on the BufferedOutputStream. As that call is idempotent, we can call it as many times as we want. The closed flag becomes unnecessary.

Suggested change

@Override

public void close() throws IOException {

if (closed) {

return;

}

closed = true;

// Ensure buffered content is flushed to the underlying stream before it is closed, including the

// write-without-active-record-set path where onFinishRecordSet is never invoked.

flush();

super.close();

}

@Override

public void close() throws IOException {

buffered.close();

}

good point, thanks

prankstrisse · 2026-09-24T14:36:12Z

Thank you so much for the review, @tpalfy ! I pushed the fixes

tpalfy · 2026-09-24T14:59:24Z

+    @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). Only a single record per
+        // FlowFile can be represented, which also matches the Confluent one-message-per-record convention.
+        if (getRecordCount() > 0) {
+            throw new IOException("Protobuf output supports only a single record because concatenated Protobuf messages cannot be delimited");
+        }
+
+        // If we are not writing an active record set, then we need to ensure that the Confluent framing is written.
+        if (!isActiveRecordSet()) {
+            flush();
+            writeConfluentFraming(buffered);
+        }
+
+        final byte[] payload = serializer.serialize(record);
+        buffered.write(payload);
+        return getSchemaReferenceAttributes();
+    }


The framing handling is confusing, and can result in corrupted stream in edge cases (empty recordset, serialization failure).

The framing is tied to the single record, so we can simply remove onBeginRecordSet() and update this method like this:

Suggested change

@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). Only a single record per

// FlowFile can be represented, which also matches the Confluent one-message-per-record convention.

if (getRecordCount() > 0) {

throw new IOException("Protobuf output supports only a single record because concatenated Protobuf messages cannot be delimited");

}

// If we are not writing an active record set, then we need to ensure that the Confluent framing is written.

if (!isActiveRecordSet()) {

flush();

writeConfluentFraming(buffered);

}

final byte[] payload = serializer.serialize(record);

buffered.write(payload);

return getSchemaReferenceAttributes();

}

@Override

public Map<String, String> writeRecord(final Record record) throws IOException {

if (getRecordCount() > 0) {

throw new IOException("Protobuf output supports only a single record");

}

final byte[] payload = serializer.serialize(record);

writeFraming(buffered);

buffered.write(payload);

return getSchemaReferenceAttributes();

}

agreed, thanks. The framing is now written together with the payload, only after the record has been serialized, and onBeginRecordSet() is removed

tpalfy · 2026-09-24T15:30:40Z

+/**
+ * Writes Records as Protocol Buffers binary content. When a {@link SchemaReferenceWriter} is
+ * configured, a Confluent wire-format header (magic byte and schema identifier) is written first;
+ * when a {@link MessageIndexWriter} is configured, the Confluent message index array follows the
+ * header. The serialized Protobuf payload is written last.
+ * <p>
+ * The Confluent framing (header and message index) is written once at the beginning of the record
+ * set, mirroring {@code WriteAvroResultWithExternalSchema}; the typical Confluent use case writes a
+ * single message per FlowFile.
+ */


Avoid Confluent-specific details.

Suggested change

/**

* Writes Records as Protocol Buffers binary content. When a {@link SchemaReferenceWriter} is

* configured, a Confluent wire-format header (magic byte and schema identifier) is written first;

* when a {@link MessageIndexWriter} is configured, the Confluent message index array follows the

* header. The serialized Protobuf payload is written last.

* <p>

* The Confluent framing (header and message index) is written once at the beginning of the record

* set, mirroring {@code WriteAvroResultWithExternalSchema}; the typical Confluent use case writes a

* single message per FlowFile.

*/

/**

* 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.

*/

thanks, done

tpalfy · 2026-09-24T15:32:01Z

+        return schemaReferenceWriter == null ? Map.of() : schemaReferenceWriter.getAttributes(recordSchema);
+    }
+
+    private void writeConfluentFraming(final OutputStream out) throws IOException {


Suggested change

private void writeConfluentFraming(final OutputStream out) throws IOException {

private void writeFraming(final OutputStream out) throws IOException {

renamed, thanks

tpalfy · 2026-09-24T16:02:49Z

+            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);


Very minor: this exception can be due to encoding failure as well.

Suggested change

throw new IllegalStateException("Failed to parse protobuf schema", e);

throw new IllegalStateException("Failed to generate Protobuf message index", e);

good point, done

tpalfy · 2026-09-24T16:20:23Z

+          string street = 1;
+          string city = 2;
+        }""";
+


All nested indexes are zeros. We can add another test to cover a bit more complex case.

Suggested change

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 {}

}

}""";

added, thanks

tpalfy · 2026-09-24T16:21:40Z

+            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})
+        );


Suggested change

);

,

Arguments.of(NON_ZERO_NESTED_INDEX_SCHEMA, new StandardMessageName(Optional.of("com.example.nested"), "Root.ChildTwo.GrandchildThree"), new int[] {1, 2, 3})

)

added, thanks

…ess review comments - Write schema reference and message index information after the record is serialized, so an empty record set or a serialization failure does not produce framing without a payload - Close the buffered stream directly so the underlying stream is closed even when flushing fails - Describe the Protobuf writer framing without format-specific details - Report message index generation failures with an accurate message - Add a message index test with non-zero nested indexes

prankstrisse force-pushed the NIFI-14210 branch 4 times, most recently from 936d100 to 3caa6e9 Compare August 3, 2026 12:13

prankstrisse marked this pull request as ready for review August 3, 2026 14:53

tpalfy self-requested a review September 23, 2026 11:14

NIFI-14210 Add record writer for Protobuf messages

1f13973

prankstrisse force-pushed the NIFI-14210 branch from 3caa6e9 to 1f13973 Compare September 24, 2026 11:53

pvillard31 reviewed Sep 24, 2026

View reviewed changes

tpalfy reviewed Sep 24, 2026

View reviewed changes

		* Import paths may contain directories, such as {@code airlines/ph/cdm/shared.proto}, so the enclosing
		* directory structure has to exist before the file is written.

		// The Schema Reference Reader is a read-side concern: a writer determines its schema from the configured
		// access strategy and writes references through the Schema Reference Writer instead.

		// Only the strategies that createSchemaDefinition supports are offered; the inherited list also contains
		// the Schema Reference Reader strategy, which cannot be used to obtain a schema for writing.

		// Preserve the schema identifier from the SchemaDefinition so the configured Schema Reference Writer can
		// write the correct schema id in the Confluent header.

	// Preserve the schema identifier from the SchemaDefinition so the configured Schema Reference Writer can
	// write the correct schema id in the Confluent header.
	// Preserve the schema identifier required by the configured Schema Reference Writer.

		.description("Service used to write the Confluent message index array identifying the target message within the schema, written after the Schema Reference Writer header. "
		+ "Applicable only when producing Confluent wire-format content.")

-        .description("Service used to write the Confluent message index array identifying the target message within the schema, written after the Schema Reference Writer header. "
-            + "Applicable only when producing Confluent wire-format content.")
+.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.")

	final MessageName messageName = messageNameResolver.getMessageName(variables, schemaDefinition, EMPTY_INPUT_STREAM);
	final MessageName messageName = messageNameResolver.getMessageName(variables, schemaDefinition, InputStream.nullInputStream());

+@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;
+}
+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;
+    }
+    try {
+        final SchemaIdentifier schemaIdentifier = SchemaIdentifier.builder()
+            .name(sha256Hex(schemaTextValue) + PROTO_EXTENSION)
+            .build();
+        final SchemaDefinition schemaDefinition = new StandardSchemaDefinition(
+            schemaIdentifier, schemaTextValue, SchemaDefinition.SchemaType.PROTOBUF);
+        final Schema compiledSchema = schemaCompiler.compileOrGetFromCache(schemaDefinition);
+        validateLiteralMessageName(validationContext, compiledSchema, results);
+    } catch (final SchemaCompilationException e) {
+        results.add(new ValidationResult.Builder()
+            .subject(SCHEMA_TEXT.getDisplayName())
+            .valid(false)
+            .explanation("Invalid Protocol Buffers schema: " + e.getMessage())
+            .build());
+    }
+}
+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 writeConfluentFraming(final OutputStream out) throws IOException {
	private void writeFraming(final OutputStream out) throws IOException {

	throw new IllegalStateException("Failed to parse protobuf schema", e);
	throw new IllegalStateException("Failed to generate Protobuf message index", e);

+    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 {}
+          }
+        }""";

-        );
+        ,
+            Arguments.of(NON_ZERO_NESTED_INDEX_SCHEMA, new StandardMessageName(Optional.of("com.example.nested"), "Root.ChildTwo.GrandchildThree"), new int[] {1, 2, 3})
+        )

Conversation

prankstrisse commented Aug 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Changes

Tracking

Issue Tracking

Pull Request Tracking

Pull Request Formatting

Verification

Build

Licensing

Documentation

Uh oh!

prankstrisse commented Sep 24, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

prankstrisse commented Sep 24, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

prankstrisse commented Aug 3, 2026 •

edited

Loading