DynamicMessage values#

com.google.protobuf.DynamicMessage holds a Protobuf message whose schema comes from a descriptor at runtime rather than from a generated class. The library transports these values and keeps them in managed state when the application supplies type information bound to a root descriptor.

The DynamicMessage example is compiled and run with the other quickstart examples; the snippets below come from it. ADR-0005 records the design.

DynamicMessage and generated common types#

Several generated message types also carry data whose shape varies at runtime. They are a different tool: their schema is fixed and known to the library through their generated class.

The generated types use generated-class type information and the registered factory, as Supported types shows. Changing a Struct key, an Any payload, or the AnyValue variant changes data, so unchanged-schema state restoration covers it. For DynamicMessage, the descriptor is the schema, and restoring with another descriptor is a schema change .

Supply a descriptor#

An application needs a root Descriptor to create or parse DynamicMessage values in any case. The example builds one from a descriptor set in the job jar. Its build has protoc write that set, with every imported file, into the classes directory, which the jar packages as the resource /event.pb:

<plugin><groupId>io.github.ascopes</groupId><artifactId>protobuf-maven-plugin</artifactId>
    <configuration><protoc>${protobuf.version}</protoc>
        <outputDescriptorFile>${project.build.outputDirectory}/event.pb</outputDescriptorFile>
        <outputDescriptorIncludeImports>true</outputDescriptorIncludeImports>
    </configuration>
</plugin>

Outside Maven, protoc --include_imports --descriptor_set_out=event.pb event.proto writes such a set. The application then builds each file with the files that it imports, which the set lists before it:

FileDescriptorSet set;
try (InputStream input = DynamicExample.class.getResourceAsStream("/event.pb")) {
    if (input == null) {
        throw new IOException("Missing descriptor set resource /event.pb");
    }
    set = FileDescriptorSet.parseFrom(input);
}
// The set lists each file after the files that it imports.
Map<String, FileDescriptor> files = new HashMap<>();
for (FileDescriptorProto file : set.getFileList()) {
    FileDescriptor[] dependencies = new FileDescriptor[file.getDependencyCount()];
    for (int i = 0; i < dependencies.length; i++) {
        dependencies[i] = files.get(file.getDependency(i));
        if (dependencies[i] == null) {
            throw new IOException("Descriptor set lacks import " + file.getDependency(i));
        }
    }
    files.put(file.getName(), FileDescriptor.buildFrom(file, dependencies));
}
// The file name is relative to the proto source root; the message name is not qualified.
Descriptor event = files.get("event.proto").findMessageTypeByName("Event");

A file’s name in the set is its path relative to the proto source root, such as event.proto, and findMessageTypeByName takes the message name without its package. A descriptor that a generated class supplies also works, but then the generated class itself is the simpler type.

Explicit type information#

ProtobufTypeInformation.of(descriptor) creates the type information, and ProtobufTypeInformation.newBuilder(descriptor) accepts the four settings of the generated-class builder. Give it to every operator that returns DynamicMessage values:

ProtobufTypeInformation<DynamicMessage> eventType =
        ProtobufTypeInformation.of(eventDescriptor());
SingleOutputStreamOperator<DynamicMessage> events =
        lines.map(new ParseEvent()).returns(eventType);

Creating the type information checks the descriptor. Every message type that the root reaches must be declared in a proto3 file, as for generated messages , with no extensions or extension ranges, no group or required field, and no field number above 536,870,911. protoc rejects groups, required fields, extension ranges, and such field numbers in proto3 files, but descriptor building accepts them in a descriptor built by hand. It also rebuilds the descriptors from their serialized form, as a TaskManager does, so a descriptor that cannot be rebuilt from its own content, such as one built against missing imports, fails here. A failure throws IllegalArgumentException that names the root and the failed rule.

Flink cannot infer this type. With the registered factory, an operator that returns DynamicMessage without .returns(...) fails with Flink’s missing-type error, whose cause names ProtobufTypeInformation.of(Descriptor); without the factory, Flink selects a generic type, which fails with generic types disabled. ProtobufTypeInformation.of(DynamicMessage.class) fails as well, because the class does not name a schema. A POJO or tuple that contains a DynamicMessage needs explicit composite type information, such as Types.POJO(MyPojo.class, Map.of("message", eventType)) or Types.TUPLE(Types.STRING, eventType).

Type information is Java-serializable and identified by the root’s full name, the descriptor content, and the settings, not by descriptor instances; a DynamicMessage type never equals the generated type of the same schema.

Values and field access#

A Descriptor is not serializable, so a function that creates values obtains its own descriptor, here when it opens:

/** Parses {@code account,amount} lines with a descriptor that each task loads itself. */
public static final class ParseEvent extends RichMapFunction<String, DynamicMessage> {
    private transient Descriptor event;

