[FLINK-22494][ha] Introduce PossibleInconsistentStateException

[XComp]

What is the purpose of the change

Updating the Checkpoint Metadata store of ZooKeeper and Kubernetes might lead into situations where the data was actually written but the request still failed (due to network issue for instance). In these cases, we end up in an inconsistent state where a reference exists in ZooKeeper or the k8s ConfigMap but the corresponding Checkpoint's state is discarded due to the failure. That might cause problems during recovery.

I rejected the idea of doing the cleanup entirely in the CheckpointCoordinator since we also use the StateHandleStore implementations in the DefaultJobGraphStore.

Brief change log

• Introduces PossibleInconsistentStateException
• Fabric8FlinkKubeClient translates network issues into PossibleInconsistentStateException
• StateHandleStore and its implementing classes handle and forward PossibleInconsistentStateException
• Extends TestingLongStateHandleHelper to work with reference instead of saving the state in the handle itself. The previous implementation caused problems when deserializing the StateHandles not referencing the actual data but providing a copy of the state.
• CheckpointCoordinator handles PossibleInconsistentStateException

Verifying this change

The following tests were extended:

• CheckpointCoordinatorFailureTest checks the failure handling for PossibleInconsistentStateException and another exception
• KubernetesStateHandleStoreTest and ZooKeeperStateHandleStoreTest were extended to cover the relevant exceptions
• Fabric8FlinkKubeClientTest was extended to cover both failure cases

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: yes
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not applicable

[flinkbot]

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit 4c7f59b (Sat Aug 28 12:09:35 UTC 2021)

Warnings:

• No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

• ❓ 1. The [description] looks good.
• ❓ 2. There is [consensus] that the contribution should go into to Flink.
• ❓ 3. Needs [attention] from.
• ❓ 4. The change fits into the overall [architecture].
• ❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
• @flinkbot approve all to approve all aspects
• @flinkbot approve-until architecture to approve everything until architecture
• @flinkbot attention @username1 [@username2 ..] to require somebody's attention
• @flinkbot disapprove architecture to remove an approval you gave earlier

[flinkbot]

CI report:

• 707261c UNKNOWN
• 44bb765 UNKNOWN
• 4c7f59b Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run travis re-run the last Travis build
• @flinkbot run azure re-run the last Azure build

[XComp]

I added this change as it makes it more handy when having the namespace also specified in the ConfigMap. See KubernetesClientTestBase.mockConfigMapRequest. It's not really necessary since NAMESPACE is also available in that class. But I ran into the issue of the missing namespace first before discovering that the namespace is set through a static variable.

[XComp]

The Kubernetes-related ITCases are currently not running. I created FLINK-20564 to cover this. I haven't had the chance to run the ITCase locally, yet. I checked that the test passes locally including my changes.

[XComp]

Build failures for 02f469d and 14a487 have been addressed in the following tickets:

• FLINK-22568 RescalingITCase.testSavepointRescalingInPartitionedOperatorStateList
• FLINK-22571 SavepointITCase.testSavepointForJobWithIteration
• FLINK-22312 YARNSessionFIFOITCase.checkForProhibitedLogContents
• FLINK-22567 OrcFsStreamingSinkITCase
• FLINK-22572 JsonFsStreamSinkITCase.testPart
• FLINK-22566 Running Kerberized YARN application on Docker test e2e test

[tillrohrmann]

Thanks for creating this PR @XComp. The changes look very good to me. I think you worked very thoroughly. Well done! I had a few minor comments. Please take a look. I think we can merge this PR once the comments are resolved.

[tillrohrmann]

Feel free to also open the PRs for the backports so that we get CI.

[XComp]

Thanks @tillrohrmann for reviewing the changes. I applied your comments and rebased the branch.

@rkhachatryan Could you give it a pass as well since I touched some code related to Checkpoints (namely refactoring the cleaner a bit). The commits that might be relevant to you are 01969be, 3b4a81f and 4b34da7

[rkhachatryan]

Thanks for the PR @XComp , it mostly LGTM.
I have one concern about the handling of NodeExistsException and some minor questions, PTAL.

[XComp]

Thanks @rkhachatryan for your comments. I addressed all of them either through the code or with a response. Please have a look.

[rkhachatryan]

Thanks for updating the PR @XComp, LGTM.

[XComp]

I finally rebase happened. I created the backports for release-1.12 and release-1.13

[XComp]

The reported build failure happened due to the unrelated issue FLINK-22595

[tillrohrmann]

Thanks for updating the PR @XComp. Merging it now.

[rmetzger]

I'm wondering if we need to also make (at least some instances) of the AlreadyExistException also an PossibleInconsistentStateException.

