[FLINK-7623] Detecting whether an operator is restored doesn't work with chained state

• Type: Bug
• Status: Closed
• Priority: Blocker
• Resolution: Fixed
• Affects Version/s: 1.3.2, 1.4.0
• Fix Version/s: 1.4.0
• Component/s: API / DataStream, Runtime / State Backends
• Assignee: Piotr Nowojski
• Reporter: Aljoscha Krettek
• Created: 14/Sep/17 12:43
• Updated: 24/Oct/17 14:28
• Resolved: 24/Oct/17 14:15

Description

Originally reported on the ML: https://lists.apache.org/thread.html/22a2cf83de3107aa81a03a921325a191c29df8aa8676798fcd497199@%3Cuser.flink.apache.org%3E

If we have a chain of operators where multiple of them have operator state, detection of the context.isRestored() flag (of CheckpointedFunction) does not work correctly. It's best exemplified using this minimal example where both the source and the flatMap have state:

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

env
		.addSource(new MaSource()).uid("source-1")
		.flatMap(new MaFlatMap()).uid("flatMap-1");

env.execute("testing");

If I do a savepoint with these UIDs, then change "source-1" to "source-2" and restore from the savepoint context.isRestored() still reports true for the source.

Issue Links

• is cloned by: FLINK-7901 Detecting whether an operator is restored doesn't work with chained state (Flink 1.3)
• links to: GitHub Pull Request #4851