Ensure state schema evolution for protobuf state objects in flink

Viewed 111

I'm trying to figure out how to enable schema evolution for protobuf state. Our flink program is written in Scala 2.11, we are using flink 1.10.1 and scalapb for protobuf code generation. RocksDB is used as a state backend.

I'm writing a TypeSerializer for a protobuf message consisting of 4 double fields. This is to be a PoC, so the message format I chose for this is purposefully simple.

This aspect of flink seems to be underdocumented. I implemented as best I could all the required methods in TypeSerializer except snapshotConfiguration. Halfway through I found ProtobufSerializer, ProtobufTypeSerializer and ProtobufTypeSerializerSnapshot classes in java API doc which were not mentioned in articles at all.

Now, specific questions:

  1. How do I write snapshotConfiguration method? I'm drawing a blank here, especially on what to include into serializer snapshot in order to be able to determine compatibility between serializers (TypeSerializerSnapshot#resolveSchemaCompatibility).
  2. What are ProtobufSerializer and ProtobufTypeSerializer for? Should I implement them instead of TypeSerializer for my needs?
  3. I'm not sure I wrote getLength correctly. I assumed 4 doubles take 4 x 24 = 96 bytes, but protobuf messages are not constant length even in case of 4 doubles.
  4. What happens with TypeSerializer methods that accept another T and try to reuse it if T is immutable? In my case they throw an exception. Are they even called if isImmutableType = true?
0 Answers
Related