Consider this case:

Error while processing AcknowledgeCheckpoint message
org.apache.flink.runtime.checkpoint.CheckpointException: Could not complete the pending checkpoint 193393. Failure reason: Failure to finalize checkpoint.
	at org.apache.flink.runtime.checkpoint.CheckpointCoordinator.completePendingCheckpoint(CheckpointCoordinator.java:1255)
	at org.apache.flink.runtime.checkpoint.CheckpointCoordinator.receiveAcknowledgeMessage(CheckpointCoordinator.java:1100)
	at org.apache.flink.runtime.scheduler.ExecutionGraphHandler.lambda$acknowledgeCheckpoint$1(ExecutionGraphHandler.java:89)
	at org.apache.flink.runtime.scheduler.ExecutionGraphHandler.lambda$processCheckpointCoordinatorMessage$3(ExecutionGraphHandler.java:119)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
	at java.base/java.lang.Thread.run(Unknown Source)
Caused by: org.apache.flink.runtime.persistence.StateHandleStore$AlreadyExistException: checkpointID-0000000000000193393 already exists in ConfigMap cm-00000000000000000000000000000000-jobmanager-leader
	at org.apache.flink.kubernetes.highavailability.KubernetesStateHandleStore.getKeyAlreadyExistException(KubernetesStateHandleStore.java:534)
	at org.apache.flink.kubernetes.highavailability.KubernetesStateHandleStore.lambda$addAndLock$0(KubernetesStateHandleStore.java:155)
	at org.apache.flink.kubernetes.kubeclient.Fabric8FlinkKubeClient.lambda$attemptCheckAndUpdateConfigMap$11(Fabric8FlinkKubeClient.java:316)
	at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(Unknown Source)
	... 3 common frames omitted

The checkpoint coordinator will clean the completed checkpoint files, but there is an entry in the checkpoint config map ... leading to issues when restoring:

Caused by: org.apache.flink.util.FlinkException: Could not retrieve checkpoint 193393 from state handle under checkpointID-0000000000000193393. This indicates that the retrieved state handle is broken. Try cleaning the state handle store ... Caused by: java.io.FileNotFoundException: No such file or directory: s3://xxx/flink-ha/xxx/completedCheckpoint72e30229420c

(cc @XComp @wangyang0918 )

[wangyang0918]

I am curious why the key checkpointID-0000000000000193393 already exists in the ConfigMap before adding a new checkpoint. Even though taking fabric8 Kubernetes client failure and retrying into account, this is also not an expected behavior.

[XComp]

@wangyang0918 has a valid point. @rmetzger can you provide more logs? It might be worth creating a ticket for it instead of discussing it in this PR 🤔

[rmetzger]

(The discussion continues on the main PR comments)

[EnriqueL8]

Hey all, just found this amazing PR for fixing an issue that I have been hitting for quite a long time. I'm running Flink 1.13.2 with Kube HA enabled and filesystem storage.

I've seen this issue for the past 6 months and resolved it by cleaning up the ConfigMap since it effectively does get deleted in the filesystem but not in the CM as stated at the top of the issue?

Caused by: org.apache.flink.util.FlinkException: Could not retrieve checkpoint 9702 from state handle under checkpointID-0000000000000009702. This indicates that the retrieved state handle is broken. Try cleaning the state handle store.

I'm just looking for some clarification if this PR does fix this problem or the newly raised suggestion from @rmetzger fixes it. Thanks so much for your help!

[XComp]

@EnriqueL8 This PR made it into 1.13.1. With you using 1.13.2 I would say it's a new issue and might be related to what @rmetzger came up with. It's hard to come up with conclusions right away without any additional logs. May you create a ticket with the logs of a run where you experienced something like that and we could start looking into the issue.

[EnriqueL8]

This is the original ticket with logs https://issues.apache.org/jira/browse/FLINK-25098 and happy to provide a more up-to-date ticket with the new logs we are seeing but it's the same issue. Let me know if you need debug logs turned on or any other configuration that might help you diagnose!

[rmetzger]

Interesting that somebody else is running into this issue too. I think this issue indeed deserves a separate Jira ticket to properly investigate it. I'll file one now.

[rmetzger]

Here's the ticket, let's continue the discussion there: https://issues.apache.org/jira/browse/FLINK-28265

[Shashank-2712]

Hey all, an updated logs file have been uploaded for the issue @EnriqueL8 mentioned in the ticket https://issues.apache.org/jira/browse/FLINK-25098 . The ticket https://issues.apache.org/jira/browse/FLINK-28265 seems to cover a different issue. Let me know if any more details/logs are required to help you diagnose. Thank you.