Run a native Protobuf DataStream job#

Run the standalone application example against a local build of the library, installed into an isolated Maven repository, or against the published library . Maven Central publishes 1.0.0 for Flink 2.x and 1.0.0-1.20 for Flink 1.20 LTS.

Build the development artifact#

Install mise , then clone this repository. mise install provides Python 3.14 and uv for the recipe’s standard-library checks. Run these commands from its root with JDK 17:

mise trust
mise install
example_repo="$(mktemp -d /tmp/protobuf-example-maven.XXXXXX)"
mise x -- just examples-verify 2.2.1 4 "$example_repo"

Keep example_repo set for the following commands in the same shell. The recipe builds the Flink 2.x library with Protobuf 4 at its 2.2.1 compile floor, installs it locally, and compiles and tests the independent application against that jar. It also packages the application, launches all four entrypoints using the packaged classes, and checks the Maven launch commands below. It needs neither a running Flink cluster nor Docker. Initial dependency downloads require network access.

The application uses the normal Maven dependency:

<dependency>
    <groupId>io.github.flink-gcp</groupId>
    <artifactId>flink-datastream-protobuf</artifactId>
    <version>${protobuf.integration.version}</version>
</dependency>

The example POM currently sets protobuf.integration.version to 1.0.0-SNAPSHOT. The library declares Flink core and protobuf-java as provided dependencies, so the application explicitly supplies its Flink dependencies and Protobuf runtime. The example also generates its own Event message using the matching protoc version. It does not import this repository’s test messages or source directories.

Both development Flink adapter builds currently use the same SNAPSHOT coordinates. Use a separate Maven repository for each adapter/Protobuf-major combination; the recipe rejects reuse across those combinations. A SNAPSHOT version alone does not identify the writer of saved state: retain the exact source revision, jar checksum, schema, and runtime versions when keeping development state. See development baselines .

Run explicit type selection#

mise x -- ./mvnw -ntp -f examples/datastream/pom.xml \
  "-Dmaven.repo.local=$example_repo" exec:exec@explicit

The result includes alice:3 and bob:8; task prefixes and output order can vary. The linked explicit example contains the imports for the snippets below.

Event is generated from the example’s event.proto; replace its import with the generated message class from the application. The example disables generic types before creating the environment:

Configuration config = new Configuration();
config.set(PipelineOptions.GENERIC_TYPES, false);

It supplies native type information to the source:

var eventType = ProtobufTypeInformation.of(Event.class);
var events =
        env.fromData(
                eventType,
                Event.newBuilder().setAccount("alice").setAmount(2).build(),
                Event.newBuilder().setAccount("bob").setAmount(7).build());

The transformation declares the same native output type:

return events.map(event -> event.toBuilder().setAmount(event.getAmount() + 1).build())
        .returns(eventType);

See ExplicitExample.java for the complete executable entrypoint, including environment creation. The library has no internal Kryo fallback. Flink can select its own generic serializer when native integration is absent; disabling generic types makes that accidental selection fail.

Run registration and state examples#

Run the factory example; its Maven execution passes the directory containing the supplied config.yaml:

mise x -- ./mvnw -ntp -f examples/datastream/pom.xml \
  "-Dmaven.repo.local=$example_repo" \
  exec:exec@registered

The result includes alice:2 and bob:7. The Maven executions use exec:exec to launch a separate JVM with the application classpath, avoiding the in-process exec:java classloader. Each command launches a fresh JVM; factory registration is process-global. The usage guide explains registration timing and nested types.

Run the state example:

mise x -- ./mvnw -ntp -f examples/datastream/pom.xml \
  "-Dmaven.repo.local=$example_repo" \
  exec:exec@stateful

The result includes alice:2:0, alice:5:2, alice:9:3, and bob:7:0. Each record contains the account, running total, and previous event amount for that account. This bounded job demonstrates ValueState and MapState values; it is not a checkpoint/savepoint recovery test. See state and recovery for persistent-state usage and the separate recovery evidence.

Run the DynamicMessage example, which builds its descriptor from a descriptor set in the application jar instead of using the generated class:

mise x -- ./mvnw -ntp -f examples/datastream/pom.xml \
  "-Dmaven.repo.local=$example_repo" \
  exec:exec@dynamic

The result includes alice:2, alice:5, alice:9, and bob:7, the running total for each account. See DynamicMessage values for the descriptor, type information, and field-access rules that it follows.

Submit the packaged application#

The verified application jar is examples/datastream/target/example-job.jar. It contains the example classes, generated Event, the Event descriptor set event.pb, library, and unrelocated Protobuf runtime. It excludes Flink classes, which the matching Flink distribution supplies. Submit the explicit example to a running Flink 2.2.1 cluster with:

"$FLINK_HOME/bin/flink" run examples/datastream/target/example-job.jar

Use -c io.github.flink.gcp.protobuf.examples.StatefulExample for the state entrypoint and -c io.github.flink.gcp.protobuf.examples.DynamicExample for the DynamicMessage entrypoint. For the registration entrypoint, use -c io.github.flink.gcp.protobuf.examples.RegisteredExample and pass examples/datastream/config after the jar path; that directory must be available where the entrypoint executes. Avoid supplying a second, conflicting Protobuf runtime through the cluster classpath. The tests launch the packaged entrypoints locally with Flink dependencies; they do not deploy a remote cluster.

Select another tested runtime#

For example, use a new repository to verify Flink 1.20 with Protobuf 3:

lts_repo="$(mktemp -d /tmp/protobuf-example-lts.XXXXXX)"
mise x -- just examples-verify 1.20.5 3 "$lts_repo"
mise x -- ./mvnw -ntp -f examples/datastream/pom.xml \
  "-Dmaven.repo.local=$lts_repo" -Dflink.version=1.20.5 \
  -Dprotobuf.version=3.25.9 exec:exec@explicit

Keep the same version overrides when running the other entrypoints. Clean and rebuild when switching runtime/gencode combinations; the recipe does this for both projects. The recipe always compiles the library with Protobuf 4: a 1.20 library for LTS, and a floor-built 2.x library for either supported 2.x consumer runtime. A Protobuf 3 application therefore runs the same library jar as a Protobuf 4 application. Mixed generated-code/runtime pairs and cross-version state restoration are outside this verification.

Run against the published library#

The application resolves the library by its Maven coordinates, so it runs against the published library without a source or classpath change. Pass the published version of the runtime’s artifact line as the recipe’s fourth argument, with a new Maven repository:

mise x -- just examples-verify 2.2.1 4 "$(mktemp -d /tmp/protobuf-example-maven.XXXXXX)" 1.0.0
mise x -- just examples-verify 1.20.5 3 "$(mktemp -d /tmp/protobuf-example-maven.XXXXXX)" 1.0.0-1.20

The recipe then builds and installs no library: the application resolves the published jar from Maven Central, the recipe compares the jar and its POM with Central’s checksums, and runs the same tests and launch goals. In your own application, depend on 1.0.0 for Flink 2.x or 1.0.0-1.20 for Flink 1.20 LTS, and stay on that artifact line . Version 1.0.0 reads no state written by a 0.x milestone, and state from a development build carries no guarantee; consult the compatibility guidance first.