flink-protobuf

Protobuf serialization for the Apache Flink DataStream API.

Flink ships no operator-to-operator serializer for protobuf messages. flink-avro integrates Avro into Flink's type system (AvroSerializer, AvroTypeInfo, state schema evolution), but protobuf has no equivalent — the official flink-protobuf module is a Table/SQL format (RowData ↔ bytes) and does not touch the DataStream path. Without help, every protobuf message in a DataStream job silently falls back to Kryo generic serialization.

This project fills the gap:

• flink-protobuf-serde/ — the library. ProtoSerializer (a TypeSerializer doing Message.toByteArray() / Parser.parseFrom(byte[])), ProtoTypeInfo, and ProtoFactory for global registration via pipeline.serialization-config, generic over any com.google.protobuf.Message. Targets Flink 1.20, Java 11. See its README.md for publishing.
• flink-protobuf-serde-1-18-test/ — a real-world test job proving the serializer works on the wire on Flink 1.18; see its README.md.
• flink-protobuf-serde-1-20-test/ — the same proof on Flink 1.20, additionally exercising global registration via the FLIP-398 pipeline.serialization-config route; see its README.md.

Using flink-protobuf-serde in your project

pom.xml

Add to your pom.xml:

Flink 1.20

<dependencies>
    <dependency>
        <groupId>com.michalklempa</groupId>
        <artifactId>flink-protobuf-serde</artifactId>
        <version>1.0.3-1.20</version>
    </dependency>
</dependencies>

Flink 1.18

<dependencies>
    <dependency>
        <groupId>com.michalklempa</groupId>
        <artifactId>flink-protobuf-serde</artifactId>
        <version>1.0.2-1.18</version>
    </dependency>
</dependencies>

