Configure native Protobuf values#

The quickstart builds and runs the application examples. The public application entrypoints are ProtobufTypeInformation and ProtobufTypeInfoFactory. The serializer and snapshot classes are internal machinery; construct types through the public factory or builder.

Explicit type information#

Use explicit type information for .returns(...) and per-type settings. The following construction example uses the generated full-runtime Event class from the quickstart. The explicit example shows the imports for these types.

ProtobufTypeInformation comes from this library; Event comes from the application’s generated code. Replace the Event import and class reference with the application’s message type. The example’s event.proto sets java_package = "io.github.flink.gcp.protobuf.examples.generated" and java_multiple_files = true, so Event is a top-level class in that Java package.

Construct the default type or configure it through the builder:

var defaults = ProtobufTypeInformation.of(Event.class);
var configured = ProtobufTypeInformation.newBuilder(Event.class)
        .deterministicSerialization(true)
        .maxMessageSize(1024 * 1024)
        .recursionLimit(64)
        .build();

The defaults disable deterministic writing, allow a 64 MiB payload and parser recursion depth of 100, and enable large-payload aliasing . Numeric limits must be positive; build() validates settings and the generated class. of(Class<T>) matches Flink’s unbounded static factory signature, so unsupported classes can compile but fail with IllegalArgumentException at construction. The builder also enforces T extends Message at compile time. Use of(Event.class) or the builder for explicit native selection. The inherited of(TypeHint) overload follows Flink’s ordinary type extraction and can select a generic type when the factory is not registered. Builder reuse does not change previously built type information or serializers. These limits constrain accepted inputs, not total heap usage or stack capacity.

Serialization rejects a value that exceeds either limit with IOException before writing its frame. It also rejects a value with an unknown field that reuses the number of a field that the containing message declares, with a wire type that field accepts, naming the wire type, the field number, and the declared field. The reader would parse such a field into the declared one. A length-delimited value of a message or map field becomes that field’s message, whose nesting could exceed the recursion limit that the writer checked; any other value becomes that field’s value, which replaces a singular value already set, adds an element to a repeated field, selects another oneof member, or fails the read for invalid UTF-8 or a malformed packed value. A repeated numeric or enum field accepts both its packed and its unpacked encoding. Parsing never produces such a field, because the parser merges those bytes into the declared field; only a value whose unknown fields the application set or merged directly can carry one, for example unknown fields copied from a message of another type. An application that builds such values fails when serializing them. Serialization likewise rejects an unknown-field number above 536,870,911, the largest field number , because its wire tag would wrap to another field’s tag, which the reader then parses as that field, or to an invalid tag for field zero, which fails the read. Other unknown fields are written unchanged, and the reader keeps them as unknown fields: those with undeclared numbers, those with a wire type that the declared field does not accept, and everything inside unknown groups.

Memory of large decoded messages#

Aliasing of large payloads is enabled by default. A payload above 4 KiB is read into its own array, and decoded bytes fields and unknown length-delimited fields may share that array instead of copying it. Disable it with largePayloadAliasing(false) on the builder:

var copying = ProtobufTypeInformation.newBuilder(Event.class)
        .largePayloadAliasing(false)
        .build();

Payloads of at most 4 KiB never share storage, and the setting changes neither the wire bytes, state compatibility, nor decoded values.

Disable aliasing when an application keeps decoded values for long:

• A small ByteString that shares a large message’s array, such as a bytes field value, keeps that whole array while it is kept in a long-lived map, cache, or heap state. With aliasing enabled, copying that one value instead also works, for example with ByteString.copyFrom(value.asReadOnlyByteBuffer()).
• Large messages kept in heap state, such as HashMapStateBackend values updated from records or restored from a checkpoint, keep their arrays when they hold a value that shares it. When a message mixes such a value with large other content, such as proto3 strings or repeated numbers, the array also holds that content’s wire form in addition to its decoded form; how much more memory this takes depends on the record’s shape.

