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:
- How do I write
snapshotConfigurationmethod? 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). - What are
ProtobufSerializerandProtobufTypeSerializerfor? Should I implement them instead ofTypeSerializerfor my needs? - I'm not sure I wrote
getLengthcorrectly. I assumed 4 doubles take 4 x 24 = 96 bytes, but protobuf messages are not constant length even in case of 4 doubles. - What happens with
TypeSerializermethods that accept another T and try to reuse it if T is immutable? In my case they throw an exception. Are they even called ifisImmutableType = true?