[
https://issues.apache.org/jira/browse/FLINK-20890?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Felipe Silvestre Santos de Morais updated FLINK-20890:
------------------------------------------------------
Description:
When a savepoint is triggered for a regular Flink Job with a keyed function,
the key is serialized with
{noformat}
flink.api.common.typeutils.base.IntSerializer{noformat}
and the value serialized with
{noformat}
flink.api.scala.typeutils.ScalaCaseClassSerializerSnapshot{noformat}
When the savepoint is loaded with the state processor api, transformed, and
rewritten to the disk, the serializer taken is different.
Key:
{noformat}
org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializerSnapshot{noformat}
Now when the savepoint written by the states processor api is loaded, there is
the exception:
{code:java}
Caused by: org.apache.flink.util.StateMigrationException: The new key
serializer (org.apache.flink.api.common.typeutils.base.IntSerializer@11a7ba62)
must be compatible with the previous key serializer
(org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer@3b56cc30).Caused
by: org.apache.flink.util.StateMigrationException: The new key serializer
(org.apache.flink.api.common.typeutils.base.IntSerializer@11a7ba62) must be
compatible with the previous key serializer
(org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer@3b56cc30). at
org.apache.flink.runtime.state.heap.HeapRestoreOperation.restore(HeapRestoreOperation.java:147)
at
org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.build(HeapKeyedStateBackendBuilder.java:114)
... 15 more
{code}
State processor api should use the same serializer of Flink since the type is
exactly the same.
I have attached a zip that contains the code to test it.
In the project zipped there are the source and rewritten savepoints.
Note:
I have tried to play with enableX/disableX serializer, but so far no success.
was:
When a savepoint is triggered for a regular Flink Job with a keyed function,
the key is serialized with
{noformat}
flink.api.common.typeutils.base.IntSerializer{noformat}
and the value serialized with
{noformat}
flink.api.scala.typeutils.ScalaCaseClassSerializerSnapshot{noformat}
When the savepoint is loaded with the state processor api, transformed, and
rewritten to the disk, the serializer taken is different.
Key:
{noformat}
org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializerSnapshot{noformat}
Now when the savepoint written by the states processor api is loaded, there is
the exception:
{code:java}
Caused by: org.apache.flink.util.StateMigrationException: The new key
serializer (org.apache.flink.api.common.typeutils.base.IntSerializer@11a7ba62)
must be compatible with the previous key serializer
(org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer@3b56cc30).Caused
by: org.apache.flink.util.StateMigrationException: The new key serializer
(org.apache.flink.api.common.typeutils.base.IntSerializer@11a7ba62) must be
compatible with the previous key serializer
(org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer@3b56cc30). at
org.apache.flink.runtime.state.heap.HeapRestoreOperation.restore(HeapRestoreOperation.java:147)
at
org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.build(HeapKeyedStateBackendBuilder.java:114)
... 15 more
{code}
State processor api should use the same serializer of Flink since the type is
exactly the same.
I have attached a zip that contains the code to test it.
In the project zipped there are the source and rewritten savepoints.
Note:
I have tried to play with enableX/disableX serializer, but so far no success.
> flink-state-processor-api: differents serializers being taken from Flink &
> State processor api
> ----------------------------------------------------------------------------------------------
>
> Key: FLINK-20890
> URL: https://issues.apache.org/jira/browse/FLINK-20890
> Project: Flink
> Issue Type: Bug
> Components: API / State Processor
> Affects Versions: 1.12.0
> Reporter: Felipe Silvestre Santos de Morais
> Priority: Major
> Fix For: 1.12.0
>
> Attachments: flinkschemaevolution2.zip
>
>
> When a savepoint is triggered for a regular Flink Job with a keyed function,
> the key is serialized with
> {noformat}
> flink.api.common.typeutils.base.IntSerializer{noformat}
> and the value serialized with
> {noformat}
> flink.api.scala.typeutils.ScalaCaseClassSerializerSnapshot{noformat}
>
> When the savepoint is loaded with the state processor api, transformed, and
> rewritten to the disk, the serializer taken is different.
> Key:
> {noformat}
> org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializerSnapshot{noformat}
>
> Now when the savepoint written by the states processor api is loaded, there
> is the exception:
> {code:java}
> Caused by: org.apache.flink.util.StateMigrationException: The new key
> serializer
> (org.apache.flink.api.common.typeutils.base.IntSerializer@11a7ba62) must be
> compatible with the previous key serializer
> (org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer@3b56cc30).Caused
> by: org.apache.flink.util.StateMigrationException: The new key serializer
> (org.apache.flink.api.common.typeutils.base.IntSerializer@11a7ba62) must be
> compatible with the previous key serializer
> (org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer@3b56cc30).
> at
> org.apache.flink.runtime.state.heap.HeapRestoreOperation.restore(HeapRestoreOperation.java:147)
> at
> org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.build(HeapKeyedStateBackendBuilder.java:114)
> ... 15 more
> {code}
>
> State processor api should use the same serializer of Flink since the type is
> exactly the same.
> I have attached a zip that contains the code to test it.
> In the project zipped there are the source and rewritten savepoints.
>
> Note:
> I have tried to play with enableX/disableX serializer, but so far no success.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
