[jira] [Comment Edited] (FLINK-11087) Broadcast state migration Incompatibility from 1.5.3 to 1.7.0

["Tzu-Li (Gordon) Tai (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-11087?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16711511#comment-16711511 ] 

Tzu-Li (Gordon) Tai edited comment on FLINK-11087 at 12/6/18 2:35 PM:
----------------------------------------------------------------------

I'm currently modifying the {{StatefulJobWBroadcastStateMigrationITCase}} to verify this.
 Currently that ITCase had no way to capture this potential bug, because the key / value types of broadcast state tested there are all identical ..

If it turns out to be true, then we'll need to lift this as a blocker bug.


was (Author: tzulitai):
I'm currently modifying the {{StatefulJobWBroadcastStateMigrationITCase}} to verify this.
Currently that ITCase had now way to capture this bug, because the key / value types of broadcast state tested there are all identical ..

If it turns out to be true, then we'll need to lift this as a blocker bug.

> Broadcast state migration Incompatibility from 1.5.3 to 1.7.0
> -------------------------------------------------------------
>
>                 Key: FLINK-11087
>                 URL: https://issues.apache.org/jira/browse/FLINK-11087
>             Project: Flink
>          Issue Type: Bug
>          Components: State Backends, Checkpointing
>    Affects Versions: 1.7.0
>         Environment: Migration from Flink 1.5.3 to Flink 1.7.0
>            Reporter: Edward Rojas
>            Priority: Major
>              Labels: Migration, State, broadcast
>
> When upgrading from Flink 1.5.3 to Flink 1.7.0, the migration of broadcast state throws the following error:
> {noformat}
> org.apache.flink.util.StateMigrationException: The new key serializer for broadcast state must not be incompatible.
> at org.apache.flink.runtime.state.DefaultOperatorStateBackend.getBroadcastState(DefaultOperatorStateBackend.java:238)
> at org.apache.flink.streaming.api.operators.co.CoBroadcastWithNonKeyedOperator.open(CoBroadcastWithNonKeyedOperator.java:87)
> at org.apache.flink.streaming.runtime.tasks.StreamTask.openAllOperators(StreamTask.java:424)
> at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:290)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:704)
> at java.lang.Thread.run(Thread.java:745){noformat}
> The broadcast is using a MapState with StringSerializer as key serializer and a custom JsonSerializer as value serializer. 
> There was no changes in the TypeSerializers used, only upgrade of version. 
>  
> With some debugging I see that at the moment of the validation of the compatibility of states in the DefaultOperatorStateBackend class, the "*registeredBroadcastStates*" containing the data about the 'old' state, contains wrong association of the key and value serializer. This is, JsonSerializer appears as key serializer and StringSerializer appears as value serializer. (when it should be the contrary)
>  
> After more digging, I see that the "OperatorBackendStateMetaInfoReaderV2V3" class is the responsible of this swap here:
> https://github.com/apache/flink/blob/release-1.7/flink-runtime/src/main/java/org/apache/flink/runtime/state/metainfo/LegacyStateMetaInfoReaders.java#L165



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)