Flink: consider add serialVersionUID for serializable classes used by operator

[chenjunjiedada]

We met a serialization issue when using new Iceberg datastream sink on the flink cluster with old iceberg datastream sink:

Caused by: java.io.InvalidClassException: org.apache.iceberg.PartitionSpec; local class incompatible: stream classdesc serialVersionUID = -4610029347110790711, local class serialVersionUID = -2608482762645185351

@openinx @JingsongLi @rdblue, Do you think we should add serialVersionUID for members in RowDataTaskWriterFactory? Thoughts?

[rdblue]

Table should not be serializable. That would leak the requirement to external connectors.

This is happening when restoring from checkpoint state written by a different version, right? If that's the case, can we serialize checkpoint state with something more reliable than Java serialization?

[chenjunjiedada]

This can happen when restoring from checkpoint/savepoint state. This can also happen when Flink cluster upgraded with existing user applications as pointed out by @JingsongLi here and might happen when user wants to try out newly released iceberg-flink-runtime on the Flink cluster.

Flink is using Kryo for state serialization according to this. But, Flink uses java serialization to transport messages and programs through RPC according to doc. So I guess serializable classes in operator would be serialized and transfer to tasks via Java serialization and as a consequence, we have to define serialVersionUID.

@openinx @JingsongLi, Any comment?

[rdblue]

@openinx, @JingsongLi, I think that this is an important problem to address before the next release. I'm concerned about trying to maintain compatibility across Iceberg versions using Java serialization, especially if serialization is automatic. We discussed this in the Iceberg sync and most people there tended to agree that Java serialization is not a good candidate for serialization across library versions.

Is there an option to serialize checkpoint state using something not based on the Java classes? What about using a JSON format that we can control?

[JingsongLi]

Hi, I think there are two cases:

1. Restoring Flink job from checkpoint/savepoint state.
2. Flink cluster upgraded with existing user applications.

In Flink, for serializing checkpoint state:

• For known objects, such as POJO, List, etc., Flink uses handwritten serializers.
• For unknown objects, Flink uses Kryo serializer. [1]

So actually, Flink not use Java serialization to serialize checkpoint state. So I think case 1 may not be a valid problem. (If Iceberg objects are good to upgrade using Kryo)

Case 2 is indeed a problem, From another perspective, can we consider customizing the Java serializer of RowDataTaskWriterFactory?

[1]https://ci.apache.org/projects/flink/flink-docs-master/dev/types_serialization.html#general-class-types

[rdblue]

My understanding is that Kryo has the same general compatibility problems as Java serialization, it is just a bit faster. According to the Kryo docs, compatibility is not guaranteed and varies across serializers. And the serializer that this would use, FieldSerializer, has guarantees similar to Java serialization:

[FieldSerializer] does not support adding, removing, or changing the type of fields without invalidating previously serialized bytes.

That's why I think that anything that is serialized and later read by different versions of Iceberg should not use default Java or Kryo serialization. I think that means we need to have a format for Flink checkpoint and savepoint state. We should also have serialization tests.

For the second case, I'm not sure I understand what the concern is. Are the tasks themselves serialized and not just the checkpoint state in this case?

[stevenzwu]

Are the tasks themselves serialized and not just the checkpoint state in this case?

@rdblue Flink jobmanager (driver) serializes the tasks and ships them to taskmanagers (workers) to execute the tasks. Any final member of operator class needs to be serializable.

It is actually unclear to me what kind of serializable guarantee that Flink provides. Does it guarantee compatibility with different binary versions in the same Flink cluster (like in rolling upgrade case)? @JingsongLi any comment?

@chenjunjiedada were you doing a rolling upgrade of the flink/iceberg version when running into the serializable problem? We probably dodged the problem because we always do red/black deployment with standalone cluster mode. In another world, we never have different versions of binaries for a Flink cluster.

I took a look at RowDataTaskWriterFactory class.

• It contains a final member of FileAppenderFactory interface, which doesn't extend from Serializable. It seems that GenericAppenderFactory impl class is actually serializable. I am not sure about SparkAppenderFactory. @rdblue should FileAppenderFactory interface be serializable?
• It also contains final members for many other serializable Iceberg classes, like PartitionSpec, Schema, FileIO, EncryptionManager. I thought the error msg was complaining about the serialVersionUID change for org.apache.iceberg.PartitionSpec class. @JingsongLi By custom serializer for RowDataTaskWriterFactory, you also mean custom serializer for all the Iceberg classes as the final members for RowDataTaskWriterFactory, right?

