[FLINK-10333] Rethink ZooKeeper based stores (SubmittedJobGraph, MesosWorker, CompletedCheckpoints)

• Type: Bug
• Status: Open
• Priority: Not a Priority
• Resolution: Unresolved
• Affects Version/s: 1.5.3, 1.6.0, 1.7.0, 1.11.3, 1.12.0
• Component/s: Runtime / Coordination
• Labels: auto-deprioritized-major auto-deprioritized-minor
• Assignee: Unassigned
• Reporter: Till Rohrmann
• Created: 13/Sep/18 08:49
• Updated: 30/Mar/22 22:38

Description

While going over the ZooKeeper based stores (ZooKeeperSubmittedJobGraphStore, ZooKeeperMesosWorkerStore, ZooKeeperCompletedCheckpointStore) and the underlying ZooKeeperStateHandleStore I noticed several inconsistencies which were introduced with past incremental changes.

• Depending whether ZooKeeperStateHandleStore#getAllSortedByNameAndLock or ZooKeeperStateHandleStore#getAllAndLock is called, deserialization problems will either lead to removing the Znode or not
• ZooKeeperStateHandleStore leaves inconsistent state in case of exceptions (e.g. #getAllAndLock won't release the acquired locks in case of a failure)
• ZooKeeperStateHandleStore has too many responsibilities. It would be better to move RetrievableStateStorageHelper out of it for a better separation of concerns
• ZooKeeperSubmittedJobGraphStore overwrites a stored JobGraph even if it is locked. This should not happen since it could leave another system in an inconsistent state (imagine a changed JobGraph which restores from an old checkpoint)
• Redundant but also somewhat inconsistent put logic in the different stores
• Shadowing of ZooKeeper specific exceptions in ZooKeeperStateHandleStore which were expected to be caught in ZooKeeperSubmittedJobGraphStore
• Getting rid of the SubmittedJobGraphListener would be helpful

These problems made me think how reliable these components actually work. Since these components are very important, I propose to refactor them.

Issue Links

• relates to: FLINK-10011 Old job resurrected during HA failover
• relates to: FLINK-10694 ZooKeeperHaServices Cleanup
• relates to: FLINK-21979 Job can be restarted from the beginning after it reached a terminal state
• relates to: FLINK-11225 Error state of addedJobGraphs when Dispatcher with concurrent revoking and granting leadership
• relates to: FLINK-4233 Simplify leader election / leader session ID assignment
• links to: Design Doc

Sub-Tasks

• Replace ZooKeeperStateHandleStore#getAllSortedByNameAndLock by getAllAndLock: Closed
• Move RetrievableStateStorageHelper out of ZooKeeperStateHandleStore: Open
• Create common ZooKeeperStateStore based on ZooKeeperStateHandleStore and RetrievableStateStorageHelper: Open
• Use ZooKeeperStateStore in ZooKeeperSubmittedJobGraphStore: Open
• Use ZooKeeperStateStore in ZooKeeperCompletedCheckpointStore: Open
• Rethink SubmittedJobGraphListener: Open
• Use ZooKeeperStateStore in ZooKeeperMesosWorkerStore: Closed
• Introduce ZooKeeperLeaderElectionServiceNG: Closed