{"record":{"id":"a799f37c6ed7db97","repo":"apache/flink","slug":"s3clientprovider-has-been-closed","errorCode":null,"errorMessage":"S3ClientProvider has been closed","messagePattern":"S3ClientProvider has been closed","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3ClientProvider.java","lineNumber":396,"sourceCode":"                                } catch (Exception e) {\n                                    LOG.warn(\"Error closing STS client\", e);\n                                }\n                            }\n                        })\n                .orTimeout(clientCloseTimeout.toMillis(), TimeUnit.MILLISECONDS)\n                .exceptionally(\n                        ex -> {\n                            LOG.error(\n                                    \"S3 client close did not complete cleanly within {}\",\n                                    clientCloseTimeout,\n                                    ex);\n                            return null;\n                        });\n    }\n\n    private void checkNotClosed() {\n        if (closed.get()) {\n            throw new IllegalStateException(\"S3ClientProvider has been closed\");\n        }\n    }\n\n    public static Builder builder() {\n        return new Builder();\n    }\n\n    public static class Builder {\n        private String accessKey;\n        private String secretKey;\n        private String region;\n        private String endpoint;\n        // All defaults are sourced from NativeS3FileSystemFactory ConfigOption.defaultValue() so\n        // that NativeS3FileSystemFactory remains the single source of truth — if a default changes\n        // there, this Builder automatically picks it up without needing a parallel edit.\n        private boolean pathStyleAccess =\n                NativeS3FileSystemFactory.PATH_STYLE_ACCESS.defaultValue();\n        private boolean chunkedEncoding =","sourceCodeStart":378,"sourceCodeEnd":414,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3ClientProvider.java#L378-L414","documentation":"Thrown by S3ClientProvider.checkNotClosed() when any method on the provider is invoked after close() has already run. The provider wraps AWS SDK v2 clients (sync, async, TransferManager) and once closed it releases them asynchronously with a clientCloseTimeout; the AtomicBoolean 'closed' latches permanently and every subsequent operation is rejected. This is a lifecycle violation, not a transient failure.","triggerScenarios":"Calling any client-getter or operation on an S3ClientProvider instance after its close() method completed: e.g. a Flink FileSystem is closed (job teardown, reconfiguration, TM shutdown) while a slow writer/recoverable-mode operation still holds a reference and then calls s3Client.createMultipartUpload(...), putObject(...), or headObject(...). Also happens when user code caches the filesystem/provider across job restarts within the same JVM/classloader.","commonSituations":"Race between task shutdown and a concurrent upload thread; recoverable commit resumed after the filesystem was closed; integration tests that close the FS then reuse it; plugin classloader reuse where a stale provider is cached statically by user code.","solutions":["Audit for use-after-close races: make sure no upload/commit thread outlives the FileSystem close (join or cancel writer threads before closing the provider).","Do not cache S3ClientProvider or the FlinkS3FileSystem statically; acquire the filesystem via FileSystem.get(Utf8...) per job lifetime so Flink's own cache governs closing.","If you need a client after close, build a fresh S3ClientProvider via S3ClientProvider.builder() instead of reusing the closed instance.","In tests, use try-with-resources around the filesystem/provider so the close order is explicit and deterministic."],"exampleFix":"// before\nFileSystem fs = FileSystem.get(s3Path);\n// ... job finishes, Flink closes fs ...\ns3Writer.resumeUpload(...); // throws: S3ClientProvider has been closed\n\n// after\ntry (FlinkS3FileSystem fs = (FlinkS3FileSystem) FileSystem.get(s3Path)) {\n    s3Writer.resumeUpload(...); // guaranteed to run before close\n}","handlingStrategy":"validation","validationCode":"// Guard operations on a possibly-closed provider via try-catch on lifecycle\ntry {\n    s3Ops.putObject(key, file);\n} catch (IllegalStateException e) {\n    if (e.getMessage() != null && e.getMessage().contains(\"has been closed\")) {\n        // provider closed underneath us: re-acquire the filesystem/provider\n        provider = S3ClientProvider.builder()/* ...same config... */.build();\n        s3Ops = provider.s3ObjectOperations();\n    } else {\n        throw e;\n    }\n}","typeGuard":null,"tryCatchPattern":"catch (IllegalStateException e) { if (e.getMessage()!=null && e.getMessage().contains(\"has been closed\")) { /* re-acquire provider, do NOT retry on the closed instance */ } else throw e; }","preventionTips":["Never cache the S3 filesystem or provider in static fields; fetch per job and let Flink's cache own the lifecycle.","Join/cancel all upload threads before closing the filesystem in tests and custom sinks.","Use try-with-resources around the filesystem/provider so close ordering is explicit.","Treat 'has been closed' as a lifecycle bug in your code, not a retryable condition."],"tags":["lifecycle","s3","resource-management","concurrency","flink"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}