This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new ee83c538eb [Improve][Zeta] Make the checkpoint monitor history size
configurable (#12789)
ee83c538eb is described below
commit ee83c538ebfadd29d60d5622afd9a12213f02d36
Author: SEZ <[email protected]>
AuthorDate: Wed Oct 21 23:08:54 2026 +0000
[Improve][Zeta] Make the checkpoint monitor history size configurable
(#12789)
---
.../fault-tolerance/checkpoint-mechanism.md | 25 +++++++
docs/en/engines/zeta/hybrid-cluster-deployment.md | 9 +++
.../fault-tolerance/checkpoint-mechanism.md | 21 ++++++
docs/zh/engines/zeta/hybrid-cluster-deployment.md | 9 +++
.../config/YamlSeaTunnelDomConfigProcessor.java | 9 +++
.../common/config/server/CheckpointConfig.java | 26 +++++++
.../common/config/server/ServerConfigOptions.java | 39 +++++++++++
.../config/YamlSeaTunnelConfigParserTest.java | 37 ++++++++++
.../src/test/resources/seatunnel.yaml | 1 +
.../checkpoint/PipelineCheckpointOverviewTest.java | 81 ++++++++++++++++++++++
.../seatunnel/engine/server/SeaTunnelServer.java | 23 +++++-
11 files changed, 279 insertions(+), 1 deletion(-)
diff --git a/docs/en/architecture/fault-tolerance/checkpoint-mechanism.md
b/docs/en/architecture/fault-tolerance/checkpoint-mechanism.md
index 31f55a179b..53de3a28a1 100644
--- a/docs/en/architecture/fault-tolerance/checkpoint-mechanism.md
+++ b/docs/en/architecture/fault-tolerance/checkpoint-mechanism.md
@@ -513,6 +513,31 @@ env {
Checkpoint storage is configured on the engine side (e.g.,
`config/seatunnel.yaml` under `seatunnel.engine.checkpoint.storage`), rather
than as job-level `env` options.
+The size of the checkpoint history the monitor retains is also engine-side:
+
+```yaml
+seatunnel:
+ engine:
+ checkpoint:
+ monitor-history-size: 32 # default
+```
+
+This is how many checkpoint history entries are kept per pipeline for the
+`/jobs/checkpoints/{jobId}` REST endpoint; completed and failed checkpoints
+share the window. Valid values are 1 to 4096.
+
+It bounds how far back a client can compare a job against its own past, which
+matters for spotting a gradual slowdown. Note that it is a count, not a
+duration: how much time it covers depends on each job's own
+`checkpoint.interval`, while the option itself is cluster-wide. At the default
+interval of 300000ms, 32 entries cover roughly 2.5 hours; on a job
+checkpointing every 10000ms they cover only about 5 minutes.
+
+The history is stored in the cluster-wide checkpoint overview map, and every
+checkpoint event rewrites that whole entry, so a larger window costs master
+CPU, network and backup traffic as well as heap — not heap alone. The value is
+read once at master startup, so changing it requires a restart.
+
### 5.2 Tuning Guidelines
**Checkpoint Interval**:
diff --git a/docs/en/engines/zeta/hybrid-cluster-deployment.md
b/docs/en/engines/zeta/hybrid-cluster-deployment.md
index cd81845ca4..ad12feb0b8 100644
--- a/docs/en/engines/zeta/hybrid-cluster-deployment.md
+++ b/docs/en/engines/zeta/hybrid-cluster-deployment.md
@@ -93,6 +93,14 @@ The timeout for checkpoints. If the checkpoint cannot be
completed within the ti
The minimum pause (in milliseconds) between consecutive checkpoints. This
ensures that checkpoints are not triggered too frequently.
+**monitor-history-size**
+
+How many checkpoint history entries the monitor keeps per pipeline for the
`/jobs/checkpoints/{jobId}` REST endpoint, between 1 and 4096, defaulting to
32. Values above 256 are accepted but log a warning at master startup, since
the cost below is paid on every checkpoint event. Completed and failed
checkpoints share the window.
+
+This is a count rather than a duration, so how much time it covers depends on
each job's own `checkpoint.interval` while this option is cluster-wide: 32
entries cover about 2.5 hours at the default interval of 300000ms, but only
about 5 minutes at an interval of 10000ms. Raise it if you run jobs with short
intervals and want to compare a job against a longer stretch of its own past.
+
+The history is stored in the cluster-wide checkpoint overview map, and every
checkpoint event rewrites that whole entry, so a larger window costs master
CPU, network and backup traffic as well as heap. The value is read once at
master startup, so changing it requires a restart.
+
Example
```yaml
@@ -106,6 +114,7 @@ seatunnel:
interval: 300000
timeout: 10000
min-pause: 5000
+ monitor-history-size: 32
```
diff --git a/docs/zh/architecture/fault-tolerance/checkpoint-mechanism.md
b/docs/zh/architecture/fault-tolerance/checkpoint-mechanism.md
index dce94220d3..715fffb6d9 100644
--- a/docs/zh/architecture/fault-tolerance/checkpoint-mechanism.md
+++ b/docs/zh/architecture/fault-tolerance/checkpoint-mechanism.md
@@ -356,6 +356,27 @@ seatunnel:
namespace: /tmp/seatunnel/checkpoint_snapshot
```
+监控保留的 checkpoint 历史条数同样在引擎侧配置：
+
+```yaml
+seatunnel:
+ engine:
+ checkpoint:
+ monitor-history-size: 32 # 默认值
+```
+
+它表示为 `/jobs/checkpoints/{jobId}` REST 端点按 pipeline 保留多少条 checkpoint
+历史记录，已完成和失败的 checkpoint 共用这个窗口，取值范围 1 到 4096。
+
+这个窗口决定了客户端最多能把作业和自己多久之前的历史做对比，对发现"逐渐变慢"
+尤为关键。注意它是**条数**而不是时长：覆盖多长时间取决于每个作业自己的
+`checkpoint.interval`，而本选项是集群级的。按默认 300000ms 的间隔，32 条约覆盖
+2.5 小时；若作业每 10000ms 做一次 checkpoint，则只覆盖约 5 分钟。
+
+历史保存在集群级的 checkpoint overview map 中，每次 checkpoint 事件都会重写整条
+记录，因此窗口越大，除堆内存外还会增加 master 的 CPU、网络与副本复制开销——不只是
+堆内存。该值只在 master 启动时读取一次，修改后需要重启。
+
说明：
- BATCH 模式下如果作业 env 未配置 `checkpoint.interval`，当前实现会禁用 checkpoint（以源码实现为准）。
- checkpoint storage 主要由引擎侧配置管理；作业级配置不应假设可以随意指定 storage type/path。
diff --git a/docs/zh/engines/zeta/hybrid-cluster-deployment.md
b/docs/zh/engines/zeta/hybrid-cluster-deployment.md
index 142fd79823..9d7d4b05af 100644
--- a/docs/zh/engines/zeta/hybrid-cluster-deployment.md
+++ b/docs/zh/engines/zeta/hybrid-cluster-deployment.md
@@ -93,6 +93,14 @@ seatunnel:
连续检查点之间的最小暂停时间(以毫秒为单位)，确保检查点不会频繁触发。
+**monitor-history-size**
+
+监控为 `/jobs/checkpoints/{jobId}` REST 接口按 pipeline 保留的 checkpoint 历史条数，取值 1 到
4096，默认 32。超过 256 的值仍然可用，但 master 启动时会打印一条警告——因为下面提到的开销是每次 checkpoint
事件都要付的。已完成和失败的 checkpoint 共用这个窗口。
+
+它是**条数**而不是时长，因此覆盖多长时间取决于每个作业自己的 `checkpoint.interval`，而本选项是集群级的：按默认 300000ms
的间隔，32 条约覆盖 2.5 小时；若间隔为 10000ms，则只覆盖约 5
分钟。如果集群上有间隔很短的作业，并且希望拿作业和更长一段自身历史做对比，应调大该值。
+
+历史保存在集群级的 checkpoint overview map 中，每次 checkpoint 事件都会重写整条记录，因此窗口越大，除堆内存外还会增加
master 的 CPU、网络与副本复制开销。该值只在 master 启动时读取一次，修改后需要重启。
+
示例
```yaml
@@ -106,6 +114,7 @@ seatunnel:
interval: 300000
timeout: 10000
min-pause: 5000
+ monitor-history-size: 32
```
**checkpoint storage**
diff --git
a/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/YamlSeaTunnelDomConfigProcessor.java
b/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/YamlSeaTunnelDomConfigProcessor.java
index 94f34f6dc3..6d47a3a16a 100644
---
a/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/YamlSeaTunnelDomConfigProcessor.java
+++
b/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/YamlSeaTunnelDomConfigProcessor.java
@@ -383,6 +383,15 @@ public class YamlSeaTunnelDomConfigProcessor extends
AbstractDomConfigProcessor
ServerConfigOptions.MasterServerConfigOptions.CHECKPOINT_MIN_PAUSE
.key(),
getTextContent(node)));
+ } else if
(ServerConfigOptions.MasterServerConfigOptions.CHECKPOINT_MONITOR_HISTORY_SIZE
+ .key()
+ .equals(name)) {
+ checkpointConfig.setMonitorHistorySize(
+ getIntegerValue(
+ ServerConfigOptions.MasterServerConfigOptions
+ .CHECKPOINT_MONITOR_HISTORY_SIZE
+ .key(),
+ getTextContent(node)));
} else if (ServerConfigOptions.MasterServerConfigOptions
.SCHEMA_CHANGE_CHECKPOINT_TIMEOUT
.key()
diff --git
a/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/server/CheckpointConfig.java
b/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/server/CheckpointConfig.java
index a7a39a1854..13ec39026f 100644
---
a/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/server/CheckpointConfig.java
+++
b/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/server/CheckpointConfig.java
@@ -34,6 +34,9 @@ public class CheckpointConfig implements Serializable {
ServerConfigOptions.MasterServerConfigOptions.CHECKPOINT_TIMEOUT.defaultValue();
private long checkpointMinPause =
ServerConfigOptions.MasterServerConfigOptions.CHECKPOINT_MIN_PAUSE.defaultValue();
+ private int monitorHistorySize =
+
ServerConfigOptions.MasterServerConfigOptions.CHECKPOINT_MONITOR_HISTORY_SIZE
+ .defaultValue();
private long schemaChangeCheckpointTimeout =
ServerConfigOptions.MasterServerConfigOptions.SCHEMA_CHANGE_CHECKPOINT_TIMEOUT
.defaultValue();
@@ -51,6 +54,29 @@ public class CheckpointConfig implements Serializable {
return retainAfterJobCancelled;
}
+ public void setMonitorHistorySize(int monitorHistorySize) {
+ // Bounded on both sides. The upper bound is not cosmetic: the history
+ // lives in the cluster-wide checkpoint overview IMap, and
+ // HazelcastCheckpointOverviewStateStore.updateOverview rewrites that
+ // whole entry through IMap.compute on every checkpoint event, with the
+ // entry listener registered includeValue=true. So an oversized window
+ // multiplies master CPU, replication and listener traffic on every
+ // trigger and acknowledgement, not just heap -- and it would do so
+ // silently.
+ checkArgument(
+ monitorHistorySize >= 1
+ && monitorHistorySize
+ <=
ServerConfigOptions.MasterServerConfigOptions
+ .MAX_CHECKPOINT_MONITOR_HISTORY_SIZE,
+ "The checkpoint monitor history size must be between 1 and "
+ + ServerConfigOptions.MasterServerConfigOptions
+ .MAX_CHECKPOINT_MONITOR_HISTORY_SIZE
+ + ", but was "
+ + monitorHistorySize
+ + ".");
+ this.monitorHistorySize = monitorHistorySize;
+ }
+
public void setCheckpointInterval(long checkpointInterval) {
checkArgument(
checkpointInterval >= MINIMAL_CHECKPOINT_TIME,
diff --git
a/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/server/ServerConfigOptions.java
b/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/server/ServerConfigOptions.java
index 16971f3e85..a55ae2430e 100644
---
a/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/server/ServerConfigOptions.java
+++
b/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/server/ServerConfigOptions.java
@@ -215,6 +215,45 @@ public class ServerConfigOptions {
.defaultValue(30000)
.withDescription("The timeout (in milliseconds) for a
checkpoint.");
+ /**
+ * Upper bound for {@link #CHECKPOINT_MONITOR_HISTORY_SIZE}. The
history lives in the
+ * cluster-wide checkpoint overview IMap, whose whole value is
re-serialised on every
+ * checkpoint event, so an unbounded value would multiply master CPU,
replication and
+ * listener traffic rather than just heap. A few thousand entries is
far beyond any useful
+ * comparison window while staying well clear of that cost.
+ */
+ public static final int MAX_CHECKPOINT_MONITOR_HISTORY_SIZE = 4096;
+
+ /**
+ * Above this, {@link #CHECKPOINT_MONITOR_HISTORY_SIZE} is legal but
worth a word at
+ * startup. The hard cap stops the pathological case; this catches the
merely expensive one,
+ * where an operator has chosen a window large enough to be felt on a
cluster with many
+ * pipelines or a short checkpoint interval. A warning rather than a
rejection, because
+ * whether the cost is acceptable depends on the cluster, not on the
number.
+ */
+ public static final int CHECKPOINT_MONITOR_HISTORY_SIZE_WARN_THRESHOLD
= 256;
+
+ public static final Option<Integer> CHECKPOINT_MONITOR_HISTORY_SIZE =
+ Options.key("monitor-history-size")
+ .intType()
+ .defaultValue(32)
+ .withDescription(
+ "How many checkpoint history entries the
monitor keeps per pipeline "
+ + "for the /jobs/checkpoints endpoint.
Completed and failed "
+ + "checkpoints share the window. This
is a count, not a "
+ + "duration: how much time it covers
depends on each job's "
+ + "own checkpoint.interval, while this
option is "
+ + "cluster-wide, so the same value
covers about 2.5 hours at "
+ + "the default interval of 300000ms
but only about 5 minutes "
+ + "at an interval of 10000ms. The
history is stored in the "
+ + "cluster-wide checkpoint overview
map, and every "
+ + "checkpoint event rewrites the whole
overview entry, so a "
+ + "larger window costs master CPU,
network and backup "
+ + "traffic as well as heap. Read once
at master startup, so "
+ + "a change needs a restart. Must be
between 1 and "
+ + MAX_CHECKPOINT_MONITOR_HISTORY_SIZE
+ + ".");
+
public static final Option<Integer> CHECKPOINT_MIN_PAUSE =
Options.key("min-pause")
.intType()
diff --git
a/seatunnel-engine/seatunnel-engine-common/src/test/java/org/apache/seatunnel/engine/common/config/YamlSeaTunnelConfigParserTest.java
b/seatunnel-engine/seatunnel-engine-common/src/test/java/org/apache/seatunnel/engine/common/config/YamlSeaTunnelConfigParserTest.java
index 828d2d3879..8fe4cbe4f6 100644
---
a/seatunnel-engine/seatunnel-engine-common/src/test/java/org/apache/seatunnel/engine/common/config/YamlSeaTunnelConfigParserTest.java
+++
b/seatunnel-engine/seatunnel-engine-common/src/test/java/org/apache/seatunnel/engine/common/config/YamlSeaTunnelConfigParserTest.java
@@ -18,7 +18,9 @@
package org.apache.seatunnel.engine.common.config;
import org.apache.seatunnel.common.utils.ReflectionUtils;
+import org.apache.seatunnel.engine.common.config.server.CheckpointConfig;
import org.apache.seatunnel.engine.common.config.server.HttpConfig;
+import org.apache.seatunnel.engine.common.config.server.ServerConfigOptions;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -58,6 +60,9 @@ public class YamlSeaTunnelConfigParserTest {
Assertions.assertEquals(
7000,
config.getEngineConfig().getCheckpointConfig().getCheckpointTimeout());
+ Assertions.assertEquals(
+ 64,
config.getEngineConfig().getCheckpointConfig().getMonitorHistorySize());
+
Assertions.assertEquals(
"hdfs",
config.getEngineConfig().getCheckpointConfig().getStorage().getStorage());
@@ -154,6 +159,38 @@ public class YamlSeaTunnelConfigParserTest {
Assertions.assertEquals(
Runtime.getRuntime().availableProcessors() * 2,
config.getEngineConfig().getSlotServiceConfig().getSlotNum());
+ // This yaml has no checkpoint block at all, so it pins the default
that
+ // keeps behaviour identical to the previously hard-coded value.
+ Assertions.assertEquals(
+
ServerConfigOptions.MasterServerConfigOptions.CHECKPOINT_MONITOR_HISTORY_SIZE
+ .defaultValue()
+ .intValue(),
+
config.getEngineConfig().getCheckpointConfig().getMonitorHistorySize());
+ Assertions.assertEquals(
+ 32,
config.getEngineConfig().getCheckpointConfig().getMonitorHistorySize());
+ }
+
+ @Test
+ public void testCheckpointMonitorHistorySizeIsBounded() {
+ CheckpointConfig checkpointConfig = new CheckpointConfig();
+ int max =
ServerConfigOptions.MasterServerConfigOptions.MAX_CHECKPOINT_MONITOR_HISTORY_SIZE;
+
+ // The bounds exist because the history lives in the cluster-wide
+ // checkpoint overview IMap, whose whole value is re-serialised on
every
+ // checkpoint event -- an oversized window would multiply master CPU
and
+ // replication traffic silently.
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
checkpointConfig.setMonitorHistorySize(0));
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
checkpointConfig.setMonitorHistorySize(-1));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> checkpointConfig.setMonitorHistorySize(max + 1));
+
+ checkpointConfig.setMonitorHistorySize(1);
+ Assertions.assertEquals(1, checkpointConfig.getMonitorHistorySize());
+ checkpointConfig.setMonitorHistorySize(max);
+ Assertions.assertEquals(max, checkpointConfig.getMonitorHistorySize());
}
@Test
diff --git
a/seatunnel-engine/seatunnel-engine-common/src/test/resources/seatunnel.yaml
b/seatunnel-engine/seatunnel-engine-common/src/test/resources/seatunnel.yaml
index 3775207470..9c2527da84 100644
--- a/seatunnel-engine/seatunnel-engine-common/src/test/resources/seatunnel.yaml
+++ b/seatunnel-engine/seatunnel-engine-common/src/test/resources/seatunnel.yaml
@@ -28,6 +28,7 @@ seatunnel:
checkpoint:
interval: 6000
timeout: 7000
+ monitor-history-size: 64
storage:
type: hdfs
max-retained: 3
diff --git
a/seatunnel-engine/seatunnel-engine-core/src/test/java/org/apache/seatunnel/engine/core/checkpoint/PipelineCheckpointOverviewTest.java
b/seatunnel-engine/seatunnel-engine-core/src/test/java/org/apache/seatunnel/engine/core/checkpoint/PipelineCheckpointOverviewTest.java
new file mode 100644
index 0000000000..2cc8a2a2f2
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-core/src/test/java/org/apache/seatunnel/engine/core/checkpoint/PipelineCheckpointOverviewTest.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.engine.core.checkpoint;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Tests the retained checkpoint history window, which is what {@code
+ * seatunnel.engine.checkpoint.monitor-history-size} configures.
+ */
+public class PipelineCheckpointOverviewTest {
+
+ private static CheckpointHistoryEntry entry(long checkpointId) {
+ CheckpointInfo info = new CheckpointInfo();
+ info.setCheckpointId(checkpointId);
+ CheckpointHistoryEntry historyEntry = new CheckpointHistoryEntry();
+ historyEntry.setCheckpointInfo(info);
+ return historyEntry;
+ }
+
+ @Test
+ public void testHistoryIsTrimmedToTheGivenMaximum() {
+ PipelineCheckpointOverview overview = new PipelineCheckpointOverview();
+
+ for (long id = 1; id <= 10; id++) {
+ overview.addHistory(entry(id), 3);
+ }
+
+ List<CheckpointHistoryEntry> history = new
ArrayList<>(overview.getHistory());
+ Assertions.assertEquals(3, history.size());
+
+ // Newest first: addHistory prepends, so the retained window is the
most
+ // recent entries, not the oldest. Anything consuming this as a trend
+ // depends on that order.
+ Assertions.assertEquals(10L,
history.get(0).getCheckpointInfo().getCheckpointId());
+ Assertions.assertEquals(9L,
history.get(1).getCheckpointInfo().getCheckpointId());
+ Assertions.assertEquals(8L,
history.get(2).getCheckpointInfo().getCheckpointId());
+ }
+
+ @Test
+ public void testHistoryHonoursAWindowLargerThanTheEntryCount() {
+ PipelineCheckpointOverview overview = new PipelineCheckpointOverview();
+
+ for (long id = 1; id <= 4; id++) {
+ overview.addHistory(entry(id), 64);
+ }
+
+ Assertions.assertEquals(4, overview.getHistory().size());
+ }
+
+ @Test
+ public void testAWindowOfOneKeepsOnlyTheNewestEntry() {
+ PipelineCheckpointOverview overview = new PipelineCheckpointOverview();
+
+ overview.addHistory(entry(1), 1);
+ overview.addHistory(entry(2), 1);
+
+ Assertions.assertEquals(1, overview.getHistory().size());
+ Assertions.assertEquals(
+ 2L,
overview.getHistory().peekFirst().getCheckpointInfo().getCheckpointId());
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/SeaTunnelServer.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/SeaTunnelServer.java
index b226e13b7a..90ee8642e7 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/SeaTunnelServer.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/SeaTunnelServer.java
@@ -22,6 +22,7 @@ import org.apache.seatunnel.common.utils.RetryUtils;
import org.apache.seatunnel.engine.common.Constant;
import org.apache.seatunnel.engine.common.config.EngineConfig;
import org.apache.seatunnel.engine.common.config.SeaTunnelConfig;
+import org.apache.seatunnel.engine.common.config.server.ServerConfigOptions;
import org.apache.seatunnel.engine.common.exception.SeaTunnelEngineException;
import
org.apache.seatunnel.engine.common.exception.SeaTunnelEngineRetryableException;
import org.apache.seatunnel.engine.core.classloader.ClassLoaderService;
@@ -215,7 +216,27 @@ public class SeaTunnelServer
private void startMaster() {
checkpointService =
new
CheckpointService(seaTunnelConfig.getEngineConfig().getCheckpointConfig());
- checkpointMonitorService = new CheckpointMonitorService(engineContext,
32);
+ int monitorHistorySize =
+
seaTunnelConfig.getEngineConfig().getCheckpointConfig().getMonitorHistorySize();
+ if (monitorHistorySize
+ > ServerConfigOptions.MasterServerConfigOptions
+ .CHECKPOINT_MONITOR_HISTORY_SIZE_WARN_THRESHOLD) {
+ // Legal but worth saying out loud: every checkpoint event rewrites
+ // the whole checkpoint overview entry in the IMap, so a large
+ // window is paid on every trigger and acknowledgement rather than
+ // once. Logged here, at the only place the value is read, so the
+ // line appears in the master log next to the rest of startup.
+ log.warn(
+ "seatunnel.engine.checkpoint.monitor-history-size is set
to {}, above the "
+ + "recommended {}. The checkpoint overview is
rewritten in full on "
+ + "every checkpoint event, so a window this large
costs master CPU, "
+ + "network and backup traffic proportional to
history size x running "
+ + "pipelines. Lower it unless you need that much
history.",
+ monitorHistorySize,
+ ServerConfigOptions.MasterServerConfigOptions
+ .CHECKPOINT_MONITOR_HISTORY_SIZE_WARN_THRESHOLD);
+ }
+ checkpointMonitorService = new CheckpointMonitorService(engineContext,
monitorHistorySize);
monitorService = Executors.newSingleThreadScheduledExecutor();
coordinatorService =
new CoordinatorService(