@openinx @JingsongLi This does remind me the potential problem of using Iceberg DataFile interface for transmission between writer and committer. With the new unaligned checkpoints feature, those Kryo serialized bytes in network buffer can be included in the checkpoints/savepoints. if we restore the checkpoint/savepoint with a new Iceberg version, we may have a problem. I probably shouldn't depend on Kryo serializer for DataFile. We may need a FlinkDataFile class to guarantee/control the compatibility. I believe this is the concern Ryan has.

[rdblue]

@stevenzwu, I meant across versions, not within a running application. I'm concerned about cases where Iceberg is upgraded and can no longer read older state. I thought that was limited to checkpoint state, but there appears to be a second case.

Is the second case a rolling deployment? That would make sense and would be a problem when updating Iceberg for many more classes than just checkpoint state.

[rdblue]

FileAppenderFactory should not be Serializable because that requirement is for Flink but not other engines.

[JingsongLi]

State compatibility

Let's focus @rdblue 's concern first: where Iceberg is upgraded and can no longer read older state.
Currently, as @stevenzwu said, only the DataFile is in Flink state. And it is serialized by Kryo.
Although Kryo can avoid serialVersionUID, adding or reducing fields will also cause compatibility problems.

Here I suggest that we need to implement Flink serializer for DataFile.
https://ci.apache.org/projects/flink/flink-docs-master/dev/stream/state/custom_serialization.html#state-serializers-and-schema-evolution
When restoring from savepoints, Flink allows changing the serializers used to read and write previously registered state, so that users are not locked in to any specific serialization schema.

Job Topology compatibility

This means we want to upgrade an existing Flink Job to a new Iceberg version. As @stevenzwu said, premise is that we have different versions of binaries for a Flink cluster. I don't think this is very important, but, if we want to solve it, we can define custom serializer for all the Iceberg classes as the final members for RowDataTaskWriterFactory. But it's going to be a little cumbersome. @stevenzwu

[stevenzwu]

Regarding DataFile, in addition to the TypeSerializer (for state) we also need to implement the custom Kryo Serializers for the network transfer btw writer and committer, since the network buffer can be included in the checkpoints due to unaligned checkpoints.

Once @chenjunjiedada confirmed that it is a rolling upgrade case, maybe we should bring this to the Flink community and see if Flink provides any serialization compatibility guarantee for that operation mode. As @JingsongLi mentioned, It is cumbersome to implement custom serializers for all the Iceberg classes. It would be nice to avoid it.

[rdblue]

While we are considering the state problem, why not go ahead and update the sink to store all the DataFile objects in a ManifestFile and use @stevenzwu's code to serialize the manifests? Then we have more confidence because we've been running this in prod for a long time now. That avoids breaking state later to move to storing manifest files, fixes the current problem, and should make storing checkpoint state a bit faster because it would be smaller. What do you think?

[JingsongLi]

we also need to implement the custom Kryo Serializers for the network transfer btw writer and committer

We can use Flink DataFileTypeSerializer to network too.

+1 to store all the DataFile objects in a ManifestFile.

But I think it is good to introduce SimpleVersionedSerializers[1] for DataFile and ManifestFile too, we may need FlinkDataFile and FlinkManifestFile. The SimpleVersionedSerializer is a simpler design than state-serializers-and-schema-evolution, it is completely internal implementation in iceberg serializers.

[1]https://ci.apache.org/projects/flink/flink-docs-master/api/java/org/apache/flink/core/io/SimpleVersionedSerialization.html

[chenjunjiedada]

@stevenzwu ， The issue didn't happen when doing the rolling upgrade. As I mentioned at the beginning, the issue happens when I submit a new application with new Iceberg version to a Flink cluster which deployed an old version iceberg. After digging deeper, we found it is because the job manager and task manager still deserialize the old version iceberg classes from Flink dependency but Flink client uses the new iceberg class from the user application. This could be resolved by setting parameter yarn.per-job-cluster.include-user-jar=FIRST so that the job manager and task manager could load new version iceberg classes at first.

For upgraded Flink cluster with existing application, @JingsongLi, Do we expect user upload application along with the iceberg-flink-runtime jar? Or the Flink cluster provides the iceberg-flink-runtime jar? In our case, the Flink cluster has the iceberg-flink-runtime jar. When the Flink cluster upgraded, the user job possibly needs to restart so it would not subject the compatibility issue. When it is a rolling upgrade, it probably still doesn't needs to transfer iceberg class between job manager and task managers during job running.