{"record":{"id":"54065a7d309296e1","repo":"zhisheng17/flink-learning","slug":"failed-to-update-configmap-since-current-kubern","errorCode":null,"errorMessage":"Failed to update ConfigMap {} since current KubernetesCheckpointIDCounter does not have the leadership.","messagePattern":"Failed to update ConfigMap (.+?) since current KubernetesCheckpointIDCounter does not have the leadership\\.","errorType":"exception","errorClass":"KubernetesException","httpStatus":null,"severity":"error","filePath":"flink-learning-k8s/flink-k8s/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesCheckpointIDCounter.java","lineNumber":111,"sourceCode":"\tpublic long getAndIncrement() throws Exception {\n\t\tfinal AtomicLong current = new AtomicLong();\n\t\tfinal boolean updated = kubeClient.checkAndUpdateConfigMap(\n\t\t\tconfigMapName,\n\t\t\tconfigMap -> {\n\t\t\t\tif (KubernetesLeaderElector.hasLeadership(configMap, lockIdentity)) {\n\t\t\t\t\tfinal long currentValue = getCurrentCounter(configMap);\n\t\t\t\t\tcurrent.set(currentValue);\n\t\t\t\t\tconfigMap.getData().put(CHECKPOINT_COUNTER_KEY, String.valueOf(currentValue + 1));\n\t\t\t\t\treturn Optional.of(configMap);\n\t\t\t\t}\n\t\t\t\treturn Optional.empty();\n\t\t\t}\n\t\t).get();\n\n\t\tif (updated) {\n\t\t\treturn current.get();\n\t\t} else {\n\t\t\tthrow new KubernetesException(\"Failed to update ConfigMap \" + configMapName +\n\t\t\t\t\" since current KubernetesCheckpointIDCounter does not have the leadership.\");\n\t\t}\n\t}\n\n\t@Override\n\tpublic long get() {\n\t\treturn kubeClient.getConfigMap(configMapName)\n\t\t\t.map(this::getCurrentCounter)\n\t\t\t.orElseThrow(() -> new FlinkRuntimeException(\n\t\t\t\tnew KubernetesException(\"ConfigMap \" + configMapName + \" does not exist.\")));\n\t}\n\n\t@Override\n\tpublic void setCount(long newCount) throws Exception {\n\t\tkubeClient.checkAndUpdateConfigMap(\n\t\t\tconfigMapName,\n\t\t\tconfigMap -> {\n\t\t\t\tif (KubernetesLeaderElector.hasLeadership(configMap, lockIdentity)) {","sourceCodeStart":93,"sourceCodeEnd":129,"githubUrl":"https://github.com/zhisheng17/flink-learning/blob/d731cee7618021be56d132cc925102ffff8d75e6/flink-learning-k8s/flink-k8s/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesCheckpointIDCounter.java#L93-L129","documentation":"KubernetesCheckpointIDCounter.getAndIncrement performs a compare-and-swap update of the checkpoint counter ConfigMap. If the update reports 'not updated', the current instance has lost leadership, so the counter refuses to hand out a checkpoint ID and throws KubernetesException.","triggerScenarios":"Calling getAndIncrement (during checkpoint triggering) when this JobManager no longer holds the leader lease — its ConfigMap CAS update fails because another (new leader) instance modified the counter first.","commonSituations":"Leader change/JobManager failover racing with an in-flight checkpoint trigger; HA lease flapping due to Kubernetes API latency; two JobManagers briefly believing they are leader; ConfigMap update conflicts from concurrent resource-version changes.","solutions":["Verify current leadership before triggering checkpoints; after losing it, stop checkpoint initiation and let the new leader take over.","Check HA ConfigMap/lease state: kubectl get configmap,leases in the namespace; confirm only one JobManager holds the lease.","Reduce checkpoint trigger overlap with failovers (adjust checkpoint interval / postpone triggering during leader transition).","Investigate Kubernetes API latency or watch disconnections that cause lease renewal failures.","Restart the deposed JobManager so it re-acquires or steps down cleanly instead of operating with stale leadership."],"exampleFix":"// before: trigger checkpoints regardless of leadership\nCompletableFuture<CompletedCheckpoint> cc = checkpointCoordinator.triggerCheckpoint(false);\n// after: guard on leadership\nif (hasLeadership.get()) {\n    checkpointCoordinator.triggerCheckpoint(false);\n} // else skip; new leader will checkpoint","handlingStrategy":"try-catch","validationCode":"// only trigger checkpoints while leadership is held\nif (!leaderElector.hasLeadership()) return; // skip triggering","typeGuard":null,"tryCatchPattern":"try {\n    long id = checkpointIDCounter.getAndIncrement();\n} catch (KubernetesException e) {\n    if (e.getMessage().contains(\"does not have the leadership\")) {\n        // stop checkpointing; wait for/fetch new leader state\n    }\n}","preventionTips":["Gate checkpoint triggering on confirmed leadership","Keep HA lease renewal reliable (watch Kubernetes API latency)","Use reasonable checkpoint intervals to avoid failover races","Monitor ConfigMap resource-version conflicts in HA storage","Restart stale JobManagers cleanly after losing leadership"],"tags":["kubernetes","flink","high-availability","checkpoint-id-counter","leadership"],"backgroundTag":"invalid-state-transition","analyzedSha":"d731cee7618021be56d132cc925102ffff8d75e6","analyzedAt":"2026-09-06T05:35:08.496Z","contentChangedAt":"2026-09-06T05:35:08.496Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}