    @Override
    public void open(OpenContext context) throws Exception {
        event = eventDescriptor();
    }

    @Override
    public DynamicMessage map(String line) {
        String[] fields = line.split(",");
        return DynamicMessage.newBuilder(event)
                .setField(event.findFieldByName("account"), fields[0])
                .setField(event.findFieldByName("amount"), Long.parseLong(fields[1]))
                .build();
    }
}

DynamicMessage field access, equality, and merging compare descriptor instances. Decoded values carry the descriptor instances of the serializer that read them, and another task, serializer, or restore has other instances. Resolve field descriptors through the value that you read or write, and build a nested value from that field’s getMessageType():

FieldDescriptor amount = event.getDescriptorForType().findFieldByName("amount");
long sum = (Long) event.getField(amount);
DynamicMessage previous = total.value();
if (previous != null) {
    FieldDescriptor saved = previous.getDescriptorForType().findFieldByName("amount");
    sum += (Long) previous.getField(saved);
}
total.update(event.toBuilder().setField(amount, sum).build());

A field descriptor from another instance fails in getField.

When writing, the serializer accepts a value whose descriptors are other instances if their content is the same as its own schema or a supported earlier schema of it, as restored values have after a schema change. It rejects other values with IOException, before writing any byte, naming the field path and the first incompatibility, such as a value of another message type in the stream. A bytes field set from a byte[] keeps that array without copying it, so treat the array as immutable afterwards. Deterministic serialization does not order DynamicMessage map entries; both supported runtimes write them in insertion order.

In the 1.0.0 release-candidate benchmarks , decoding the benchmark’s scalar, nested, and collections records as DynamicMessage values took 5.3 to 6.3 times as long as decoding the generated messages with the same content, and writing them took 1.5 to 1.8 times as long. The serializer decodes both with the parser of the value’s type, so the decoding difference is the cost of protobuf-java’s DynamicMessage parsing. Writing values whose descriptors are equal-content instances from elsewhere took about as long as writing values with the serializer’s own descriptors; the measured differences stayed within the variation between adjacent benchmark lanes.

Keys and state#

Key streams and MapState by stable scalar values extracted from the message, as for generated messages:

return events.keyBy(
                event ->
                        (String)
                                event.getField(
                                        event.getDescriptorForType()
                                                .findFieldByName("account")),
                Types.STRING)
        .process(new RunningTotal(eventType))
        .returns(Types.STRING)
        .uid("account-totals");

DynamicMessage keys are not supported: their hash codes include the descriptor instance’s identity, which differs between the tasks that send and receive a record. See scalar keys . State descriptors take the same type information as the stream:

private final ProtobufTypeInformation<DynamicMessage> eventType;
private transient ValueState<DynamicMessage> total;

public RunningTotal(ProtobufTypeInformation<DynamicMessage> eventType) {
    this.eventType = eventType;
}

@Override
public void open(OpenContext context) {
    total = getRuntimeContext().getState(new ValueStateDescriptor<>("total", eventType));
}

State and restore#

DynamicMessage state uses its own serializer snapshot, which records the root’s full name, the complete normalized descriptor set, and the settings. A restore rebuilds the saved descriptors without the application’s descriptor or any application class. The new application’s type information then decides whether it can read the saved state, by the schema-change rules of generated messages and the same rules for the size and depth limits and the other settings, except that no Java class name is involved: the root’s full name must stay the same.

A restore that these rules reject fails with Flink’s StateMigrationException for an incompatible new serializer, which names no reason. Check a schema change against the rules above before deploying it, and test the restore with a savepoint as the upgrade procedure describes.

After a supported schema change, restored values do not all carry the new descriptor:

• Values from operator state, and ValueState and MapState values on the HashMap backend, carry the saved descriptor. A field that only the new schema declares has no field descriptor in them, so findFieldByName returns null.
• RocksDB reads keyed state with the new application’s serializer, so its values carry the new descriptor.
• Values that the new serializer writes and the next restore reads carry the new descriptor.

Saved generated-message state does not restore into a DynamicMessage type, and saved DynamicMessage state does not restore into a generated type, even for the same schema; switching between them requires fresh state or replayed input. Restoring DynamicMessage state into a generated type, or generated state into a DynamicMessage type whose application still contains the generated class, fails with StateMigrationException. When the application no longer contains the generated class, restoring generated state fails earlier, because the saved class cannot load.

The DynamicMessage recovery tests verify checkpoint recovery, savepoint restores with unchanged and supported schemas, the descriptors that restored values carry, and the rejections above on HashMap and RocksDB, with a DynamicMessage application whose jar contains no generated class.

Scope and exclusions#

A stream or state whose values use several root schemas is not supported; use one type information object per schema. The library also does not discover descriptors automatically, integrate a schema registry, support lite messages, or use an extension registry.