The library declaresflink-coreandprotobuf-javaasprovided` — your job supplies its own Flink and protobuf versions.

Wiring into the application

Flink 1.19+

Global registration (Flink 1.19+) — the FLIP-398 pipeline.serialization-config option. One entry registering ProtoFactory for com.google.protobuf.AbstractMessage (the superclass of every generated message) covers all protobuf types: plain TypeInformation.of(YourMessage.class), lambda return-type extraction and POJO fields all resolve to ProtoTypeInfo automatically — no per-stream wiring.

In the cluster config.yaml (zero code; the option needs the standard YAML parser, i.e. config.yaml, not the legacy flink-conf.yaml):

pipeline.serialization-config:
  - com.google.protobuf.AbstractMessage: {type: typeinfo, class: com.michalklempa.flink.protobuf.ProtoFactory}

Or programmatically, through the same public option:

Configuration configuration = new Configuration();
configuration.set(PipelineOptions.SERIALIZATION_CONFIG, List.of(
        "com.google.protobuf.AbstractMessage: {type: typeinfo, class: com.michalklempa.flink.protobuf.ProtoFactory}"));

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);

There is deliberately no register()-style code route in this library: the only programmatic seam, TypeExtractor.registerFactory, is @Internal API. The configuration route reaches the same registry through stable public surface — it is exactly what FLIP-398 designed for this purpose.

In Flink 1.19+, the older Flink 1.18 way also works.

Flink 1.18+

On Flink 1.18, protobuf fields require explicit type information — there is no autodiscovery. There are two ways to specify it:

1. Explicit ProtoTypeInfo.proto() for direct protobuf streams

When a stream contains protobuf messages directly, attach ProtoTypeInfo.proto(...) to source constructors and .returns(...) after lambdas:

GeneratorFunction<Long, LatLng> generator = index -> LatLng.newBuilder()
        .setLatitude(48.14 + index * 0.0001)
        .setLongitude(17.10 + index * 0.0002)
        .build();

DataGeneratorSource<LatLng> source = new DataGeneratorSource<>(
        generator,
        Long.MAX_VALUE,
        RateLimiterStrategy.perSecond(1),
        ProtoTypeInfo.proto(LatLng.class));

DataStream<LatLng> positions =
        env.fromSource(source, WatermarkStrategy.noWatermarks(), "position");

After lambdas whose return type Flink cannot infer:

DataStream<LatLng> positions = input
        .map(value -> toLatLng(value))
        .returns(ProtoTypeInfo.proto(LatLng.class));

2. Annotation-based discovery for POJO fields with protobuf

When a protobuf message is a field in a regular Java POJO, annotate it with @TypeInfo(ProtoFactory.class). Flink discovers this annotation via reflection and applies the protobuf serializer to that field:

public class PositionEvent {
    public long timestamp;
    
    @TypeInfo(ProtoFactory.class)
    public LatLng latLng;
    
    public PositionEvent() {}
    
    public PositionEvent(long timestamp, LatLng latLng) {
        this.timestamp = timestamp;
        this.latLng = latLng;
    }
}

When your stream produces PositionEvent objects (via a map, filter, or other transformation), Flink automatically discovers the @TypeInfo annotation on the latLng field and applies ProtoFactory to create the correct serializer:

DataStream<LatLng> latLngs = /* source of LatLng protobuf messages */;

DataStream<PositionEvent> events = latLngs.map(latLng -> 
    new PositionEvent(System.currentTimeMillis(), latLng)
);

The annotation eliminates the need for .returns(ProtoTypeInfo.proto(...)) when the field type is already known through the POJO class structure.

Verifying the serde is actually used

Flink falls back to Kryo silently — a job can run fine while never touching this library. To verify the wiring, disable generic types:

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.getConfig().disableGenericTypes();

The config form is pipeline.generic-types: false. With this set, any stream that would use Kryo fails fast at job translation:

UnsupportedOperationException: Generic types have been disabled in the ExecutionConfig
and type com.google.type.LatLng is treated as a generic type.

If the job starts, every protobuf record on the wire goes through ProtoSerializer. Keeping the flag on permanently is a good guard against accidental Kryo regressions.

Status

Operator-to-operator serialization is implemented and verified on Flink 1.18 and 1.20 (both the configuration route and explicit wiring on 1.20). The 1.18-built jar (1.0.0-1.18-SNAPSHOT) also runs unchanged on Flink 1.20 — 1.20's default createSerializer(SerializerConfig) delegates to the still-abstract createSerializer(ExecutionConfig); this build overrides the SerializerConfig variant directly, which is what Flink 2.x requires (the ExecutionConfig overload is removed there). Keyed/operator state, schema evolution are out-of-scope of this project. Do not use Protobuf for Flink state.

Related work

Existing approaches to protobuf-in-Flink differ in where the serializer hooks into the stack, and that placement decides everything downstream: performance envelope, state compatibility, and whether Flink's Kryo kill-switch can be used at all.

Library	Hook level	disableGenericTypes() usable	Notes
twitter/chill (chill-protobuf)	Kryo default serializer	❌	registerTypeWithKryoSerializer(...); Kryo 2.x, matches Flink 1.x
magro/kryo-serializers (de.javakaffee)	Kryo default serializer	❌	same approach; recent releases target Kryo 5, incompatible with Flink 1.x's Kryo 2.24
findify/flink-protobuf	TypeInformation	✅	Scala/ScalaPB-first, Flink 1.13 era, dormant since 2021
Apache flink-protobuf (official)	Table/SQL format	n/a	RowData ↔ bytes for SQL connectors; does not touch the DataStream path
Apache flink-avro	TypeInformation	✅	the archetype — Avro only, includes state schema evolution
this project	TypeInformation	✅	plain Java, any Message, Flink 1.18 & 1.20

Kryo-hook family (chill-protobuf, de.javakaffee's ProtobufSerializer): both wrap toByteArray()/parseFrom() in a com.esotericsoftware.kryo.Serializer registered via env.getConfig().registerTypeWithKryoSerializer(...). The payload bytes are protobuf, but the message type remains a generic type to Flink — records still travel through KryoSerializer as the outer envelope. The consequence is architectural, not cosmetic: calling env.getConfig().disableGenericTypes() makes such jobs fail, because the setting forbids exactly the code path these libraries live on. You must keep Kryo fallback enabled globally, which means a forgotten registration for some other type degrades silently instead of failing fast. There is also a version-coupling cost: the Kryo serializer API must match Flink's bundled Kryo (2.24 for Flink 1.x), which rules out current de.javakaffee releases (Kryo 5). Performance-wise it should not matter if coupling is done via Kryo or directly on TypeInformation, same Protobuf toBytes() is used in both cases.

TypeInformation family (flink-avro, findify, this project): the serializer plugs into Flink's own type system as a first-class TypeSerializer, so protobuf types stop being generic types altogether. disableGenericTypes() becomes usable as a guard — the design goal here: the strictest serialization setting stays on, protobuf flows through a purpose-built serializer, and anything that would silently fall back to Kryo fails at translation time instead. Compared to findify's library, this project is plain-Java-first (no ScalaPB/Scala dependency, no default instance parameter at every call site).

License

Apache-2.0