apache/flink · error · IllegalStateException

FileSystem has been closed for bucket: {}. Operations are no

Error message

FileSystem has been closed for bucket: {}. Operations are no longer permitted.

What it means

Every mutating operation on NativeS3FileSystem starts with checkNotClosed(), which throws IllegalStateException once the closed AtomicBoolean is set by closeAsync(). S3 filesystem instances are cached and shut down on configuration changes or JVM/TaskManager shutdown, and using a stale reference afterwards is a programming error.

Source

Thrown at flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystem.java:593

                                        LOG.error(
                                                "FileSystem close did not complete cleanly within {} for bucket: {}",
                                                fsCloseTimeout,
                                                bucketName,
                                                error);
                                    }
                                });
        FutureUtils.assertNoException(closeFuture);
        return closeFuture;
    }

    /**
     * Verifies that the filesystem has not been closed.
     *
     * @throws IllegalStateException if the filesystem has been closed
     */
    private void checkNotClosed() {
        if (closed.get()) {
            throw new IllegalStateException(
                    "FileSystem has been closed for bucket: "
                            + bucketName
                            + ". Operations are no longer permitted.");
        }
    }
}

View on GitHub (pinned to 2f3c205e92)

Solutions

Do not cache FileSystem instances; always re-acquire via FileSystem.get(uri) (the UDF/sink should fetch per use or rely on Flink's cache) so you get a live instance.
Ensure no writes happen after closing output streams / after job termination; sequence your shutdown after all tasks finish.
If triggered by a config change mid-run, restart the affected jobs so they pick up fresh, open filesystem instances.

Example fix

// before
private static final FileSystem FS = FileSystem.get("s3://bucket");
// ... later, after cache invalidation:
FS.create(path, WriteMode.OVERWRITE); // IllegalStateException

// after
// re-acquire each time; FileSystem.get returns the cached (or new) open instance
FileSystem fs = FileSystem.get(new Path("s3://bucket/").toUri());
fs.create(path, WriteMode.OVERWRITE);

Defensive patterns

Strategy: try-catch

Validate before calling

// always re-acquire instead of caching
FileSystem fs = FileSystem.get(new Path("s3://my-bucket/").toUri());
fs.create(path, WriteMode.OVERWRITE); // cache hands out a live instance

Try / catch

try {
    fs.create(path, WriteMode.OVERWRITE);
} catch (IllegalStateException e) {
    if (e.getMessage() != null && e.getMessage().contains("closed")) {
        fs = FileSystem.get(new Path("s3://my-bucket/").toUri()); // re-acquire live instance
        fs.create(path, WriteMode.OVERWRITE);
    } else {
        throw e;
    }
}

Prevention

Never store FileSystem instances in static fields; use FileSystem.get each time and let Flink's cache manage lifecycle.
Finish all writes before closing output / job teardown to avoid shutdown races.
After changing fs.s3.* config, expect cached instances to be closed — restart affected jobs.

When it happens

Trigger: Holding a NativeS3FileSystem reference across a closeAsync() (e.g. after FileSystem cache invalidation due to config change, or TaskManager shutdown) and then calling open/create/delete/rename/copyFiles on it.

Common situations: Caching FileSystem instances in user code beyond job lifecycle; reconfiguring fs.s3.* options which invalidates and closes cached instances; tests that close filesystems but keep references; races during shutdown where a task still writes output.

Related errors

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/f77b01cca46c32ea. Report an issue: GitHub.