Add the possibility to handle state evolution without breaking state compatibility

[arnaud-daroussin]

Hi @novakov-alexey,

I noticed several SerializerSnapshot cannot handle evolutions that break serialization backward compatibility because they don't serialize their version number. And so when reading back the serialized stream, they don't know which version of the serializer wrote it, to be able to react and recover from a previous version.

Identified serializer snapshots are:

• CollectionSerializerSnapshot
• CoproductSerializerSnapshot
• MappedSerializerSnapshot
• ScalaCaseObjectSerializerSnapshot

Flink provides 2 utility methods to handle this the right way without headache:

• TypeSerializerSnapshot.writeVersionedSnapshot
• TypeSerializerSnapshot.readVersionedSnapshot

An example on how to react on different version numbers can be found here: https://github.com/apache/flink/blob/release-1.20/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerSnapshot.java#L162

For exemple I think you could have introduced "class arity usage" feature in case class smoothly, without breaking state compatibility with previous versions. Like it is done here: https://github.com/apache/flink/blob/release-1.20/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RowSerializer.java#L501

Unfortunately, there is no possibility to add usage of version number without breaking state compatibility one last time, that's why I suggest to make this change starting with Flink 2 only, as state compatibility is not an option between Flink 1 and 2 anyway.

What do you think of this?

Do you want to do it yourself, or can I propose a PR?

Thanks

[novakov-alexey]

Hi @arnaud-daroussin,

I think it is great idea to leverage this concept of the versioned snapshot.
Please go ahead with proposing a PR. 👍

It would also great to improve that code related to the "class arity usage", even if breaks state one more time.
We can evaluate the impact. I think people already went over that breaking change on the case class arity.

[arnaud-daroussin]

I also think it's not necessary to come back on the class arity breaking change, it was one year ago. It was only a concrete example.

What do you think to be conservative and do these breaking changes only for Flink 2, or do you prefer to accept breaking changes also for Flink 1 (semantically it would be a version 1.20.1_2.0.0)?

[novakov-alexey]

Ok. We can try to make it first on Flink 2 and then evaluate whether it can be also applied to Flink 1.

Please check the latest releases I made the other days.
There is no suffix version in the release name anymore. Just a version of the library.

[arnaud-daroussin]

Hi @novakov-alexey,
I let you evaluate if you want it also for Flink 1.

[novakov-alexey]

Hi @arnaud-daroussin ,

I would do it for Flink 1.x if we can implement it with some feature flag to enable versioned snapshot.