This is an automated email from the ASF dual-hosted git repository.
zhangshenghang 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 25ad75cd65 [Improve][Zeta] Exclude runtime metrics snapshots from IMAP
persistence (#11244)
25ad75cd65 is described below
commit 25ad75cd65cea92f936c03bb77b946099ad8732a
Author: Doyeon Kim <[email protected]>
AuthorDate: Wed Jul 8 10:27:48 2026 +0900
[Improve][Zeta] Exclude runtime metrics snapshots from IMAP persistence
(#11244)
---
docs/en/engines/zeta/hybrid-cluster-deployment.md | 5 ++
.../engines/zeta/separated-cluster-deployment.md | 5 ++
docs/zh/engines/zeta/hybrid-cluster-deployment.md | 5 ++
.../engines/zeta/separated-cluster-deployment.md | 5 ++
.../engine/server/persistence/FileMapStore.java | 54 ++++++++++++++++++++++
.../server/persistence/FileMapStoreTest.java | 21 +++++++++
6 files changed, 95 insertions(+)
diff --git a/docs/en/engines/zeta/hybrid-cluster-deployment.md
b/docs/en/engines/zeta/hybrid-cluster-deployment.md
index 1c23bbee60..16aa323c15 100644
--- a/docs/en/engines/zeta/hybrid-cluster-deployment.md
+++ b/docs/en/engines/zeta/hybrid-cluster-deployment.md
@@ -319,6 +319,11 @@ map:
fs.defaultFS: file:///
```
+Note: `engine_runningJobMetrics` stores high-frequency runtime metrics
snapshots and is
+intentionally excluded from persistent IMAP storage even when `map.engine*`
uses `map-store`. This
+avoids excessive WAL growth for observability-only state. After an engine
restart, running-job
+metrics are rebuilt from subsequent reports instead of continuing from the
pre-restart snapshot.
+
If using OSS, you can configure it as follows:
```yaml
diff --git a/docs/en/engines/zeta/separated-cluster-deployment.md
b/docs/en/engines/zeta/separated-cluster-deployment.md
index 8cf659363d..02c07441c2 100644
--- a/docs/en/engines/zeta/separated-cluster-deployment.md
+++ b/docs/en/engines/zeta/separated-cluster-deployment.md
@@ -280,6 +280,11 @@ map:
fs.defaultFS: file:///
```
+Note: `engine_runningJobMetrics` stores high-frequency runtime metrics
snapshots and is
+intentionally excluded from persistent IMAP storage even when `map.engine*`
uses `map-store`. This
+avoids excessive WAL growth for observability-only state. After an engine
restart, running-job
+metrics are rebuilt from subsequent reports instead of continuing from the
pre-restart snapshot.
+
If you use OSS, you can configure it like this:
```yaml
diff --git a/docs/zh/engines/zeta/hybrid-cluster-deployment.md
b/docs/zh/engines/zeta/hybrid-cluster-deployment.md
index 991278d985..eafa13115c 100644
--- a/docs/zh/engines/zeta/hybrid-cluster-deployment.md
+++ b/docs/zh/engines/zeta/hybrid-cluster-deployment.md
@@ -315,6 +315,11 @@ map:
fs.defaultFS: file:///
```
+说明：`engine_runningJobMetrics` 保存的是高频运行时指标快照。即使通过 `map.engine*`
+配置了 `map-store`，它也会被有意排除在持久化 IMAP 存储之外，以避免仅用于可观测性的状态导致
+WAL 持续膨胀。Engine 重启后，running-job metrics 不会延续重启前的 snapshot，而是由后续
+report 重新构建。
+
如果您使用 OSS，可以像这样配置：
```yaml
diff --git a/docs/zh/engines/zeta/separated-cluster-deployment.md
b/docs/zh/engines/zeta/separated-cluster-deployment.md
index 7ad82b0d93..b9cdfbeb66 100644
--- a/docs/zh/engines/zeta/separated-cluster-deployment.md
+++ b/docs/zh/engines/zeta/separated-cluster-deployment.md
@@ -285,6 +285,11 @@ map:
fs.defaultFS: file:///
```
+说明：`engine_runningJobMetrics` 保存的是高频运行时指标快照。即使通过 `map.engine*`
+配置了 `map-store`，它也会被有意排除在持久化 IMAP 存储之外，以避免仅用于可观测性的状态导致
+WAL 持续膨胀。Engine 重启后，running-job metrics 不会延续重启前的 snapshot，而是由后续
+report 重新构建。
+
如果您使用 OSS，可以像这样配置：
```yaml
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/persistence/FileMapStore.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/persistence/FileMapStore.java
index 2900430440..6230ff80c4 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/persistence/FileMapStore.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/persistence/FileMapStore.java
@@ -22,6 +22,7 @@ import
org.apache.seatunnel.shade.com.google.common.collect.Maps;
import org.apache.seatunnel.engine.common.utils.FactoryUtil;
import org.apache.seatunnel.engine.imap.storage.api.IMapStorage;
import org.apache.seatunnel.engine.imap.storage.api.IMapStorageFactory;
+import
org.apache.seatunnel.engine.server.common.statestore.EngineStateStoreNames;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.map.MapLoaderLifecycleSupport;
@@ -42,6 +43,15 @@ public class FileMapStore implements MapStore<Object,
Object>, MapLoaderLifecycl
@Override
public void init(HazelcastInstance hazelcastInstance, Properties
properties, String mapName) {
+ if (EngineStateStoreNames.RUNNING_JOB_METRICS.equals(mapName)) {
+ this.mapStorage = NoOpMapStorage.INSTANCE;
+ log.info(
+ "Skip persistence for map '{}' because runtime metrics
snapshots are auxiliary "
+ + "observability state and should not write to
persistent IMAP storage.",
+ mapName);
+ return;
+ }
+
Map<String, Object> initMap = new
HashMap<>(Maps.fromProperties(properties));
String storageType = (String) initMap.get("type");
try {
@@ -107,4 +117,48 @@ public class FileMapStore implements MapStore<Object,
Object>, MapLoaderLifecycl
public Iterable<Object> loadAllKeys() {
return mapStorage.loadAllKeys();
}
+
+ private static final class NoOpMapStorage implements IMapStorage {
+ private static final NoOpMapStorage INSTANCE = new NoOpMapStorage();
+
+ @Override
+ public void initialize(Map<String, Object> properties) {
+ // no-op
+ }
+
+ @Override
+ public boolean store(Object key, Object value) {
+ return true;
+ }
+
+ @Override
+ public java.util.Set<Object> storeAll(Map<Object, Object> map) {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public boolean delete(Object key) {
+ return true;
+ }
+
+ @Override
+ public java.util.Set<Object> deleteAll(Collection<Object> keys) {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Map<Object, Object> loadAll() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public java.util.Set<Object> loadAllKeys() {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public void destroy(boolean deleteAllFileFlag) {
+ // no-op
+ }
+ }
}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/persistence/FileMapStoreTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/persistence/FileMapStoreTest.java
index 3c6d9e1c61..a1067c4f58 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/persistence/FileMapStoreTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/persistence/FileMapStoreTest.java
@@ -18,6 +18,7 @@
package org.apache.seatunnel.engine.server.persistence;
import org.apache.seatunnel.engine.imap.storage.file.common.FileConstants;
+import
org.apache.seatunnel.engine.server.common.statestore.EngineStateStoreNames;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -28,6 +29,7 @@ import org.junit.jupiter.api.io.TempDir;
import com.hazelcast.core.HazelcastInstance;
import java.nio.file.Path;
+import java.util.Collections;
import java.util.Properties;
import static org.mockito.Mockito.mock;
@@ -101,6 +103,25 @@ public class FileMapStoreTest {
"Exception message should identify the unknown type. Got: " +
msg);
}
+ @Test
+ public void testMetricsSnapshotMapSkipsPersistentStorageInitialization() {
+ FileMapStore store = new FileMapStore();
+ Properties props = new Properties();
+ props.setProperty("type", "non-existent-storage-type");
+
+ Assertions.assertDoesNotThrow(
+ () ->
+ store.init(
+ mock(HazelcastInstance.class),
+ props,
+ EngineStateStoreNames.RUNNING_JOB_METRICS));
+ Assertions.assertEquals(Collections.emptySet(), store.loadAllKeys());
+ Assertions.assertDoesNotThrow(() -> store.store("k", "v"));
+ Assertions.assertDoesNotThrow(() -> store.delete("k"));
+
+ store.destroy();
+ }
+
private Properties buildLocalFsProperties(String namespace) {
Properties props = new Properties();
// "hdfs" is the factoryIdentifier of IMapFileStorageFactory — covers
local fs via
