This is an automated email from the ASF dual-hosted git repository.
snuyanzin 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 bb8599fa780 Revert "[FLINK-39900][forst] Add configuration option for
setting the number of ForSt transfer threads (#28373)"
bb8599fa780 is described below
commit bb8599fa780a157085f972fdfc58fa6cba80e8a4
Author: Sergey Nuyanzin <[email protected]>
AuthorDate: Sat Aug 1 00:40:14 2026 +0200
Revert "[FLINK-39900][forst] Add configuration option for setting the
number of ForSt transfer threads (#28373)"
This reverts commit 01a5a7f78b4b52417d19cf061a12e5cc110f16c0.
---
.../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, 8 insertions(+), 52 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/expert_forst_section.html
b/docs/layouts/shortcodes/generated/expert_forst_section.html
index e28f17e855a..81ec1999458 100644
--- a/docs/layouts/shortcodes/generated/expert_forst_section.html
+++ b/docs/layouts/shortcodes/generated/expert_forst_section.html
@@ -20,12 +20,6 @@
<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 7c5d99395ab..f66c1af4d8b 100644
--- a/docs/layouts/shortcodes/generated/forst_configuration.html
+++ b/docs/layouts/shortcodes/generated/forst_configuration.html
@@ -38,12 +38,6 @@
<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 68aa3918712..d9b2bc3dd58 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(
- optionsContainer.getDataTransferThreadNum(),
+ ForStStateDataTransfer.DEFAULT_THREAD_NUM,
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 0ecc7cdc9c8..8a816e3e1b1 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,19 +300,4 @@ 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 6d7b225af23..cb29cf1857c 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,10 +324,6 @@ 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 852fa08405e..0305f8cd02e 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,6 +64,9 @@ 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 ca3d683a484..f205fe21b9f 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(
- optionsContainer.getDataTransferThreadNum(),
+ ForStStateDataTransfer.DEFAULT_THREAD_NUM,
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 0f6dcac1e73..0e83b22a0ce 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(
- optionsContainer.getDataTransferThreadNum(),
+ ForStStateDataTransfer.DEFAULT_THREAD_NUM,
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 ef478795dd3..1505057f8e2 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,19 +815,6 @@ 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 92411da631f..034df702554 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,7 +33,6 @@ 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;
@@ -200,8 +199,7 @@ class ForStIncrementalSnapshotStrategyTest {
CompositeKeySerializationUtils.computeRequiredBytesInKeyGroupPrefix(2),
UUID.randomUUID(),
new TreeMap<>(),
- new ForStStateDataTransfer(
-
ForStOptions.CHECKPOINT_TRANSFER_THREAD_NUM.defaultValue()),
+ new
ForStStateDataTransfer(ForStStateDataTransfer.DEFAULT_THREAD_NUM),
-1);
}
@@ -233,8 +231,7 @@ class ForStIncrementalSnapshotStrategyTest {
new KeyGroupRange(0, 1),
CompositeKeySerializationUtils.computeRequiredBytesInKeyGroupPrefix(2),
UUID.randomUUID(),
- new ForStStateDataTransfer(
-
ForStOptions.CHECKPOINT_TRANSFER_THREAD_NUM.defaultValue()));
+ new
ForStStateDataTransfer(ForStStateDataTransfer.DEFAULT_THREAD_NUM));
}
private FsCheckpointStreamFactory createFsCheckpointStreamFactory() throws
IOException {
