You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "tchivs (via GitHub)" <gi...@apache.org> on 2026/10/21 04:26:21 UTC
[I] [Bug] S3FileIO keeps reusing a cached S3A client after its HTTP connection pool shuts down [paimon]
tchivs opened a new issue, #10711:
URL: https://github.com/apache/paimon/issues/10711
### Search before asking
I searched for `Connection pool shut down` and S3 cache recovery. #5998 has the same symptom but describes a Flink/Paimon plugin conflict with older versions. This report reproduces the cache-recovery problem in a standalone JVM, without the Flink S3 plugin. #8548 concerns a different filesystem lifecycle/resource-leak issue.
### Paimon version
- Incident: a downstream Paimon 1.4.0-based build (not an unmodified ASF release).
- Local reproduction and fix verification: downstream fork, baseline `68a82e2f1`, project version `1.5-SNAPSHOT`, Hadoop S3A 3.4.2, AWS SDK v2 2.29.52.
- Upstream **source inspection**, not release-binary runtime verification: the same cache behavior is present in [release-2.1.0 S3FileIO](https://github.com/apache/paimon/blob/release-2.1.0/paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/S3FileIO.java) and [master at 8c09824c060520f3f1f4a03ce2369de371e5c7c6](https://github.com/apache/paimon/blob/8c09824c060520f3f1f4a03ce2369de371e5c7c6/paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/S3FileIO.java).
### Compute Engine
Incident: Flink 2.2.1. Deterministic local reproduction: standalone Java 8 JVM with real MinIO (`minio/minio:RELEASE.2025-04-22T22-12-26Z`), Paimon S3FileIO, and the default Hadoop S3 client factory. No Flink process or Flink S3 filesystem plugin is needed.
### Minimal reproduce step
1. Configure two `S3FileIO` instances with identical bucket/options. Write and read an object so both instances use the shared cached S3A filesystem.
2. Using that filesystem's SDK S3 client, issue a PUT whose request-body stream throws a deliberately injected `OutOfMemoryError`. This does **not** exhaust the JVM heap. Verify that the exception stack includes `org.apache.http.impl.execchain.MainClientExec`.
3. Call `first.exists(path)`. It throws `IllegalStateException: Connection pool shut down`.
4. Call `peer.getFileStatus(path)` or construct another `S3FileIO` with identical options. On the unpatched implementation, it reuses the poisoned cached client and fails again.
The input-stream injection isolates the failure mechanism from the original reason for heap exhaustion. Separately, a local regression reproduces the same persistence by shutting down only the Apache HTTP transport while keeping S3A and its credential provider alive.
<details>
<summary>Standalone executable reproduction / recovery smoke</summary>
Place the following class in package `org.apache.paimon.s3` so it can access the protected filesystem lookup. Compile it against the S3 implementation classes and its Maven dependency classpath. Pass a disposable local MinIO endpoint as the first argument. The credentials below are MinIO's default local test credentials, not production credentials. Unset HTTP/HTTPS proxy environment variables for this loopback smoke.
The program expects recovery; on the unpatched implementation the final peer lookup fails. With the local patch it prints all four PASS lines and verifies the original object bytes. Stop/remove the disposable MinIO container after the run.
```java
package org.apache.paimon.s3;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import org.apache.hadoop.fs.s3a.S3AFileSystem;
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.PositionOutputStream;
import org.apache.paimon.fs.SeekableInputStream;
import org.apache.paimon.options.Options;
import software.amazon.awssdk.core.sync.RequestBody;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.model.PutObjectRequest;
public class PaimonS3RecoverySmoke {
public static void main(String[] args) throws Exception {
Options options = new Options();
options.set("s3.endpoint", args[0]);
options.set("s3.endpoint.region", "us-east-1");
options.set("s3.path.style.access", "true");
options.set("s3.access.key", "minioadmin");
options.set("s3.secret.key", "minioadmin");
options.set("s3.bucket.probe", "0");
options.set("s3.connection.ssl.enabled", "false");
S3FileIO first = new S3FileIO();
S3FileIO peer = new S3FileIO();
first.configure(CatalogContext.create(options));
peer.configure(CatalogContext.create(options));
String bucket = "s3-recovery-" + System.nanoTime();
Path path = new Path("s3a://" + bucket + "/known-data");
org.apache.hadoop.fs.Path hp = new org.apache.hadoop.fs.Path(path.toUri());
S3AFileSystem failed = (S3AFileSystem) first.getFileSystem(hp);
S3Client sdk = failed.getS3AInternals().getAmazonS3Client("smoke");
S3AFileSystem recovered = null;
try {
sdk.createBucket(b -> b.bucket(bucket));
byte[] expected = "persisted-before-pool-failure".getBytes(StandardCharsets.UTF_8);
try (PositionOutputStream out = first.newOutputStream(path, false)) {
out.write(expected);
}
if (peer.getFileStatus(path).getLen() != expected.length) {
throw new AssertionError("Initial write not visible to peer");
}
System.out.println("PASS: default S3A factory writes and peer reads real MinIO object");
try {
sdk.putObject(PutObjectRequest.builder().bucket(bucket).key("injected-error").build(),
RequestBody.fromContentProvider(() -> new InputStream() {
public int read() { throw new OutOfMemoryError("injected smoke error; heap not exhausted"); }
public int read(byte[] b, int off, int len) { return read(); }
}, 1024, "application/octet-stream"));
throw new AssertionError("Injected Error did not reach caller");
} catch (OutOfMemoryError error) {
boolean inHttpClient = Arrays.stream(error.getStackTrace())
.anyMatch(f -> f.getClassName().equals("org.apache.http.impl.execchain.MainClientExec"));
if (!inHttpClient) {
error.printStackTrace();
throw new AssertionError("Error did not originate inside Apache request execution");
}
System.out.println("PASS: injected Error traverses MainClientExec and reaches caller");
}
try {
first.exists(path);
throw new AssertionError("Closed pool failure was silently replayed or not induced");
} catch (IllegalStateException error) {
if (!"Connection pool shut down".equals(error.getMessage())) throw error;
System.out.println("PASS: next operation reports exact Connection pool shut down");
}
if (peer.getFileStatus(path).getLen() != expected.length) {
throw new AssertionError("Peer did not recover");
}
recovered = (S3AFileSystem) first.getFileSystem(hp);
if (recovered == failed || recovered != peer.getFileSystem(hp)) {
throw new AssertionError("Caches disagree on replacement");
}
byte[] actual = new byte[expected.length];
try (SeekableInputStream in = first.newInputStream(path)) {
int pos = 0;
while (pos < actual.length) {
int n = in.read(actual, pos, actual.length - pos);
if (n < 0) throw new AssertionError("Truncated object");
pos += n;
}
}
if (!Arrays.equals(expected, actual)) throw new AssertionError("Stored data changed");
System.out.println("PASS: same JVM, same and peer FileIO recover; original bytes intact");
S3Client replacement = recovered.getS3AInternals().getAmazonS3Client("smoke cleanup");
replacement.deleteObject(b -> b.bucket(bucket).key("known-data"));
replacement.deleteBucket(b -> b.bucket(bucket));
} finally {
if (recovered != null) recovered.close();
failed.close();
}
}
}
```
</details>
### What doesn't meet your expectations?
A permanently unusable transport remains pinned in the shared S3 filesystem cache. Job/task restarts that reuse the plugin classloader can encounter the same poisoned client. In the reported incident, restarting the TaskManager process cleared the symptom.
This is **not** a request to hide the initial OOM or silently replay an interrupted write. The failing operation should still propagate. A later safe caller retry or task restart should be able to obtain a fresh client instead of being permanently tied to the dead pool.
### Anything else?
#### Evidence and mechanism
AWS documents that a `java.lang.Error`, including `OutOfMemoryError`, can [shut down the Apache HTTP connection pool](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/troubleshooting.html#faq-connection-pool-shutdown-exception). This need not call `S3AFileSystem.close()`.
Paimon's S3 implementation has two cache layers:
- `HadoopCompliantFileIO.fsMap`: per-FileIO filesystem references.
- `S3FileIO.CACHE`: static shared filesystem references, populated through `computeIfAbsent`.
Neither cache reacts to a permanently failed HTTP transport. Clearing only `fsMap` is insufficient because `createFileSystem` returns the same object from the static cache. A `FileSystem.isClosed()` guard is not a solution: it is not a Hadoop FileSystem API, and filesystem lifecycle state does not necessarily reflect an independently shut-down HTTP pool.
Representative failure chain:
```text
java.lang.IllegalStateException: Connection pool shut down
org.apache.http.impl.conn.PoolingHttpClientConnectionManager.requestConnection
software.amazon.awssdk.http.apache.internal.impl.ApacheSdkHttpClient.execute
software.amazon.awssdk.services.s3.DefaultS3Client.headObject
org.apache.hadoop.fs.s3a.impl.S3AStoreImpl.headObject
org.apache.hadoop.fs.s3a.S3AFileSystem.getFileStatus / exists
org.apache.paimon.s3.HadoopCompliantFileIO.getFileStatus / exists
```
In the incident, heap OOM preceded repeated pool failures. The exact pool shutdown call was not captured, so the incident's triggering cause remains a hypothesis. The standalone injected-Error smoke does establish the mechanism without any dependency conflict being necessary. This does not rule out separate packaging problems in other deployments.
#### Local repair and verification
The local patch appends an AWS `ExecutionInterceptor` through S3A's client-manager creation hook. When an execution failure contains the exact `IllegalStateException: Connection pool shut down`, it marks **that filesystem instance** unusable. On the next lookup, per-instance references refresh from an atomically replaced shared cache entry. The old filesystem is closed during replacement, not inside an SDK callback. Late failures from the old client cannot invalidate the replacement.
No automatic operation retries are added; lazy reads and multipart SDK requests use the same failure notification. An open stream or interrupted upload is not transparently resumed.
Verification completed locally:
- Regression failed before the patch at the second FileIO's lookup with the exact pool-shutdown exception.
- Five regression tests pass: shared cache recovery and stale-reference handling, lazy reads, multipart failure, access-denied non-invalidation, concurrent replacement reuse.
- Real MinIO/default-client-factory smoke: write/read succeeds; injected Error reaches MainClientExec; next operation reports pool shutdown; subsequent same/peer FileIO reads succeed in the same JVM with original bytes intact.
Not verified: the unmodified official 2.1.0 binary, the full Flink incident on a deployed patched TaskManager, or recovery of the original memory-pressure cause. Would the community consider this cache-recovery gap a bug and the above invalidation approach appropriate?
### Are you willing to submit a PR?
- [ ] I'm willing to submit a PR (not committed yet).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org