{"record":{"id":"6393870e14db91af","repo":"apache/flink","slug":"blob-s-already-exists-during-attempted-commit","errorCode":null,"errorMessage":"Blob %s already exists during attempted commit","messagePattern":"Blob (.+?) already exists during attempted commit","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"flink-filesystems/flink-gs-fs-hadoop/src/main/java/org/apache/flink/fs/gs/writer/GSRecoverableWriterCommitter.java","lineNumber":89,"sourceCode":"        Preconditions.checkArgument(composeMaxBlobs > 0);\n        this.composeMaxBlobs = composeMaxBlobs;\n    }\n\n    GSRecoverableWriterCommitter(\n            GSBlobStorage storage, GSFileSystemOptions options, GSCommitRecoverable recoverable) {\n        this(storage, options, recoverable, BlobUtils.COMPOSE_MAX_BLOBS);\n    }\n\n    @Override\n    public void commit() throws IOException {\n        LOGGER.trace(\"Committing recoverable with options {}: {}\", options, recoverable);\n\n        // see discussion: https://github.com/apache/flink/pull/15599#discussion_r623127365\n        // first, make sure the final blob doesn't already exist\n        Optional<GSBlobStorage.BlobMetadata> blobMetadata =\n                storage.getMetadata(recoverable.finalBlobIdentifier);\n        if (blobMetadata.isPresent()) {\n            throw new IOException(\n                    String.format(\n                            \"Blob %s already exists during attempted commit\",\n                            recoverable.finalBlobIdentifier));\n        }\n\n        // write the final blob\n        writeFinalBlob();\n\n        // clean up after successful commit\n        cleanupTemporaryBlobs();\n    }\n\n    @Override\n    public void commitAfterRecovery() throws IOException {\n        LOGGER.trace(\n                \"Committing recoverable after recovery with options {}: {}\", options, recoverable);\n\n        // see discussion: https://github.com/apache/flink/pull/15599#discussion_r623127365","sourceCodeStart":71,"sourceCodeEnd":107,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-filesystems/flink-gs-fs-hadoop/src/main/java/org/apache/flink/fs/gs/writer/GSRecoverableWriterCommitter.java#L71-L107","documentation":"GSRecoverableWriterCommitter.commit first checks that the final blob does not already exist, and throws this IOException if it does. The GCS committer implements create-new semantics (see the linked PR discussion): a commit must atomically create the target blob, so an existing blob means the commit is a duplicate or conflicting with existing data rather than an idempotent repeat.","triggerScenarios":"Calling commit() twice for the same recoverable (e.g. job manager and a recovered task both committing); two jobs/parts writing to the same final blob path; a previous commit succeeded and cleanup was interrupted, then commit is retried; recovery logic that re-commits instead of using commitAfterRecovery-style handling.","commonSituations":"Task failover replaying the commit step without recovery awareness; misconfigured sink output paths causing multiple writers to target one file; retrying a job from a checkpoint whose pending file was already fully committed and renamed.","solutions":["Make each writer/commit target a unique final blob path (e.g. part files with unique part indexes) so concurrent or retried commits cannot collide","On recovery, detect that the blob already exists and skip re-committing instead of calling commit() again (treat the earlier commit as done, then clean up temp blobs)","Inspect the existing blob: if it was produced by the same in-flight attempt, remove leftover temporary component blobs and continue without re-committing"],"exampleFix":"// before\ncommitter.commit();\n// after: guard on recovery\ngoogStorage.delete(finalBlobIdentifier) only if safe, or skip commit when metadata already present\nif (storage.getMetadata(recoverable.finalBlobIdentifier).isEmpty()) {\n    committer.commit();\n}","handlingStrategy":"validation","validationCode":"// before committing, check for an existing final blob (idempotent skip)\nif (storage.getMetadata(recoverable.finalBlobIdentifier).isPresent()) {\n    // previous commit already landed; skip commit, optionally clean temp blobs\n    cleanupTemporaryBlobs();\n    return;\n}\ncommitter.commit();","typeGuard":null,"tryCatchPattern":"try {\n    committer.commit();\n} catch (IOException e) {\n    if (e.getMessage().contains(\"already exists\")) {\n        // decide: identical prior commit -> treat as success + cleanup; conflict -> fail loudly\n    } else {\n        throw e;\n    }\n}","preventionTips":["Give every writer unique final part-file names so commits cannot collide","Make recovery logic aware that commit may have already succeeded","Always run cleanup of temporary blobs after successful commits"],"tags":["gcs","commit","idempotency","recovery"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}