Records that are processed and then dropped, and values read from RocksDB state for one access, keep their arrays only as long as the application keeps the values, so aliasing costs no retained memory there.

HashMap state and operator state restored from a checkpoint or savepoint are read with the setting recorded when that state was saved, and RocksDB keyed state with the new setting. A changed setting reaches those restored heap values from the next restore after a new checkpoint or savepoint, for state that the restored job registers again.

Factory registration#

The example’s config.yaml contains ordinary Flink pipeline configuration . It holds the two options needed for this registration example; the library does not define a separate configuration-file format. Add these options to the Flink configuration used when constructing the job, or load a configuration directory explicitly as shown below. If pipeline.serialization-config already has entries for other types, add this entry to the existing list.

Apply the configuration before any type extraction, including TypeInformation.of, TypeHint, and source/operator construction:

pipeline.generic-types: false
pipeline.serialization-config:
  - com.google.protobuf.AbstractMessage: {type: typeinfo, class: io.github.flink.gcp.protobuf.ProtobufTypeInfoFactory}

What each entry means#

Use both fully qualified class names exactly as listed in the table; they are class names, not package prefixes or wildcard patterns. The class value is the factory, not ProtobufTypeInformation, ProtobufTypeSerializer, or an application’s generated message class. For example, the application’s message is io.github.flink.gcp.protobuf.examples.generated.Event, but that name does not replace either class name in this superclass registration. The same registration covers supported generated classes in other Java packages, so a separate entry for each message is unnecessary. The library jar, generated application classes, and matching protobuf-java runtime must be available on the application’s classpath; YAML does not install these dependencies. See Flink’s type information factory documentation for the underlying mechanism.

Load the configuration before constructing the job#

The factory class is named in YAML, so this Java code needs no ProtobufTypeInfoFactory import. The registration application reads the actual config.yaml before constructing its job:

var config = GlobalConfiguration.loadConfiguration(args[0]);
try (var env = StreamExecutionEnvironment.getExecutionEnvironment(config)) {
    env.setParallelism(2);
    messages(env)
            .rebalance()
            .map(event -> event.getAccount() + ":" + event.getAmount())
            .returns(Types.STRING)
            .print();
    env.execute("Registered Protobuf values");
}

The argument args[0] names the directory containing config.yaml, not the YAML file itself. In the quickstart , exec:exec@registered passes examples/datastream/config as an absolute path to the application. For another application, supply a configuration directory that exists where its entrypoint runs and pass the loaded Configuration to the environment before creating sources or operators. Loading YAML produces a Configuration; configuring the environment applies the factory registration. Placing a file beside a job jar without loading it does not apply these options.

Registration scope and per-type settings#

On Flink 1.20, put the YAML list in config.yaml. The legacy flink-conf.yaml parser does not read the list, and that file takes precedence when both files exist. Register against AbstractMessage: superclass lookup reaches generated classes, while registering the Message interface does not. The registry is process-global and offers no per-job isolation. Explicit construction neither registers the factory nor changes its defaults. The factory uses the default settings; per-type settings require explicit TypeInformation . Do not add maxMessageSize, recursionLimit, deterministicSerialization, or largePayloadAliasing to this YAML entry; configure those library settings through ProtobufTypeInformation.newBuilder(...).

Nested values and supported types#

Inference from an erased runtime List instance is not promised. Keep generic types disabled, including when explicit type information is used elsewhere in the job. The common-types guide covers generated Google Struct/Value/ListValue, Any, scalar/time wrappers, and OpenTelemetry AnyValue/ArrayValue/KeyValueList. Struct keys and AnyValue variants are data within a fixed schema; they do not require DynamicMessage. Any payload bytes remain opaque, without automatic unpacking or schema validation.

Every message type that a payload reaches, including types in imported files, must be declared in a file with syntax = "proto3". Construction rejects a payload graph that reaches a message declared in a proto2 or editions file, naming the first such message and its file; this includes google/protobuf/descriptor.proto types, while the well-known types such as Struct, Any, and Timestamp are proto3. Construction also rejects lite messages and payload messages that declare extensions. The library concerns internal DataStream transport and managed values, not connector boundary encoders, ProtoJSON, OTLP export, or the separate Flink Table/SQL Protobuf format.

