[jira] [Created] (FLINK-20890) flink-state-processor-api: differents serializers being taken from Flink & State processor api

[Shang Yuanchun (Jira)]

Felipe Silvestre Santos de Morais created FLINK-20890:
---------------------------------------------------------

             Summary: 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
             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)