{"record":{"id":"3022f7b4031842d3","repo":"apache/seatunnel","slug":"failed-to-persist-readytoclosestartingtask-to-imap","errorCode":null,"errorMessage":"Failed to persist readyToCloseStartingTask to IMap, key: %s","messagePattern":"Failed to persist readyToCloseStartingTask to IMap, key: (.+?)","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"critical","filePath":"seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java","lineNumber":613,"sourceCode":"                                                    ? new HashSet<>((Set<TaskLocation>) exist)\n                                                    : new HashSet<>();\n                                    merged.addAll(readyToCloseStartingTask);\n                                    return merged;\n                                });\n                        return null;\n                    },\n                    new RetryUtils.RetryMaterial(\n                            Constant.OPERATION_RETRY_TIME,\n                            true,\n                            ExceptionUtil::isOperationNeedRetryException,\n                            Constant.OPERATION_RETRY_SLEEP));\n        } catch (Exception e) {\n            LOG.error(\n                    \"Failed to persist readyToCloseStartingTask to IMap after retries, key: {}.\"\n                            + \" Failing the job to avoid an unrecoverable stuck state on master failover.\",\n                    readyToCloseImapKey,\n                    e);\n            throw new RuntimeException(\n                    \"Failed to persist readyToCloseStartingTask to IMap, key: \"\n                            + readyToCloseImapKey,\n                    e);\n        }\n    }\n\n    protected void readyToCloseIdleTask(TaskLocation taskLocation) {\n        if (plan.getStartingSubtasks().contains(taskLocation)) {\n            throw new UnsupportedOperationException(\"Unsupported close starting task\");\n        }\n\n        LOG.info(\n                \"Received close idle task, task id: {}, pipeline id: {}, job id: {}, detail: {}\",\n                taskLocation.getTaskID(),\n                taskLocation.getPipelineId(),\n                taskLocation.getJobId(),\n                taskLocation);\n        synchronized (readyToCloseIdleTask) {","sourceCodeStart":595,"sourceCodeEnd":631,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java#L595-L631","documentation":"CheckpointCoordinator.updateReadyToCloseStartingTask persists the readyToCloseStartingTask marker to the IMap with retries; after exhausting retries it logs and throws a RuntimeException, deliberately failing the job so it doesn't get stuck in an unrecoverable state on master failover. Persistence of this marker is a required step of the checkpoint close protocol.","triggerScenarios":"All IMap put retries fail during readyToClose: sustained Hazelcast connectivity loss, repeated operation timeouts, partition owner unavailable during failover, or serialization errors on the value.","commonSituations":"Master node losing Hazelcast connection mid-checkpoint; long network partition between cluster members; Hazelcast cluster overloaded so operations time out beyond the retry budget.","solutions":["Check the wrapped cause and Hazelcast logs for timeouts/partition events during the failure window","Verify network stability between cluster nodes; fix the partition or connectivity problem before re-running","Increase Hazelcast operation timeout / retry budget if workloads legitimately cause long operation pauses","Resubmit the job from a compatible checkpoint state once the cluster is healthy — the job was intentionally failed to avoid a stuck state","Review IMap backup counts so the key survives single-node failures"],"exampleFix":"// before\n// no handling: job fails with raw RuntimeException\ncoordinator.readyToClose(taskLocation);\n\n// after\ntry {\n    coordinator.readyToClose(taskLocation);\n} catch (RuntimeException e) {\n    LOG.error(\"readyToClose persist failed; will resubmit job from last checkpoint\", e);\n    resubmitJobFromCheckpoint(jobId);\n}","handlingStrategy":"retry","validationCode":"// ensure cluster is healthy before protocol-critical writes\nboolean healthy = hazelcastInstance.getCluster().getMembers().size() == expectedMembers;","typeGuard":null,"tryCatchPattern":"try {\n    coordinator.readyToClose(taskLocation);\n} catch (RuntimeException e) {\n    LOG.error(\"IMap persist failed for {}\", readyToCloseImapKey, e.getCause());\n    resubmitJobFromLastCheckpoint(jobId);\n}","preventionTips":["Ensure stable networking between master and workers","Configure IMap backups so keys survive node loss","Treat the thrown exception as intentional job-fail; recover by checkpoint-based resubmission"],"tags":["zeta-engine","checkpoint","imap","hazelcast","failover"],"backgroundTag":"database-write-failed","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}