DynamicMessage values#

The DynamicMessage class does not identify a schema, so DynamicMessage values need type information bound to a root descriptor:

ProtobufTypeInformation<DynamicMessage> info = ProtobufTypeInformation.of(descriptor);

ProtobufTypeInformation.newBuilder(descriptor) accepts the same four settings as the generated-class builder, and every operator that returns DynamicMessage needs .returns(info). See DynamicMessage values for the complete guide.

Scalar keys#

Messages are values, not supported DataStream partitioning keys or MapState user keys. Extract a stable scalar such as an account string. There is no message-key opt-in, including with deterministic serialization enabled. Both inferred and explicitly typed keyBy can bypass isKeyType(), so lack of an exception does not establish support. Deterministic bytes neither stabilize generated-message hashing across classloaders/JVMs nor define canonical key bytes across builds.

State and recovery#

The state example partitions by a scalar and assigns a stable operator UID. RunningTotal is a nested operator class defined in that example; its total and recent fields have types ValueState<Event> and MapState<String, Event> respectively. The linked source includes the complete operator and its Flink imports.

return events.keyBy(Event::getAccount, Types.STRING)
        .process(new RunningTotal())
        .returns(Types.STRING)
        .uid("account-totals");

It creates native value descriptors in the operator’s open method:

var eventType = ProtobufTypeInformation.of(Event.class);
total = getRuntimeContext().getState(new ValueStateDescriptor<>("total", eventType));
recent =
        getRuntimeContext()
                .getMapState(
                        new MapStateDescriptor<>("recent", Types.STRING, eventType));

MapState<String, Event> is supported: the map key is a string and its value is Protobuf. MapState<Event, ...> remains unsupported even if a particular runtime accepts its descriptor. The bounded example checks state updates during execution and may finish before a checkpoint could complete.

For a continuously running application using these types, enable periodic checkpointing with env.enableCheckpointing(...) and configure persistent checkpoint storage accessible to the cluster. Flink’s configured restart strategy can then recover the job from a completed checkpoint. For an intentional stop and restart, including a restart with a supported schema change, use Flink’s savepoint operations on that running application:

"$FLINK_HOME/bin/flink" stop --savepointPath "$SAVEPOINT_DIRECTORY" "$JOB_ID"
"$FLINK_HOME/bin/flink" run -s "$SAVEPOINT_PATH" -c com.example.Application application.jar

Here Application is the application’s entrypoint, not the bounded demonstration job. Keep the generated Java class and Protobuf message names, operator UIDs, state names, and matching runtime/artifact line unchanged. The schema may stay the same or change as listed in supported schema changes ; the restored job rejects other changes to the declarations that saved values reach. Reader size/depth limits may stay the same or increase; deterministic-writing mode and large-payload aliasing may change in either direction. A successful Protobuf parse alone does not establish state compatibility.

The MiniCluster recovery suite verifies completed-checkpoint recovery and separate-job savepoint restoration for HashMap and RocksDB, including continued processing. It also verifies restoration across supported schema changes and rejection of unsupported changes and reduced reader limits. It is separate from these application examples and the historical Spike 0 . Cross-Flink, cross-Protobuf, backend migration, and rescaling claims require their own evidence and are not established by these examples.

Development upgrades and API reference#

Current 1.0.0-SNAPSHOT development builds restore the supported schema changes. Read Protobuf/state compatibility before changing an application or dependency. Version 1.0.0, the first Maven Central release, is the stabilization point; see the compatibility from 1.0.0 .

Generate the API reference with mise x -- just docs-javadoc, then open target/apidocs/index.html. The documentation site includes this API reference for each retained version . The public documentation site is live and linked from README; final live verification after the search initialization repair remains tracked in #19 .