You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Samrat002 (via GitHub)" <gi...@apache.org> on 2026/04/19 15:51:02 UTC
[PR] [FLINK-39482][filesystem] Support configurable maxConnections in S3ClientProvider [flink]
Samrat002 opened a new pull request, #27970:
URL: https://github.com/apache/flink/pull/27970
## What is the purpose of the change
This pull request prevents S3 connection pool exhaustion during RocksDB state restore when using the Native S3 filesystem. When NativeS3BulkCopyHelper fires concurrent downloads
via S3TransferManager, each multipart download can consume multiple HTTP connections. With the default pool size of 50 and a batch concurrency of 16, the pool can be exhausted,
causing downloads to hang until the SDK's 10-second acquire timeout expires — resulting in opaque SdkClientException failures during checkpoint restore.
The fix introduces a configurable s3.connection.max option, clamps s3.bulk-copy.max-concurrent to the connection pool size, raises the connection acquisition timeout from the SDK
default of 10s to the user-configured connection timeout
## Brief change log
*(for example:)*
- *The TaskInfo is stored in the blob store on job creation time as a persistent artifact*
- *Deployments RPC transmits only the blob storage reference*
- *TaskManagers retrieve the TaskInfo from the blob cache*
## Verifying this change
- Added testConnectionPoolExhaustedDetection in NativeS3BulkCopyHelperTest to verify detection of the SDK's connection acquire timeout message in both direct and nested causal
chains, as well as false-positive resistance
- Added testEmptyRequestListIsNoOp in NativeS3BulkCopyHelperTest to verify no NPE when no files are requested
- Added testInvalidMaxConnectionsThrowsException and testInvalidBulkCopyMaxConcurrentThrowsException in NativeS3FileSystemFactoryTest to verify config validation through the
factory
- Added testBulkCopyMaxConcurrentClampedToMaxConnections and testBulkCopyMaxConcurrentPreservedWithinMaxConnections in NativeS3FileSystemFactoryTest to verify the clamping logic
end-to-end through factory → filesystem → bulk copy helper
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with @Public(Evolving): no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: yes (affects state restore from S3 checkpoints)
- The S3 file system connector: yes
## Documentation
- Does this pull request introduce a new feature? yes
- If yes, how is the feature documented? JavaDocs
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39482][filesystem] Support configurable maxConnections in S3ClientProvider [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #27970:
URL: https://github.com/apache/flink/pull/27970#discussion_r3109225035
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3BulkCopyHelper.java:
##########
@@ -113,12 +152,18 @@ public void copyFiles(
}
if (copyFutures.size() >= maxConcurrentCopies || i == requests.size() - 1) {
+ batchNumber++;
+ LOG.debug(
+ "Waiting for batch {}/{} ({} files)",
+ batchNumber,
+ totalBatches,
+ copyFutures.size());
waitForCopies(copyFutures);
copyFutures.clear();
}
}
- LOG.info("Completed bulk copy of {} files", requests.size());
+ LOG.info("Completed bulk copy of {} files", totalFiles);
Review Comment:
I am curious - the total number of files could be different to the total number of requests. does that mean a request could be for more than one file . nit: maybe put out the number of requests made as well in trace - as least as debug - so the log would show what is happening around this.
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39482][filesystem] Support configurable maxConnections in S3ClientProvider [flink]
Posted by "gaborgsomogyi (via GitHub)" <gi...@apache.org>.
gaborgsomogyi commented on PR #27970:
URL: https://github.com/apache/flink/pull/27970#issuecomment-4286201676
Intended to merge this unless comments arrive
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39482][filesystem] Support configurable maxConnections in S3ClientProvider [flink]
Posted by "Samrat002 (via GitHub)" <gi...@apache.org>.
Samrat002 commented on code in PR #27970:
URL: https://github.com/apache/flink/pull/27970#discussion_r3110658596
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactoryTest.java:
##########
@@ -353,6 +353,87 @@ void testEmptyRegionFallsBackToAutodiscovery() throws Exception {
}
}
+ @Test
+ void testInvalidMaxConnectionsThrowsException() {
+ NativeS3FileSystemFactory factory = new NativeS3FileSystemFactory();
+ Configuration config = new Configuration();
+ config.setString("s3.access-key", "test-access-key");
+ config.setString("s3.secret-key", "test-secret-key");
+ config.setString("s3.region", "us-east-1");
+ config.set(NativeS3FileSystemFactory.MAX_CONNECTIONS, 0);
+ config.setString("io.tmp.dirs", System.getProperty("java.io.tmpdir"));
+
+ factory.configure(config);
+
+ URI fsUri = URI.create("s3://test-bucket/");
+ assertThatThrownBy(() -> factory.create(fsUri))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ void testInvalidBulkCopyMaxConcurrentThrowsException() {
+ NativeS3FileSystemFactory factory = new NativeS3FileSystemFactory();
+ Configuration config = new Configuration();
+ config.setString("s3.access-key", "test-access-key");
+ config.setString("s3.secret-key", "test-secret-key");
+ config.setString("s3.region", "us-east-1");
+ config.set(NativeS3FileSystemFactory.BULK_COPY_MAX_CONCURRENT, 0);
+ config.setString("io.tmp.dirs", System.getProperty("java.io.tmpdir"));
+
+ factory.configure(config);
+
+ URI fsUri = URI.create("s3://test-bucket/");
+ assertThatThrownBy(() -> factory.create(fsUri))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("maxConcurrentCopies must be positive");
+ }
+
+ @Test
+ void testBulkCopyMaxConcurrentClampedToMaxConnections() throws Exception {
+ NativeS3FileSystemFactory factory = new NativeS3FileSystemFactory();
+ Configuration config = new Configuration();
+ config.setString("s3.access-key", "test-access-key");
+ config.setString("s3.secret-key", "test-secret-key");
+ config.setString("s3.region", "us-east-1");
+ config.set(NativeS3FileSystemFactory.BULK_COPY_ENABLED, true);
+ config.set(NativeS3FileSystemFactory.BULK_COPY_MAX_CONCURRENT, 32);
+ config.set(NativeS3FileSystemFactory.MAX_CONNECTIONS, 10);
+ config.setString("io.tmp.dirs", System.getProperty("java.io.tmpdir"));
+
+ factory.configure(config);
+
+ URI fsUri = URI.create("s3://test-bucket/");
+ FileSystem fs = factory.create(fsUri);
+
+ assertThat(fs).isInstanceOf(NativeS3FileSystem.class);
+ NativeS3FileSystem nativeFs = (NativeS3FileSystem) fs;
+ assertThat(nativeFs.getBulkCopyHelper()).isNotNull();
+ assertThat(nativeFs.getBulkCopyHelper().getEffectiveMaxConcurrentCopies()).isEqualTo(10);
Review Comment:
This follows the existing pattern in the same file https://github.com/apache/flink/blob/FLINK-39482/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java#L346-L358 clamps
READ_BUFFER_SIZE to [64KB, 4MB] with LOG.warn, and the comment explicitly says "We clip rather than throw to provide flexibility while preventing extreme values". The maxConcurrentCopies clamping follows the same philosophy for consistency.
The LOG.warn emits a message naming both config keys and the clamped value:
`s3.bulk-copy.max-concurrent` (32) exceeds `s3.connection.max` (10). Clamping concurrent copies to 10 to prevent connection pool exhaustion.
Throwing would have a practical downside. a user who sets `s3.bulk-copy.max-concurrent=32` has a valid configuration today (default s3.connection.max=50). If they later lower maxConnections for an unrelated reason (e.g., reducing memory footprint), their previously working job would fail to start. for a misconfiguration that has a safe, deterministic automatic resolution.
I understand the concern about user intent being silently overridden. If you feel strongly about fail-fast here, I'm open to changing it, but we should then also change the READ_BUFFER_SIZE clamping for consistency, which would be a separate patch. WDYT?
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39482][filesystem] Support configurable maxConnections in S3ClientProvider [flink]
Posted by "Samrat002 (via GitHub)" <gi...@apache.org>.
Samrat002 commented on code in PR #27970:
URL: https://github.com/apache/flink/pull/27970#discussion_r3110704838
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3BulkCopyHelper.java:
##########
@@ -181,8 +226,42 @@ private void waitForCopies(List<CompletableFuture<CompletedCopy>> futures) throw
Thread.currentThread().interrupt();
throw new IOException("Bulk copy interrupted", e);
} catch (ExecutionException e) {
- throw new IOException("Bulk copy failed", e.getCause());
+ Throwable cause = e.getCause();
+ if (isConnectionPoolExhausted(cause)) {
+ throw new IOException(
+ String.format(
+ "S3 connection pool exhausted during bulk copy. "
+ + "The configured connection pool size (%d) could not serve "
+ + "the concurrent download requests (%d). "
+ + "Consider reducing '%s' or increasing '%s'.",
+ maxConnections,
+ maxConcurrentCopies,
+ "s3.bulk-copy.max-concurrent",
+ "s3.connection.max"),
Review Comment:
added the change. PTAL
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39482][filesystem] Support configurable maxConnections in S3ClientProvider [flink]
Posted by "Samrat002 (via GitHub)" <gi...@apache.org>.
Samrat002 commented on code in PR #27970:
URL: https://github.com/apache/flink/pull/27970#discussion_r3110578098
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3BulkCopyHelper.java:
##########
@@ -70,10 +74,37 @@ class NativeS3BulkCopyHelper {
private final S3TransferManager transferManager;
private final int maxConcurrentCopies;
+ private final int maxConnections;
- public NativeS3BulkCopyHelper(S3TransferManager transferManager, int maxConcurrentCopies) {
+ /**
+ * Creates a new bulk copy helper.
+ *
+ * @param transferManager the S3 transfer manager for async downloads
+ * @param maxConcurrentCopies the requested maximum number of concurrent copy operations
+ * @param maxConnections the HTTP connection pool size; if {@code maxConcurrentCopies} exceeds
+ * this value, it is clamped down to prevent connection pool exhaustion
+ */
+ NativeS3BulkCopyHelper(
+ S3TransferManager transferManager, int maxConcurrentCopies, int maxConnections) {
+ checkArgument(maxConcurrentCopies > 0, "maxConcurrentCopies must be positive");
+ checkArgument(maxConnections > 0, "maxConnections must be positive");
this.transferManager = transferManager;
- this.maxConcurrentCopies = maxConcurrentCopies;
+ this.maxConnections = maxConnections;
+ if (maxConcurrentCopies > maxConnections) {
+ LOG.warn(
+ "s3.bulk-copy.max-concurrent ({}) exceeds s3.connection.max ({}). "
+ + "Clamping concurrent copies to {} to prevent connection pool exhaustion.",
Review Comment:
Thanks for the review. @davidradl
> I suggest saying limiting rather thanclamping as it is clearer.
"clamping" is the standard CS term for bounding a value to a range ( `Math.clamp()` in Java 21), and this codebase already uses it in the same file at
https://github.com/apache/flink/blob/b0ea378c64d/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java#L338:
```
// Validate
and clamp read buffer size to sensible range".
```
I'd prefer to keep the term consistent with the existing codebase usage.
> I suggest calling the maxConcurrentCopies requestedMaxConcurrentCopies and maybe you could change the getter to getMaxConcurrentCopies. I do not think the caller needs to be aware that this is effective.
the @param Javadoc already documents it as "the requested maximum", so keeping the concise maxConcurrentCopies avoids verbosity in the internal API. However, agreed on the getter rename is much better. I have updated to `getMaxConcurrentCopies()`.
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39482][filesystem] Support configurable maxConnections in S3ClientProvider [flink]
Posted by "Samrat002 (via GitHub)" <gi...@apache.org>.
Samrat002 commented on code in PR #27970:
URL: https://github.com/apache/flink/pull/27970#discussion_r3110726922
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3BulkCopyHelper.java:
##########
@@ -181,8 +226,42 @@ private void waitForCopies(List<CompletableFuture<CompletedCopy>> futures) throw
Thread.currentThread().interrupt();
throw new IOException("Bulk copy interrupted", e);
} catch (ExecutionException e) {
- throw new IOException("Bulk copy failed", e.getCause());
+ Throwable cause = e.getCause();
+ if (isConnectionPoolExhausted(cause)) {
Review Comment:
Yes. The clamping ensures `maxConcurrentCopies <= maxConnections`, but S3TransferManager uses multipart downloads for files larger than 8MB. Each part opens a separate HTTP connection from the shared pool. So with `maxConcurrentCopies=10` and `maxConnections=10`, downloading 10 large files simultaneously, where each splits into 4 parts = 40 connections needed from a pool of 10.
The clamping eliminates the obvious case (concurrency alone exceeding the pool). This detection handles the remaining case where the per-file connection multiplier, which depends on runtime file sizes and is not known at configuration time, causes exhaustion. Without it, the operator gets a raw SDK timeout `Acquire operation took longer than the configured maximum time` with no indication of what to do. With it, they get an actionable message pointing to both config keys.
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39482][filesystem] Support configurable maxConnections in S3ClientProvider [flink]
Posted by "Samrat002 (via GitHub)" <gi...@apache.org>.
Samrat002 commented on code in PR #27970:
URL: https://github.com/apache/flink/pull/27970#discussion_r3110630845
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactoryTest.java:
##########
@@ -353,6 +353,87 @@ void testEmptyRegionFallsBackToAutodiscovery() throws Exception {
}
}
+ @Test
+ void testInvalidMaxConnectionsThrowsException() {
+ NativeS3FileSystemFactory factory = new NativeS3FileSystemFactory();
+ Configuration config = new Configuration();
+ config.setString("s3.access-key", "test-access-key");
+ config.setString("s3.secret-key", "test-secret-key");
+ config.setString("s3.region", "us-east-1");
+ config.set(NativeS3FileSystemFactory.MAX_CONNECTIONS, 0);
+ config.setString("io.tmp.dirs", System.getProperty("java.io.tmpdir"));
+
+ factory.configure(config);
+
+ URI fsUri = URI.create("s3://test-bucket/");
+ assertThatThrownBy(() -> factory.create(fsUri))
+ .isInstanceOf(IllegalArgumentException.class);
Review Comment:
added the change . PTAL
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39482][filesystem] Support configurable maxConnections in S3ClientProvider [flink]
Posted by "Samrat002 (via GitHub)" <gi...@apache.org>.
Samrat002 commented on PR #27970:
URL: https://github.com/apache/flink/pull/27970#issuecomment-4278407984
@gaborgsomogyi PTAL
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39482][filesystem] Support configurable maxConnections in S3ClientProvider [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #27970:
URL: https://github.com/apache/flink/pull/27970#discussion_r3109122242
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3BulkCopyHelper.java:
##########
@@ -70,10 +74,37 @@ class NativeS3BulkCopyHelper {
private final S3TransferManager transferManager;
private final int maxConcurrentCopies;
+ private final int maxConnections;
- public NativeS3BulkCopyHelper(S3TransferManager transferManager, int maxConcurrentCopies) {
+ /**
+ * Creates a new bulk copy helper.
+ *
+ * @param transferManager the S3 transfer manager for async downloads
+ * @param maxConcurrentCopies the requested maximum number of concurrent copy operations
+ * @param maxConnections the HTTP connection pool size; if {@code maxConcurrentCopies} exceeds
+ * this value, it is clamped down to prevent connection pool exhaustion
+ */
+ NativeS3BulkCopyHelper(
+ S3TransferManager transferManager, int maxConcurrentCopies, int maxConnections) {
+ checkArgument(maxConcurrentCopies > 0, "maxConcurrentCopies must be positive");
+ checkArgument(maxConnections > 0, "maxConnections must be positive");
this.transferManager = transferManager;
- this.maxConcurrentCopies = maxConcurrentCopies;
+ this.maxConnections = maxConnections;
+ if (maxConcurrentCopies > maxConnections) {
+ LOG.warn(
+ "s3.bulk-copy.max-concurrent ({}) exceeds s3.connection.max ({}). "
+ + "Clamping concurrent copies to {} to prevent connection pool exhaustion.",
Review Comment:
nits :
- I suggest saying `limiting` rather than` clamping ` as it is clearer.
- I suggest calling the maxConcurrentCopies requestedMaxConcurrentCopies and maybe you could change the getter to getMaxConcurrentCopies. I do not think the caller needs to be aware that this is effective.
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39482][filesystem] Support configurable maxConnections in S3ClientProvider [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #27970:
URL: https://github.com/apache/flink/pull/27970#discussion_r3109122242
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3BulkCopyHelper.java:
##########
@@ -70,10 +74,37 @@ class NativeS3BulkCopyHelper {
private final S3TransferManager transferManager;
private final int maxConcurrentCopies;
+ private final int maxConnections;
- public NativeS3BulkCopyHelper(S3TransferManager transferManager, int maxConcurrentCopies) {
+ /**
+ * Creates a new bulk copy helper.
+ *
+ * @param transferManager the S3 transfer manager for async downloads
+ * @param maxConcurrentCopies the requested maximum number of concurrent copy operations
+ * @param maxConnections the HTTP connection pool size; if {@code maxConcurrentCopies} exceeds
+ * this value, it is clamped down to prevent connection pool exhaustion
+ */
+ NativeS3BulkCopyHelper(
+ S3TransferManager transferManager, int maxConcurrentCopies, int maxConnections) {
+ checkArgument(maxConcurrentCopies > 0, "maxConcurrentCopies must be positive");
+ checkArgument(maxConnections > 0, "maxConnections must be positive");
this.transferManager = transferManager;
- this.maxConcurrentCopies = maxConcurrentCopies;
+ this.maxConnections = maxConnections;
+ if (maxConcurrentCopies > maxConnections) {
+ LOG.warn(
+ "s3.bulk-copy.max-concurrent ({}) exceeds s3.connection.max ({}). "
+ + "Clamping concurrent copies to {} to prevent connection pool exhaustion.",
Review Comment:
nit : I suggest saying `limiting` rather than` clamping ` as it is clearer.
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39482][filesystem] Support configurable maxConnections in S3ClientProvider [flink]
Posted by "gaborgsomogyi (via GitHub)" <gi...@apache.org>.
gaborgsomogyi commented on code in PR #27970:
URL: https://github.com/apache/flink/pull/27970#discussion_r3109655746
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3BulkCopyHelperTest.java:
##########
@@ -105,4 +110,42 @@ void testExtractKeyVeryLongPath() {
path.append("file.txt");
assertThat(helper.extractKey("s3://bucket/" + path)).isEqualTo(path.toString());
}
+
+ @Test
Review Comment:
Maybe parameterized with exception, `true|false`?
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactoryTest.java:
##########
@@ -353,6 +353,87 @@ void testEmptyRegionFallsBackToAutodiscovery() throws Exception {
}
}
+ @Test
+ void testInvalidMaxConnectionsThrowsException() {
+ NativeS3FileSystemFactory factory = new NativeS3FileSystemFactory();
+ Configuration config = new Configuration();
+ config.setString("s3.access-key", "test-access-key");
+ config.setString("s3.secret-key", "test-secret-key");
+ config.setString("s3.region", "us-east-1");
+ config.set(NativeS3FileSystemFactory.MAX_CONNECTIONS, 0);
+ config.setString("io.tmp.dirs", System.getProperty("java.io.tmpdir"));
+
+ factory.configure(config);
+
+ URI fsUri = URI.create("s3://test-bucket/");
+ assertThatThrownBy(() -> factory.create(fsUri))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ void testInvalidBulkCopyMaxConcurrentThrowsException() {
+ NativeS3FileSystemFactory factory = new NativeS3FileSystemFactory();
+ Configuration config = new Configuration();
+ config.setString("s3.access-key", "test-access-key");
+ config.setString("s3.secret-key", "test-secret-key");
+ config.setString("s3.region", "us-east-1");
+ config.set(NativeS3FileSystemFactory.BULK_COPY_MAX_CONCURRENT, 0);
+ config.setString("io.tmp.dirs", System.getProperty("java.io.tmpdir"));
+
+ factory.configure(config);
+
+ URI fsUri = URI.create("s3://test-bucket/");
+ assertThatThrownBy(() -> factory.create(fsUri))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("maxConcurrentCopies must be positive");
+ }
+
+ @Test
+ void testBulkCopyMaxConcurrentClampedToMaxConnections() throws Exception {
+ NativeS3FileSystemFactory factory = new NativeS3FileSystemFactory();
+ Configuration config = new Configuration();
+ config.setString("s3.access-key", "test-access-key");
+ config.setString("s3.secret-key", "test-secret-key");
+ config.setString("s3.region", "us-east-1");
+ config.set(NativeS3FileSystemFactory.BULK_COPY_ENABLED, true);
+ config.set(NativeS3FileSystemFactory.BULK_COPY_MAX_CONCURRENT, 32);
+ config.set(NativeS3FileSystemFactory.MAX_CONNECTIONS, 10);
+ config.setString("io.tmp.dirs", System.getProperty("java.io.tmpdir"));
+
+ factory.configure(config);
+
+ URI fsUri = URI.create("s3://test-bucket/");
+ FileSystem fs = factory.create(fsUri);
+
+ assertThat(fs).isInstanceOf(NativeS3FileSystem.class);
+ NativeS3FileSystem nativeFs = (NativeS3FileSystem) fs;
+ assertThat(nativeFs.getBulkCopyHelper()).isNotNull();
+ assertThat(nativeFs.getBulkCopyHelper().getEffectiveMaxConcurrentCopies()).isEqualTo(10);
Review Comment:
This is silent killer from user perspective because the intention is ignored. I would throw exception where we explain the situation.
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3BulkCopyHelper.java:
##########
@@ -181,8 +226,42 @@ private void waitForCopies(List<CompletableFuture<CompletedCopy>> futures) throw
Thread.currentThread().interrupt();
throw new IOException("Bulk copy interrupted", e);
} catch (ExecutionException e) {
- throw new IOException("Bulk copy failed", e.getCause());
+ Throwable cause = e.getCause();
+ if (isConnectionPoolExhausted(cause)) {
Review Comment:
Could this happen is we set proper boundaries?
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3BulkCopyHelper.java:
##########
@@ -181,8 +226,42 @@ private void waitForCopies(List<CompletableFuture<CompletedCopy>> futures) throw
Thread.currentThread().interrupt();
throw new IOException("Bulk copy interrupted", e);
} catch (ExecutionException e) {
- throw new IOException("Bulk copy failed", e.getCause());
+ Throwable cause = e.getCause();
+ if (isConnectionPoolExhausted(cause)) {
+ throw new IOException(
+ String.format(
+ "S3 connection pool exhausted during bulk copy. "
+ + "The configured connection pool size (%d) could not serve "
+ + "the concurrent download requests (%d). "
+ + "Consider reducing '%s' or increasing '%s'.",
+ maxConnections,
+ maxConcurrentCopies,
+ "s3.bulk-copy.max-concurrent",
+ "s3.connection.max"),
Review Comment:
Maybe we can use the key value to avoid copy-paste
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactoryTest.java:
##########
@@ -353,6 +353,87 @@ void testEmptyRegionFallsBackToAutodiscovery() throws Exception {
}
}
+ @Test
+ void testInvalidMaxConnectionsThrowsException() {
+ NativeS3FileSystemFactory factory = new NativeS3FileSystemFactory();
+ Configuration config = new Configuration();
+ config.setString("s3.access-key", "test-access-key");
+ config.setString("s3.secret-key", "test-secret-key");
+ config.setString("s3.region", "us-east-1");
+ config.set(NativeS3FileSystemFactory.MAX_CONNECTIONS, 0);
+ config.setString("io.tmp.dirs", System.getProperty("java.io.tmpdir"));
+
+ factory.configure(config);
+
+ URI fsUri = URI.create("s3://test-bucket/");
+ assertThatThrownBy(() -> factory.create(fsUri))
+ .isInstanceOf(IllegalArgumentException.class);
Review Comment:
As a general saying it would be good to check message with the significant part like `message.contains("s3.connection.max must be positive whole number")` or something similar.
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3BulkCopyHelper.java:
##########
@@ -181,8 +226,42 @@ private void waitForCopies(List<CompletableFuture<CompletedCopy>> futures) throw
Thread.currentThread().interrupt();
throw new IOException("Bulk copy interrupted", e);
} catch (ExecutionException e) {
- throw new IOException("Bulk copy failed", e.getCause());
+ Throwable cause = e.getCause();
+ if (isConnectionPoolExhausted(cause)) {
+ throw new IOException(
+ String.format(
+ "S3 connection pool exhausted during bulk copy. "
+ + "The configured connection pool size (%d) could not serve "
+ + "the concurrent download requests (%d). "
+ + "Consider reducing '%s' or increasing '%s'.",
Review Comment:
Like the exaplanation👍
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactoryTest.java:
##########
@@ -353,6 +353,87 @@ void testEmptyRegionFallsBackToAutodiscovery() throws Exception {
}
}
+ @Test
+ void testInvalidMaxConnectionsThrowsException() {
+ NativeS3FileSystemFactory factory = new NativeS3FileSystemFactory();
+ Configuration config = new Configuration();
+ config.setString("s3.access-key", "test-access-key");
+ config.setString("s3.secret-key", "test-secret-key");
+ config.setString("s3.region", "us-east-1");
+ config.set(NativeS3FileSystemFactory.MAX_CONNECTIONS, 0);
+ config.setString("io.tmp.dirs", System.getProperty("java.io.tmpdir"));
+
+ factory.configure(config);
+
+ URI fsUri = URI.create("s3://test-bucket/");
+ assertThatThrownBy(() -> factory.create(fsUri))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ void testInvalidBulkCopyMaxConcurrentThrowsException() {
+ NativeS3FileSystemFactory factory = new NativeS3FileSystemFactory();
+ Configuration config = new Configuration();
+ config.setString("s3.access-key", "test-access-key");
+ config.setString("s3.secret-key", "test-secret-key");
+ config.setString("s3.region", "us-east-1");
+ config.set(NativeS3FileSystemFactory.BULK_COPY_MAX_CONCURRENT, 0);
+ config.setString("io.tmp.dirs", System.getProperty("java.io.tmpdir"));
+
+ factory.configure(config);
+
+ URI fsUri = URI.create("s3://test-bucket/");
+ assertThatThrownBy(() -> factory.create(fsUri))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("maxConcurrentCopies must be positive");
+ }
+
+ @Test
+ void testBulkCopyMaxConcurrentClampedToMaxConnections() throws Exception {
+ NativeS3FileSystemFactory factory = new NativeS3FileSystemFactory();
+ Configuration config = new Configuration();
+ config.setString("s3.access-key", "test-access-key");
+ config.setString("s3.secret-key", "test-secret-key");
+ config.setString("s3.region", "us-east-1");
+ config.set(NativeS3FileSystemFactory.BULK_COPY_ENABLED, true);
+ config.set(NativeS3FileSystemFactory.BULK_COPY_MAX_CONCURRENT, 32);
+ config.set(NativeS3FileSystemFactory.MAX_CONNECTIONS, 10);
Review Comment:
Just for my own understanding. BULK_COPY_MAX_CONCURRENT drives bulk operation concurrency but then what exactly MAX_CONNECTIONS drives (I read the config explanation but that's a bit cloudy)?
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39482][filesystem] Support configurable maxConnections in S3ClientProvider [flink]
Posted by "gaborgsomogyi (via GitHub)" <gi...@apache.org>.
gaborgsomogyi commented on code in PR #27970:
URL: https://github.com/apache/flink/pull/27970#discussion_r3110854788
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactoryTest.java:
##########
@@ -353,6 +353,89 @@ void testEmptyRegionFallsBackToAutodiscovery() throws Exception {
}
}
+ @Test
+ void testInvalidMaxConnectionsThrowsException() {
+ NativeS3FileSystemFactory factory = new NativeS3FileSystemFactory();
+ Configuration config = new Configuration();
+ config.setString("s3.access-key", "test-access-key");
+ config.setString("s3.secret-key", "test-secret-key");
+ config.setString("s3.region", "us-east-1");
+ config.set(NativeS3FileSystemFactory.MAX_CONNECTIONS, 0);
+ config.setString("io.tmp.dirs", System.getProperty("java.io.tmpdir"));
+
+ factory.configure(config);
+
+ URI fsUri = URI.create("s3://test-bucket/");
+ assertThatThrownBy(() -> factory.create(fsUri))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("s3.connection.max")
+ .hasMessageContaining("must be a positive integer");
+ }
+
+ @Test
+ void testInvalidBulkCopyMaxConcurrentThrowsException() {
+ NativeS3FileSystemFactory factory = new NativeS3FileSystemFactory();
+ Configuration config = new Configuration();
+ config.setString("s3.access-key", "test-access-key");
+ config.setString("s3.secret-key", "test-secret-key");
+ config.setString("s3.region", "us-east-1");
+ config.set(NativeS3FileSystemFactory.BULK_COPY_MAX_CONCURRENT, 0);
+ config.setString("io.tmp.dirs", System.getProperty("java.io.tmpdir"));
+
+ factory.configure(config);
+
+ URI fsUri = URI.create("s3://test-bucket/");
+ assertThatThrownBy(() -> factory.create(fsUri))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("maxConcurrentCopies must be positive");
Review Comment:
Maybe the following to have a single pattern
```
.hasMessageContaining("s3.bulk-copy.max-concurrent")
.hasMessageContaining("must be a positive integer");
```
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactoryTest.java:
##########
@@ -353,6 +353,87 @@ void testEmptyRegionFallsBackToAutodiscovery() throws Exception {
}
}
+ @Test
+ void testInvalidMaxConnectionsThrowsException() {
+ NativeS3FileSystemFactory factory = new NativeS3FileSystemFactory();
+ Configuration config = new Configuration();
+ config.setString("s3.access-key", "test-access-key");
+ config.setString("s3.secret-key", "test-secret-key");
+ config.setString("s3.region", "us-east-1");
+ config.set(NativeS3FileSystemFactory.MAX_CONNECTIONS, 0);
+ config.setString("io.tmp.dirs", System.getProperty("java.io.tmpdir"));
+
+ factory.configure(config);
+
+ URI fsUri = URI.create("s3://test-bucket/");
+ assertThatThrownBy(() -> factory.create(fsUri))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ void testInvalidBulkCopyMaxConcurrentThrowsException() {
+ NativeS3FileSystemFactory factory = new NativeS3FileSystemFactory();
+ Configuration config = new Configuration();
+ config.setString("s3.access-key", "test-access-key");
+ config.setString("s3.secret-key", "test-secret-key");
+ config.setString("s3.region", "us-east-1");
+ config.set(NativeS3FileSystemFactory.BULK_COPY_MAX_CONCURRENT, 0);
+ config.setString("io.tmp.dirs", System.getProperty("java.io.tmpdir"));
+
+ factory.configure(config);
+
+ URI fsUri = URI.create("s3://test-bucket/");
+ assertThatThrownBy(() -> factory.create(fsUri))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("maxConcurrentCopies must be positive");
+ }
+
+ @Test
+ void testBulkCopyMaxConcurrentClampedToMaxConnections() throws Exception {
+ NativeS3FileSystemFactory factory = new NativeS3FileSystemFactory();
+ Configuration config = new Configuration();
+ config.setString("s3.access-key", "test-access-key");
+ config.setString("s3.secret-key", "test-secret-key");
+ config.setString("s3.region", "us-east-1");
+ config.set(NativeS3FileSystemFactory.BULK_COPY_ENABLED, true);
+ config.set(NativeS3FileSystemFactory.BULK_COPY_MAX_CONCURRENT, 32);
+ config.set(NativeS3FileSystemFactory.MAX_CONNECTIONS, 10);
+ config.setString("io.tmp.dirs", System.getProperty("java.io.tmpdir"));
+
+ factory.configure(config);
+
+ URI fsUri = URI.create("s3://test-bucket/");
+ FileSystem fs = factory.create(fsUri);
+
+ assertThat(fs).isInstanceOf(NativeS3FileSystem.class);
+ NativeS3FileSystem nativeFs = (NativeS3FileSystem) fs;
+ assertThat(nativeFs.getBulkCopyHelper()).isNotNull();
+ assertThat(nativeFs.getBulkCopyHelper().getEffectiveMaxConcurrentCopies()).isEqualTo(10);
Review Comment:
I was thinking about it and the main philosophy from my perspective is `just works` with the least amunt of config tuning (one config change pulls the other one too). This fits into this to re-considered my initial thought so we're fine. Of course this is valid as long as workloads with clamped values are working like charm🙂
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39482][filesystem] Support configurable maxConnections in S3ClientProvider [flink]
Posted by "gaborgsomogyi (via GitHub)" <gi...@apache.org>.
gaborgsomogyi commented on PR #27970:
URL: https://github.com/apache/flink/pull/27970#issuecomment-4282606776
After the nit fix + other comments resolution it's good to go from my perspective.
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39482][filesystem] Support configurable maxConnections in S3ClientProvider [flink]
Posted by "Samrat002 (via GitHub)" <gi...@apache.org>.
Samrat002 commented on code in PR #27970:
URL: https://github.com/apache/flink/pull/27970#discussion_r3110600877
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3BulkCopyHelper.java:
##########
@@ -113,12 +152,18 @@ public void copyFiles(
}
if (copyFutures.size() >= maxConcurrentCopies || i == requests.size() - 1) {
+ batchNumber++;
+ LOG.debug(
+ "Waiting for batch {}/{} ({} files)",
+ batchNumber,
+ totalBatches,
+ copyFutures.size());
waitForCopies(copyFutures);
copyFutures.clear();
}
}
- LOG.info("Completed bulk copy of {} files", requests.size());
+ LOG.info("Completed bulk copy of {} files", totalFiles);
Review Comment:
Good question.
Each CopyRequest maps to exactly one file by contract of
https://github.com/apache/flink/blob/master/flink-core/src/main/java/org/apache/flink/core/fs/PathsCopyingFileSystem.java#L46-L55.
it's a single getSource() → getDestination() pair with a getSize() for that one file. So requests.size() and "total files" are always identical, which is why totalFiles = requests.size() is a direct assignment.
The variable was renamed purely for readability in the log statements that follow.
IMO, no additional logging needed since they represent the same value.
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39482][filesystem] Support configurable maxConnections in S3ClientProvider [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #27970:
URL: https://github.com/apache/flink/pull/27970#discussion_r3109241098
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3BulkCopyHelperTest.java:
##########
@@ -105,4 +110,42 @@ void testExtractKeyVeryLongPath() {
path.append("file.txt");
assertThat(helper.extractKey("s3://bucket/" + path)).isEqualTo(path.toString());
}
+
+ @Test
+ void testConnectionPoolExhaustedDetection() {
+ assertThat(
+ NativeS3BulkCopyHelper.isConnectionPoolExhausted(
+ SdkClientException.builder()
+ .message(
+ "Unable to execute HTTP request: "
+ + "Acquire operation took longer than the configured maximum time.")
+ .build()))
+ .isTrue();
+ assertThat(
+ NativeS3BulkCopyHelper.isConnectionPoolExhausted(
+ SdkClientException.builder()
+ .message("Unable to execute HTTP request")
+ .cause(
+ new RuntimeException(
+ "channel acquisition failed",
+ new java.util.concurrent.TimeoutException(
Review Comment:
If the connection pool is exhausted - I assume we wait to give it time to recover? I suggest having a test that starts with it exhausted then it becomes available and carries on working.
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39482][filesystem] Support configurable maxConnections in S3ClientProvider [flink]
Posted by "Samrat002 (via GitHub)" <gi...@apache.org>.
Samrat002 commented on code in PR #27970:
URL: https://github.com/apache/flink/pull/27970#discussion_r3110623978
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3BulkCopyHelperTest.java:
##########
@@ -105,4 +110,42 @@ void testExtractKeyVeryLongPath() {
path.append("file.txt");
assertThat(helper.extractKey("s3://bucket/" + path)).isEqualTo(path.toString());
}
+
+ @Test
+ void testConnectionPoolExhaustedDetection() {
+ assertThat(
+ NativeS3BulkCopyHelper.isConnectionPoolExhausted(
+ SdkClientException.builder()
+ .message(
+ "Unable to execute HTTP request: "
+ + "Acquire operation took longer than the configured maximum time.")
+ .build()))
+ .isTrue();
+ assertThat(
+ NativeS3BulkCopyHelper.isConnectionPoolExhausted(
+ SdkClientException.builder()
+ .message("Unable to execute HTTP request")
+ .cause(
+ new RuntimeException(
+ "channel acquisition failed",
+ new java.util.concurrent.TimeoutException(
Review Comment:
To clarify the design, we don't wait-and-retry on pool exhaustion. The behaviour is fail-fast. when the SDK's connection acquire timeout expires, we catch it and throw an actionable IOException that tells the operator exactly which config keys to adjust (`s3.bulk-copy.max-concurrent / s3.connection.max`).
There is no recovery path within a single copyFiles invocation. Flink's checkpoint restore framework handles the retry at a higher level.
The primary defence is the clamping logic, which prevents the most common exhaustion scenario upfront. The `isConnectionPoolExhausted` detection is defence-in-depth for edge cases where S3TransferManager's multipart downloads consume multiple HTTP connections per file (files > 8MB are split into parts, each part = separate connection), which can exceed the pool even after clamping.
A test simulating real pool exhaustion → recovery would require standing up a real Netty connection pool against an S3-compatible endpoint and inducing timing-dependent acquire timeouts, which will be inherently flaky in CI. The detection logic is covered by the message-matching unit test, which validates that we correctly identify the SDK's characteristic "Acquire operation took longer than" message in both direct and nested causal chains.
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39482][filesystem] Support configurable maxConnections in S3ClientProvider [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #27970:
URL: https://github.com/apache/flink/pull/27970#issuecomment-4276273814
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b0ea378c64d5c0d787029c0330e3acc022d93f51",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "b0ea378c64d5c0d787029c0330e3acc022d93f51",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b0ea378c64d5c0d787029c0330e3acc022d93f51 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39482][filesystem] Support configurable maxConnections in S3ClientProvider [flink]
Posted by "Samrat002 (via GitHub)" <gi...@apache.org>.
Samrat002 commented on code in PR #27970:
URL: https://github.com/apache/flink/pull/27970#discussion_r3110692287
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactoryTest.java:
##########
@@ -353,6 +353,87 @@ void testEmptyRegionFallsBackToAutodiscovery() throws Exception {
}
}
+ @Test
+ void testInvalidMaxConnectionsThrowsException() {
+ NativeS3FileSystemFactory factory = new NativeS3FileSystemFactory();
+ Configuration config = new Configuration();
+ config.setString("s3.access-key", "test-access-key");
+ config.setString("s3.secret-key", "test-secret-key");
+ config.setString("s3.region", "us-east-1");
+ config.set(NativeS3FileSystemFactory.MAX_CONNECTIONS, 0);
+ config.setString("io.tmp.dirs", System.getProperty("java.io.tmpdir"));
+
+ factory.configure(config);
+
+ URI fsUri = URI.create("s3://test-bucket/");
+ assertThatThrownBy(() -> factory.create(fsUri))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ void testInvalidBulkCopyMaxConcurrentThrowsException() {
+ NativeS3FileSystemFactory factory = new NativeS3FileSystemFactory();
+ Configuration config = new Configuration();
+ config.setString("s3.access-key", "test-access-key");
+ config.setString("s3.secret-key", "test-secret-key");
+ config.setString("s3.region", "us-east-1");
+ config.set(NativeS3FileSystemFactory.BULK_COPY_MAX_CONCURRENT, 0);
+ config.setString("io.tmp.dirs", System.getProperty("java.io.tmpdir"));
+
+ factory.configure(config);
+
+ URI fsUri = URI.create("s3://test-bucket/");
+ assertThatThrownBy(() -> factory.create(fsUri))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("maxConcurrentCopies must be positive");
+ }
+
+ @Test
+ void testBulkCopyMaxConcurrentClampedToMaxConnections() throws Exception {
+ NativeS3FileSystemFactory factory = new NativeS3FileSystemFactory();
+ Configuration config = new Configuration();
+ config.setString("s3.access-key", "test-access-key");
+ config.setString("s3.secret-key", "test-secret-key");
+ config.setString("s3.region", "us-east-1");
+ config.set(NativeS3FileSystemFactory.BULK_COPY_ENABLED, true);
+ config.set(NativeS3FileSystemFactory.BULK_COPY_MAX_CONCURRENT, 32);
+ config.set(NativeS3FileSystemFactory.MAX_CONNECTIONS, 10);
Review Comment:
They operate at different layers:
1. `s3.connection.max` - the HTTP connection pool size in the underlying HTTP clients (Apache for sync ops, Netty for async ops). This is a shared pool: every S3 API call, like GetObject, PutObject, HeadObject, ListObjectsV2, etc., borrows a connection from it and returns it when done. This maps directly to
https://github.com/apache/flink/blob/FLINK-39482/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3ClientProvider.java#L391 on the Netty async
client and maxConnections on the Apache sync client.
2. `s3.bulk-copy.max-concurrent` - how many S3 download operations NativeS3BulkCopyHelper fires in parallel during state restore (the batch size in copyFiles).
The root cause of FLINK-39482: these two layers interact because S3TransferManager uses multipart downloads for files > 8MB, each part is a separate HTTP connection from the shared pool. So `maxConcurrentCopies=16 files × ~4 parts each = ~64` HTTP connections needed, but the pool only has 50 → acquire timeout → opaque SdkClientException.
Does this configuration make sense, or are there any suggestions to simplify it?
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39482][filesystem] Support configurable maxConnections in S3ClientProvider [flink]
Posted by "gaborgsomogyi (via GitHub)" <gi...@apache.org>.
gaborgsomogyi merged PR #27970:
URL: https://github.com/apache/flink/pull/27970
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org