Ivan San Jose created BEAM-10223:
------------------------------------
Summary: AvroCoder has references to the encoded/decoded class
Key: BEAM-10223
URL: https://issues.apache.org/jira/browse/BEAM-10223
Project: Beam
Issue Type: Bug
Components: sdk-java-core
Affects Versions: 2.22.0
Reporter: Ivan San Jose
Hi, may be the JIRA issue title is not so much descriptive, but I couldn't find
anything better sorry.
Let me explain the problem:
When using Flink as runner, Beam coders are wrapped into Flink's
TyperSerializers, and, according to
https://github.com/apache/flink/blob/master/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerConfigSnapshot.java#L95
, those coders are java serialized inside the checkpoint.
The problem is that coders have a reference to the class which is being
coded/decoded (Class<T>), so when the coder is serialized, is done along the
POJO model, using Java serialization.
Why this is a problem?
This is a problem is you are working with checkpoints, because checkpoint
restoring will be broken as soon as you change anything in POJO model (even you
are using a coder which supports schema evolution and the change is following
its evolution rules):
{code}
Caused by: java.io.InvalidClassException:
internal.model.dimension.POJODimension; local class incompatible: stream
classdesc serialVersionUID = -223148029368332375, local class serialVersionUID
= 4489864664852536553
at java.io.ObjectStreamClass.initNonProxy(ObjectStreamClass.java:699)
at
java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1884)
at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1750)
at java.io.ObjectInputStream.readClass(ObjectInputStream.java:1715)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1555)
at
java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2286)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2210)
at
java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2068)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1572)
at
java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2286)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2210)
at
java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2068)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1572)
at
java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2286)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2210)
at
java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2068)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1572)
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:430)
at
org.apache.flink.api.common.typeutils.TypeSerializerSerializationUtil$TypeSerializerSerializationProxy.read(TypeSerializerSerializationUtil.java:301)
at
org.apache.flink.api.common.typeutils.TypeSerializerSerializationUtil.tryReadSerializer(TypeSerializerSerializationUtil.java:116)
at
org.apache.flink.api.common.typeutils.TypeSerializerConfigSnapshot.readSnapshot(TypeSerializerConfigSnapshot.java:113)
at
org.apache.flink.api.common.typeutils.TypeSerializerSnapshot.readVersionedSnapshot(TypeSerializerSnapshot.java:174)
at
org.apache.flink.api.common.typeutils.TypeSerializerSnapshotSerializationUtil$TypeSerializerSnapshotSerializationProxy.deserializeV2(TypeSerializerSnapshotSerializationUtil.java:179)
at
org.apache.flink.api.common.typeutils.TypeSerializerSnapshotSerializationUtil$TypeSerializerSnapshotSerializationProxy.read(TypeSerializerSnapshotSerializationUtil.java:150)
at
org.apache.flink.api.common.typeutils.TypeSerializerSnapshotSerializationUtil.readSerializerSnapshot(TypeSerializerSnapshotSerializationUtil.java:76)
at
org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshotReadersWriters$CurrentReaderImpl.readStateMetaInfoSnapshot(StateMetaInfoSnapshotReadersWriters.java:219)
at
org.apache.flink.runtime.state.OperatorBackendSerializationProxy.read(OperatorBackendSerializationProxy.java:119)
at
org.apache.flink.runtime.state.OperatorStateRestoreOperation.restore(OperatorStateRestoreOperation.java:83)
... 13 more
{code}
So, in order to be able to use existing checkpoints after compatible changes
has been done in the Java model, references to Class<T> in Beam Coder should be
removed.
Note that this JIRA ticket is only referencing to AvroCoder, and is the only
one fixed into GitHub related pull request.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
