This is an automated email from the ASF dual-hosted git repository.
Zakelly pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 01a5a7f78b4 [FLINK-39900][forst] Add configuration option for setting
the number of ForSt transfer threads (#28373)
01a5a7f78b4 is described below
commit 01a5a7f78b4b52417d19cf061a12e5cc110f16c0
Author: Francis <[email protected]>
AuthorDate: Fri Jul 31 15:47:28 2026 +0200
[FLINK-39900][forst] Add configuration option for setting the number of
ForSt transfer threads (#28373)
---------
Co-authored-by: Purushottam Sinha <[email protected]>
---
.../shortcodes/generated/expert_forst_section.html | 6 ++++++
.../layouts/shortcodes/generated/forst_configuration.html | 6 ++++++
.../flink/state/forst/ForStKeyedStateBackendBuilder.java | 2 +-
.../java/org/apache/flink/state/forst/ForStOptions.java | 15 +++++++++++++++
.../apache/flink/state/forst/ForStResourceContainer.java | 4 ++++
.../state/forst/datatransfer/ForStStateDataTransfer.java | 3 ---
.../forst/restore/ForStIncrementalRestoreOperation.java | 2 +-
.../forst/sync/ForStSyncKeyedStateBackendBuilder.java | 2 +-
.../flink/state/forst/ForStStateBackendConfigTest.java | 13 +++++++++++++
.../snapshot/ForStIncrementalSnapshotStrategyTest.java | 7 +++++--
10 files changed, 52 insertions(+), 8 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/expert_forst_section.html
b/docs/layouts/shortcodes/generated/expert_forst_section.html
index 81ec1999458..e28f17e855a 100644
--- a/docs/layouts/shortcodes/generated/expert_forst_section.html
+++ b/docs/layouts/shortcodes/generated/expert_forst_section.html
@@ -20,6 +20,12 @@
<td>Integer</td>
<td>When the number of eviction that a block in hot link is moved
to cold link reaches this value, the block will be blocked from being promoted
to the head of the LRU list. The default value is '3'.</td>
</tr>
+ <tr>
+
<td><h5>state.backend.forst.checkpoint.transfer.thread.num</h5></td>
+ <td style="word-wrap: break-word;">4</td>
+ <td>Integer</td>
+ <td>The number of threads used to transfer files during checkpoint
(writing or copying files to the checkpoint storage) and restore (transferring
state files back to the ForSt working directory). Consider increasing this
value when snapshotting or restoring large state. Note that the pool is created
per state backend instance. Setting it too high can saturate network bandwidth
or trigger rate limiting depending on the remote storage; setting it too low
can lead to long checkp [...]
+ </tr>
<tr>
<td><h5>state.backend.forst.executor.inline-coordinator</h5></td>
<td style="word-wrap: break-word;">false</td>
diff --git a/docs/layouts/shortcodes/generated/forst_configuration.html
b/docs/layouts/shortcodes/generated/forst_configuration.html
index f66c1af4d8b..7c5d99395ab 100644
--- a/docs/layouts/shortcodes/generated/forst_configuration.html
+++ b/docs/layouts/shortcodes/generated/forst_configuration.html
@@ -38,6 +38,12 @@
<td>MemorySize</td>
<td>An upper-bound of the size that can be used for cache. User
should specify at least one cache size limit to enable the cache, either this
option or the 'state.backend.forst.cache.reserve-size' option. They can be set
simultaneously, and in this case, cache will grow if meet the requirements of
both two options. The default value is '0 bytes', meaning that this option is
disabled. </td>
</tr>
+ <tr>
+
<td><h5>state.backend.forst.checkpoint.transfer.thread.num</h5></td>
+ <td style="word-wrap: break-word;">4</td>
+ <td>Integer</td>
+ <td>The number of threads used to transfer files during checkpoint
(writing or copying files to the checkpoint storage) and restore (transferring
state files back to the ForSt working directory). Consider increasing this
value when snapshotting or restoring large state. Note that the pool is created
per state backend instance. Setting it too high can saturate network bandwidth
or trigger rate limiting depending on the remote storage; setting it too low
can lead to long checkp [...]
+ </tr>
<tr>
<td><h5>state.backend.forst.executor.inline-coordinator</h5></td>
<td style="word-wrap: break-word;">false</td>
diff --git
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackendBuilder.java
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackendBuilder.java
index d9b2bc3dd58..68aa3918712 100644
---
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackendBuilder.java
+++
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackendBuilder.java
@@ -442,7 +442,7 @@ public class ForStKeyedStateBackendBuilder<K>
long lastCompletedCheckpointId) {
ForStStateDataTransfer stateTransfer =
new ForStStateDataTransfer(
- ForStStateDataTransfer.DEFAULT_THREAD_NUM,
+ optionsContainer.getDataTransferThreadNum(),
optionsContainer.getFileSystem());
if (enableIncrementalCheckpointing) {
diff --git
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptions.java
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptions.java
index 8a816e3e1b1..0ecc7cdc9c8 100644
---
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptions.java
+++
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptions.java
@@ -300,4 +300,19 @@ public class ForStOptions {
+ " Only valid when '"
+ EXECUTOR_WRITE_IO_INLINE.key()
+ "' is false.");
+
+ @Documentation.Section(Documentation.Sections.EXPERT_FORST)
+ public static final ConfigOption<Integer> CHECKPOINT_TRANSFER_THREAD_NUM =
+
ConfigOptions.key("state.backend.forst.checkpoint.transfer.thread.num")
+ .intType()
+ .defaultValue(4)
+ .withDescription(
+ "The number of threads used to transfer files
during checkpoint (writing or copying "
+ + "files to the checkpoint storage) and
restore (transferring state files back to "
+ + "the ForSt working directory). Consider
increasing this value when snapshotting or "
+ + "restoring large state. Note that the
pool is created per state backend instance. "
+ + "Setting it too high can saturate
network bandwidth or trigger rate limiting depending "
+ + "on the remote storage; setting it too
low can lead to long checkpoint durations "
+ + "or timeouts for large state. "
+ + "The default value is '4'.");
}
diff --git
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java
index cb29cf1857c..6d7b225af23 100644
---
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java
+++
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java
@@ -324,6 +324,10 @@ public final class ForStResourceContainer implements
AutoCloseable {
return configuration.get(ForStOptions.EXECUTOR_WRITE_IO_PARALLELISM);
}
+ public int getDataTransferThreadNum() {
+ return configuration.get(ForStOptions.CHECKPOINT_TRANSFER_THREAD_NUM);
+ }
+
/**
* Prepare local and remote directories.
*
diff --git
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/datatransfer/ForStStateDataTransfer.java
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/datatransfer/ForStStateDataTransfer.java
index 0305f8cd02e..852fa08405e 100644
---
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/datatransfer/ForStStateDataTransfer.java
+++
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/datatransfer/ForStStateDataTransfer.java
@@ -64,9 +64,6 @@ import static
org.apache.flink.util.concurrent.Executors.newDirectExecutorServic
public class ForStStateDataTransfer implements Closeable {
private static final Logger LOG =
LoggerFactory.getLogger(ForStStateDataTransfer.class);
- // TODO: Add ConfigOption replace this field after ForSt checkpoint
implementation stable
- public static final int DEFAULT_THREAD_NUM = 4;
-
protected final ExecutorService executorService;
@Nullable private final ForStFlinkFileSystem forStFs;
diff --git
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStIncrementalRestoreOperation.java
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStIncrementalRestoreOperation.java
index f205fe21b9f..ca3d683a484 100644
---
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStIncrementalRestoreOperation.java
+++
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStIncrementalRestoreOperation.java
@@ -265,7 +265,7 @@ public class ForStIncrementalRestoreOperation<K> implements
ForStRestoreOperatio
private void transferAllStateHandles(List<StateHandleTransferSpec> specs)
throws Exception {
try (ForStStateDataTransfer transfer =
new ForStStateDataTransfer(
- ForStStateDataTransfer.DEFAULT_THREAD_NUM,
+ optionsContainer.getDataTransferThreadNum(),
optionsContainer.getFileSystem())) {
transfer.transferAllStateDataToDirectory(
optionsContainer.getPathContainer(),
diff --git
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/sync/ForStSyncKeyedStateBackendBuilder.java
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/sync/ForStSyncKeyedStateBackendBuilder.java
index 0e83b22a0ce..0f6dcac1e73 100644
---
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/sync/ForStSyncKeyedStateBackendBuilder.java
+++
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/sync/ForStSyncKeyedStateBackendBuilder.java
@@ -465,7 +465,7 @@ public class ForStSyncKeyedStateBackendBuilder<K> extends
AbstractKeyedStateBack
ForStStateDataTransfer stateTransfer =
new ForStStateDataTransfer(
- ForStStateDataTransfer.DEFAULT_THREAD_NUM,
+ optionsContainer.getDataTransferThreadNum(),
optionsContainer.getFileSystem());
if (enableIncrementalCheckpointing) {
diff --git
a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendConfigTest.java
b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendConfigTest.java
index 1505057f8e2..ef478795dd3 100644
---
a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendConfigTest.java
+++
b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendConfigTest.java
@@ -815,6 +815,19 @@ class ForStStateBackendConfigTest {
}
}
+ @Test
+ public void testConfigureCheckpointTransferThreadNumber() throws Exception
{
+ ForStStateBackend forStStateBackend = new ForStStateBackend();
+ Configuration configuration = new Configuration();
+
configuration.setString(ForStOptions.CHECKPOINT_TRANSFER_THREAD_NUM.key(),
"10");
+ forStStateBackend = forStStateBackend.configure(configuration,
getClass().getClassLoader());
+
+ try (ForStResourceContainer resourceContainer =
+ forStStateBackend.createOptionsAndResourceContainer(null)) {
+ assertEquals(10, resourceContainer.getDataTransferThreadNum());
+ }
+ }
+
private void verifySetParameter(Runnable setter) {
assertThatThrownBy(setter::run).isInstanceOf(IllegalArgumentException.class);
}
diff --git
a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/snapshot/ForStIncrementalSnapshotStrategyTest.java
b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/snapshot/ForStIncrementalSnapshotStrategyTest.java
index 034df702554..92411da631f 100644
---
a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/snapshot/ForStIncrementalSnapshotStrategyTest.java
+++
b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/snapshot/ForStIncrementalSnapshotStrategyTest.java
@@ -33,6 +33,7 @@ import
org.apache.flink.runtime.state.filesystem.FsCheckpointStreamFactory;
import
org.apache.flink.runtime.state.v2.RegisteredKeyValueStateBackendMetaInfo;
import org.apache.flink.state.forst.ForStExtension;
import org.apache.flink.state.forst.ForStOperationUtils;
+import org.apache.flink.state.forst.ForStOptions;
import org.apache.flink.state.forst.datatransfer.ForStStateDataTransfer;
import org.apache.flink.testutils.junit.utils.TempDirUtils;
@@ -199,7 +200,8 @@ class ForStIncrementalSnapshotStrategyTest {
CompositeKeySerializationUtils.computeRequiredBytesInKeyGroupPrefix(2),
UUID.randomUUID(),
new TreeMap<>(),
- new
ForStStateDataTransfer(ForStStateDataTransfer.DEFAULT_THREAD_NUM),
+ new ForStStateDataTransfer(
+
ForStOptions.CHECKPOINT_TRANSFER_THREAD_NUM.defaultValue()),
-1);
}
@@ -231,7 +233,8 @@ class ForStIncrementalSnapshotStrategyTest {
new KeyGroupRange(0, 1),
CompositeKeySerializationUtils.computeRequiredBytesInKeyGroupPrefix(2),
UUID.randomUUID(),
- new
ForStStateDataTransfer(ForStStateDataTransfer.DEFAULT_THREAD_NUM));
+ new ForStStateDataTransfer(
+
ForStOptions.CHECKPOINT_TRANSFER_THREAD_NUM.defaultValue()));
}
private FsCheckpointStreamFactory createFsCheckpointStreamFactory() throws
IOException {
