This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12789-fb8e7a7eeaa8ece1a4fc169619a8cc7b2c3f69f2 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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(
