{"record":{"id":"5b03487f3d2c742f","repo":"apache/flink","slug":"failed-to-start-multipart-upload-for-key","errorCode":null,"errorMessage":"Failed to start multipart upload for key: {}","messagePattern":"Failed to start multipart upload for key: (.+?)","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3ObjectOperations.java","lineNumber":147,"sourceCode":"        this.s3Client = s3Client;\n        this.transferManager = transferManager;\n        this.bucketName = bucketName;\n        this.useAsyncOperations = useAsyncOperations && transferManager != null;\n        this.encryptionConfig =\n                encryptionConfig != null ? encryptionConfig : S3EncryptionConfig.none();\n    }\n\n    public String startMultiPartUpload(String key) throws IOException {\n        try {\n            CreateMultipartUploadRequest.Builder requestBuilder =\n                    CreateMultipartUploadRequest.builder().bucket(bucketName).key(key);\n            applyEncryption(requestBuilder);\n\n            CreateMultipartUploadResponse response =\n                    s3Client.createMultipartUpload(requestBuilder.build());\n            return response.uploadId();\n        } catch (S3Exception e) {\n            throw new IOException(\"Failed to start multipart upload for key: \" + key, e);\n        }\n    }\n\n    private void applyEncryption(CreateMultipartUploadRequest.Builder requestBuilder) {\n        if (!encryptionConfig.isEnabled()) {\n            return;\n        }\n        requestBuilder.serverSideEncryption(encryptionConfig.getServerSideEncryption());\n        if (encryptionConfig.getEncryptionType() == S3EncryptionConfig.EncryptionType.SSE_KMS) {\n            if (encryptionConfig.getKmsKeyId() != null) {\n                requestBuilder.ssekmsKeyId(encryptionConfig.getKmsKeyId());\n            }\n            if (encryptionConfig.hasEncryptionContext()) {\n                requestBuilder.ssekmsEncryptionContext(\n                        encryptionConfig.serializeEncryptionContext());\n            }\n        }\n    }","sourceCodeStart":129,"sourceCodeEnd":165,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3ObjectOperations.java#L129-L165","documentation":"startMultiPartUpload() calls S3Client.createMultipartUpload (CreateMultipartUploadRequest with bucket, key, and server-side encryption applied per S3EncryptionConfig). Any S3Exception from the service — access denied, no such bucket, KMS key errors, invalid endpoint — is wrapped in this IOException with the object key. The uploadId this call returns is the handle for all subsequent part uploads, so nothing downstream can proceed until it succeeds.","triggerScenarios":"IAM role missing s3:PutObject / s3:CreateMultipartUpload on the bucket or prefix; bucket in another account without cross-account policy; SSE-KMS configured but the role lacks kms:GenerateDataKey on the CMK (CreateMultipartUpload is where SSE headers are validated); wrong bucket name or endpoint override; bucket-owner-enforced-object-writes with wrong principal.","commonSituations":"First write to a new bucket with restrictive policy; enabling fs.s3.encryption without granting the Flink role KMS permissions; typos in fs.s3.bucket or endpoint; VPC endpoint policies blocking multipart creation; expired temporary credentials.","solutions":["Read the nested S3Exception status/code: 403 AccessDenied -> fix IAM (s3:PutObject, s3:CreateMultipartUpload, and with SSE-KMS kms:GenerateDataKey+ kms:Decrypt on the key); 404 NoSuchBucket -> fix bucket name/endpoint.","Verify with aws s3api create-multipart-upload --bucket B --key test using the same credentials/role the Flink job uses.","If using SSE-KMS, confirm fs.s3.kms-key.id points to a key the role can use and that the key is enabled in the region.","Check endpoint/region/path-style settings match the bucket (MinIO: fs.s3.path.style.access: true + endpoint + s3.region)."],"exampleFix":"# before: policy grants only s3:PutObject\n{ \"Effect\": \"Allow\", \"Action\": [\"s3:PutObject\"], \"Resource\": \"arn:aws:s3:::my-bucket/*\" }\n\n# after: include multipart actions\n{ \"Effect\": \"Allow\",\n  \"Action\": [\"s3:PutObject\", \"s3:AbortMultipartUpload\", \"s3:ListMultipartUploadParts\", \"s3:ListBucketMultipartUploads\"],\n  \"Resource\": [\"arn:aws:s3:::my-bucket/*\", \"arn:aws:s3:::my-bucket\"] }","handlingStrategy":"retry","validationCode":"// Pre-flight permission probe with the job's credentials\ntry (S3Client c = S3Client.create()) {\n    c.createMultipartUpload(b -> b.bucket(bucket).key(\"__flink_probe__\"));\n    // abort immediately; failure here reproduces the IAM problem before a job runs\n}","typeGuard":null,"tryCatchPattern":"catch (IOException e) { S3Exception s3 = (S3Exception) e.getCause(); switch (s3.statusCode()) { case 403: fail IAM audit; case 404: fix bucket/endpoint; case 500/503: retry with backoff; default: rethrow; } }","preventionTips":["Grant s3:PutObject, s3:AbortMultipartUpload, s3:ListMultipartUploadParts, s3:ListBucketMultipartUploads to the job role up front.","With SSE-KMS also grant kms:GenerateDataKey/kms:Decrypt on the configured key.","Run a one-object write smoke test against new buckets/policies before launching jobs."],"tags":["s3","multipart-upload","iam","permissions","network","flink"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}