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 d774e8ee8c [Feature][Zeta] Decouple Hazelcast IMap via StateStore
Abstraction (#10812)
d774e8ee8c is described below
commit d774e8ee8cea6148a7f95562a411c9f770c70aa0
Author: Doyeon Kim <[email protected]>
AuthorDate: Sun Jun 21 23:12:28 2026 +0900
[Feature][Zeta] Decouple Hazelcast IMap via StateStore Abstraction (#10812)
---
.../apache/seatunnel/engine/common/Constant.java | 4 -
.../engine/server/CoordinatorService.java | 169 +----------
.../seatunnel/engine/server/SeaTunnelServer.java | 98 ++-----
.../engine/server/TaskExecutionService.java | 9 +-
.../monitor/CheckpointMonitorService.java | 213 +-------------
.../server/common/SeaTunnelEngineContext.java | 69 +++++
.../statestore/AuthoritativeStateStores.java | 38 +++
.../common/statestore/AuxiliaryStateStores.java | 54 ++++
.../DefaultAuthoritativeStateStores.java | 21 ++
.../statestore/DefaultAuxiliaryStateStores.java | 66 +++++
.../common/statestore/EngineStateStoreNames.java | 31 ++
.../common/statestore/EngineStateStores.java | 75 +++++
.../common/statestore/IterableStateStore.java | 57 ++++
.../server/common/statestore/StateStore.java | 100 +++++++
.../checkpoint/CheckpointOverviewStateStore.java | 67 +++++
.../HazelcastCheckpointOverviewStateStore.java | 282 ++++++++++++++++++
.../hazelcast/HazelcastEngineStateStores.java | 100 +++++++
.../metrics/MetricsSnapshotStateStore.java | 113 ++++++++
.../HazelcastMetricsSnapshotStateStore.java | 322 +++++++++++++++++++++
.../server/execution/TaskExecutionContext.java | 29 +-
.../exports/EngineStateStoreMetricExports.java | 5 +-
.../CoordinatorServicePipelineCleanupTest.java | 12 +-
.../engine/server/CoordinatorServiceTest.java | 45 +--
.../HazelcastCheckpointOverviewStateStoreTest.java | 133 +++++++++
.../HazelcastMetricsSnapshotStateStoreTest.java | 203 +++++++++++++
.../EngineStateStoreLogicalMetricExportsTest.java | 3 +-
.../exports/EngineStateStoreMetricExportsTest.java | 5 +-
27 files changed, 1838 insertions(+), 485 deletions(-)
diff --git
a/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/Constant.java
b/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/Constant.java
index ab5e522fdf..582c7f0cc6 100644
---
a/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/Constant.java
+++
b/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/Constant.java
@@ -57,14 +57,10 @@ public class Constant {
public static final String IMAP_CHECKPOINT_ID = "engine_checkpoint-id-map";
- public static final String IMAP_RUNNING_JOB_METRICS =
"engine_runningJobMetrics";
-
public static final String IMAP_PENDING_PIPELINE_CLEANUP =
"engine_pendingPipelineCleanup";
public static final String IMAP_PENDING_JOB_CLEANUP =
"engine_pendingJobCleanup";
- public static final String IMAP_CHECKPOINT_MONITOR =
"engine_checkpoint_monitor";
-
public static final String IMAP_CONNECTOR_JAR_REF_COUNTERS =
"engine_connectorJarRefCounters";
public static final String PROP_FILE = "zeta.version.properties";
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/CoordinatorService.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/CoordinatorService.java
index 6d1453b91b..4cd0c92475 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/CoordinatorService.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/CoordinatorService.java
@@ -51,6 +51,8 @@ import org.apache.seatunnel.engine.core.job.JobDAGInfo;
import org.apache.seatunnel.engine.core.job.JobImmutableInformation;
import org.apache.seatunnel.engine.core.job.JobInfo;
import org.apache.seatunnel.engine.core.job.PipelineStatus;
+import org.apache.seatunnel.engine.server.common.SeaTunnelEngineContext;
+import
org.apache.seatunnel.engine.server.common.statestore.metrics.MetricsSnapshotStateStore;
import org.apache.seatunnel.engine.server.dag.DAGUtils;
import org.apache.seatunnel.engine.server.dag.physical.PhysicalVertex;
import org.apache.seatunnel.engine.server.dag.physical.PipelineLocation;
@@ -67,13 +69,11 @@ import
org.apache.seatunnel.engine.server.execution.PendingJobInfo;
import org.apache.seatunnel.engine.server.execution.PendingSourceState;
import org.apache.seatunnel.engine.server.execution.TaskExecutionState;
import org.apache.seatunnel.engine.server.execution.TaskGroupLocation;
-import org.apache.seatunnel.engine.server.execution.TaskLocation;
import org.apache.seatunnel.engine.server.master.JobHistoryService;
import org.apache.seatunnel.engine.server.master.JobMaster;
import org.apache.seatunnel.engine.server.master.cleanup.JobCleanupRecord;
import org.apache.seatunnel.engine.server.master.cleanup.PipelineCleanupRecord;
import org.apache.seatunnel.engine.server.metrics.JobMetricsUtil;
-import org.apache.seatunnel.engine.server.metrics.SeaTunnelMetricsContext;
import
org.apache.seatunnel.engine.server.resourcemanager.NoEnoughResourceException;
import org.apache.seatunnel.engine.server.resourcemanager.ResourceManager;
import
org.apache.seatunnel.engine.server.resourcemanager.ResourceManagerFactory;
@@ -88,15 +88,11 @@ import
org.apache.seatunnel.engine.server.utils.PeekBlockingQueue;
import com.hazelcast.cluster.Address;
import com.hazelcast.config.Config;
-import com.hazelcast.core.EntryEvent;
import com.hazelcast.core.HazelcastInstanceNotActiveException;
import com.hazelcast.internal.serialization.Data;
import com.hazelcast.internal.services.MembershipServiceEvent;
import com.hazelcast.logging.ILogger;
import com.hazelcast.map.IMap;
-import com.hazelcast.map.listener.EntryAddedListener;
-import com.hazelcast.map.listener.EntryRemovedListener;
-import com.hazelcast.map.listener.EntryUpdatedListener;
import com.hazelcast.ringbuffer.Ringbuffer;
import com.hazelcast.spi.impl.NodeEngineImpl;
import lombok.NonNull;
@@ -113,7 +109,6 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
-import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -135,6 +130,7 @@ import static
org.apache.seatunnel.engine.server.metrics.JobMetricsUtil.toJobMet
public class CoordinatorService {
private static final int PIPELINE_CLEANUP_INTERVAL_SECONDS = 60;
private final NodeEngineImpl nodeEngine;
+ private final SeaTunnelEngineContext engineContext;
private final ILogger logger;
private volatile ResourceManager resourceManager;
@@ -197,8 +193,6 @@ public class CoordinatorService {
*/
private IMap<PipelineLocation, Map<TaskGroupLocation, SlotProfile>>
ownedSlotProfilesIMap;
- private IMap<Long, HashMap<TaskLocation, SeaTunnelMetricsContext>>
metricsImap;
-
private IMap<PipelineLocation, PipelineCleanupRecord>
pendingPipelineCleanupIMap;
private IMap<Long, JobCleanupRecord> pendingJobCleanupIMap;
@@ -229,25 +223,13 @@ public class CoordinatorService {
private final AtomicBoolean coordinatorServiceCleared = new
AtomicBoolean(false);
- private final AtomicLong runningJobMetricsPartitionKeyCount = new
AtomicLong();
-
- private final AtomicLong runningJobMetricsTaskContextCount = new
AtomicLong();
-
- private final Object runningJobMetricsStatsLock = new Object();
-
- private final Set<Long> runningJobMetricsDirtyKeys =
ConcurrentHashMap.newKeySet();
-
- private final Map<Long, RunningJobMetricsStats>
runningJobMetricsStatsByJobId = new HashMap<>();
-
- private final AtomicBoolean runningJobMetricsInitializing = new
AtomicBoolean(false);
-
- private volatile UUID runningJobMetricsListenerId;
-
public CoordinatorService(
@NonNull NodeEngineImpl nodeEngine,
@NonNull SeaTunnelServer seaTunnelServer,
+ @NonNull SeaTunnelEngineContext engineContext,
EngineConfig engineConfig) {
this.nodeEngine = nodeEngine;
+ this.engineContext = engineContext;
this.engineConfig = engineConfig;
this.logger = nodeEngine.getLogger(getClass());
this.executorService = createCoordinatorExecutor();
@@ -538,8 +520,6 @@ public class CoordinatorService {
nodeEngine.getHazelcastInstance().getMap(Constant.IMAP_STATE_TIMESTAMPS);
ownedSlotProfilesIMap =
nodeEngine.getHazelcastInstance().getMap(Constant.IMAP_OWNED_SLOT_PROFILES);
- metricsImap =
nodeEngine.getHazelcastInstance().getMap(Constant.IMAP_RUNNING_JOB_METRICS);
- initRunningJobMetricsStoreStats();
pendingPipelineCleanupIMap =
nodeEngine.getHazelcastInstance().getMap(Constant.IMAP_PENDING_PIPELINE_CLEANUP);
pendingJobCleanupIMap =
@@ -1124,7 +1104,6 @@ public class CoordinatorService {
if (!coordinatorServiceCleared.compareAndSet(false, true)) {
return;
}
- removeRunningJobMetricsListener();
// interrupt all JobMaster
runningJobMasterMap.values().forEach(JobMaster::interrupt);
if (isWaitStrategy) {
@@ -1834,7 +1813,6 @@ public class CoordinatorService {
if (pipelineCleanupScheduler != null) {
pipelineCleanupScheduler.shutdown();
}
- removeRunningJobMetricsListener();
clearCoordinatorService();
awaitSchedulerTermination("master active listener",
masterActiveListener);
awaitSchedulerTermination("pipeline cleanup scheduler",
pipelineCleanupScheduler);
@@ -2097,11 +2075,11 @@ public class CoordinatorService {
}
public long getRunningJobMetricsPartitionKeyCount() {
- return runningJobMetricsPartitionKeyCount.get();
+ return getMetricsSnapshotStateStore().activePartitionKeyCount();
}
public long getRunningJobMetricsTaskContextCount() {
- return runningJobMetricsTaskContextCount.get();
+ return getMetricsSnapshotStateStore().size();
}
public EngineConfig getEngineConfig() {
@@ -2109,8 +2087,8 @@ public class CoordinatorService {
}
@VisibleForTesting
- protected IMap<Long, HashMap<TaskLocation, SeaTunnelMetricsContext>>
getMetricsImap() {
- return metricsImap;
+ protected MetricsSnapshotStateStore getMetricsSnapshotStateStore() {
+ return engineContext.getStateStores().metricsSnapshotStore();
}
@VisibleForTesting
@@ -2128,135 +2106,6 @@ public class CoordinatorService {
}
}
- private void initRunningJobMetricsStoreStats() {
- removeRunningJobMetricsListener();
- runningJobMetricsListenerId =
- metricsImap.addEntryListener(new
RunningJobMetricsEntryListener(), true);
- runningJobMetricsInitializing.set(true);
- runningJobMetricsDirtyKeys.clear();
-
- Map<Long, RunningJobMetricsStats> snapshotStats = new HashMap<>();
- metricsImap.forEach(
- (partitionKey, metrics) ->
- snapshotStats.put(partitionKey,
toRunningJobMetricsStats(metrics)));
-
- Set<Long> dirtyKeys = new HashSet<>(runningJobMetricsDirtyKeys);
- for (Long partitionKey : dirtyKeys) {
- snapshotStats.put(
- partitionKey,
toRunningJobMetricsStats(metricsImap.get(partitionKey)));
- }
-
- synchronized (runningJobMetricsStatsLock) {
- runningJobMetricsStatsByJobId.clear();
- runningJobMetricsPartitionKeyCount.set(0L);
- runningJobMetricsTaskContextCount.set(0L);
- snapshotStats.forEach(this::replaceRunningJobMetricsStatsLocked);
- runningJobMetricsInitializing.set(false);
-
- Set<Long> postSnapshotDirtyKeys = new
HashSet<>(runningJobMetricsDirtyKeys);
- runningJobMetricsDirtyKeys.clear();
- for (Long partitionKey : postSnapshotDirtyKeys) {
- replaceRunningJobMetricsStatsLocked(
- partitionKey,
toRunningJobMetricsStats(metricsImap.get(partitionKey)));
- }
- }
- }
-
- private RunningJobMetricsStats toRunningJobMetricsStats(
- Map<TaskLocation, SeaTunnelMetricsContext> metrics) {
- if (metrics == null || metrics.isEmpty()) {
- return RunningJobMetricsStats.EMPTY;
- }
- return new RunningJobMetricsStats(1L, metrics.size());
- }
-
- private void replaceRunningJobMetricsStatsLocked(
- Long partitionKey, RunningJobMetricsStats stats) {
- RunningJobMetricsStats currentStats =
runningJobMetricsStatsByJobId.get(partitionKey);
- if (currentStats != null) {
-
runningJobMetricsPartitionKeyCount.addAndGet(-currentStats.partitionKeyCount);
-
runningJobMetricsTaskContextCount.addAndGet(-currentStats.taskContextCount);
- }
-
- if (stats.isEmpty()) {
- runningJobMetricsStatsByJobId.remove(partitionKey);
- return;
- }
-
- runningJobMetricsStatsByJobId.put(partitionKey, stats);
- runningJobMetricsPartitionKeyCount.addAndGet(stats.partitionKeyCount);
- runningJobMetricsTaskContextCount.addAndGet(stats.taskContextCount);
- }
-
- private void removeRunningJobMetricsListener() {
- if (metricsImap != null && runningJobMetricsListenerId != null) {
- metricsImap.removeEntryListener(runningJobMetricsListenerId);
- runningJobMetricsListenerId = null;
- }
- runningJobMetricsInitializing.set(false);
- runningJobMetricsDirtyKeys.clear();
- synchronized (runningJobMetricsStatsLock) {
- runningJobMetricsStatsByJobId.clear();
- }
- runningJobMetricsPartitionKeyCount.set(0L);
- runningJobMetricsTaskContextCount.set(0L);
- }
-
- private final class RunningJobMetricsEntryListener
- implements EntryAddedListener<Long, HashMap<TaskLocation,
SeaTunnelMetricsContext>>,
- EntryUpdatedListener<Long, HashMap<TaskLocation,
SeaTunnelMetricsContext>>,
- EntryRemovedListener<Long, HashMap<TaskLocation,
SeaTunnelMetricsContext>> {
-
- @Override
- public void entryAdded(
- EntryEvent<Long, HashMap<TaskLocation,
SeaTunnelMetricsContext>> event) {
- replaceRunningJobMetricsStats(event.getKey(), event.getValue());
- }
-
- @Override
- public void entryUpdated(
- EntryEvent<Long, HashMap<TaskLocation,
SeaTunnelMetricsContext>> event) {
- replaceRunningJobMetricsStats(event.getKey(), event.getValue());
- }
-
- @Override
- public void entryRemoved(
- EntryEvent<Long, HashMap<TaskLocation,
SeaTunnelMetricsContext>> event) {
- replaceRunningJobMetricsStats(event.getKey(), null);
- }
- }
-
- private void replaceRunningJobMetricsStats(
- Long partitionKey, Map<TaskLocation, SeaTunnelMetricsContext>
metrics) {
- if (runningJobMetricsInitializing.get()) {
- runningJobMetricsDirtyKeys.add(partitionKey);
- return;
- }
- synchronized (runningJobMetricsStatsLock) {
- if (runningJobMetricsInitializing.get()) {
- runningJobMetricsDirtyKeys.add(partitionKey);
- return;
- }
- replaceRunningJobMetricsStatsLocked(partitionKey,
toRunningJobMetricsStats(metrics));
- }
- }
-
- private static final class RunningJobMetricsStats {
- private static final RunningJobMetricsStats EMPTY = new
RunningJobMetricsStats(0L, 0L);
-
- private final long partitionKeyCount;
- private final long taskContextCount;
-
- private RunningJobMetricsStats(long partitionKeyCount, long
taskContextCount) {
- this.partitionKeyCount = partitionKeyCount;
- this.taskContextCount = taskContextCount;
- }
-
- private boolean isEmpty() {
- return partitionKeyCount == 0L && taskContextCount == 0L;
- }
- }
-
@VisibleForTesting
public PeekBlockingQueue<PendingJobInfo> getPendingJobQueue() {
return pendingJobQueue;
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 c7a362c53f..37f39b03f8 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
@@ -27,6 +27,10 @@ import
org.apache.seatunnel.engine.common.exception.SeaTunnelEngineRetryableExce
import org.apache.seatunnel.engine.core.classloader.ClassLoaderService;
import org.apache.seatunnel.engine.core.classloader.DefaultClassLoaderService;
import
org.apache.seatunnel.engine.server.checkpoint.monitor.CheckpointMonitorService;
+import org.apache.seatunnel.engine.server.common.SeaTunnelEngineContext;
+import org.apache.seatunnel.engine.server.common.statestore.EngineStateStores;
+import
org.apache.seatunnel.engine.server.common.statestore.hazelcast.HazelcastEngineStateStores;
+import
org.apache.seatunnel.engine.server.common.statestore.metrics.MetricsSnapshotStateStore;
import org.apache.seatunnel.engine.server.dag.physical.PipelineLocation;
import org.apache.seatunnel.engine.server.execution.ExecutionState;
import org.apache.seatunnel.engine.server.execution.TaskGroupLocation;
@@ -58,14 +62,11 @@ import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import java.sql.DriverManager;
-import java.util.HashMap;
-import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
@Slf4j
public class SeaTunnelServer
@@ -87,6 +88,7 @@ public class SeaTunnelServer
public static final String SERVICE_NAME = "st:impl:seaTunnelServer";
+ @Getter private SeaTunnelEngineContext engineContext;
private NodeEngineImpl nodeEngine;
private final LiveOperationRegistry liveOperationRegistry;
@@ -146,6 +148,12 @@ public class SeaTunnelServer
// TODO Determine whether to execute there method on the master node
according to the deploy
// type
+ EngineStateStores stateStores =
+ new HazelcastEngineStateStores(
+ nodeEngine,
+
seaTunnelConfig.getEngineConfig().getJobMetricsPartitionCount());
+ this.engineContext =
SeaTunnelEngineContext.builder(stateStores).build();
+
classLoaderService =
new DefaultClassLoaderService(
seaTunnelConfig.getEngineConfig().isClassloaderCacheMode(), nodeEngine);
@@ -191,10 +199,11 @@ public class SeaTunnelServer
private void startMaster() {
checkpointService =
new
CheckpointService(seaTunnelConfig.getEngineConfig().getCheckpointConfig());
- checkpointMonitorService = new CheckpointMonitorService(nodeEngine,
32);
+ checkpointMonitorService = new CheckpointMonitorService(engineContext,
32);
monitorService = Executors.newSingleThreadScheduledExecutor();
coordinatorService =
- new CoordinatorService(nodeEngine, this,
seaTunnelConfig.getEngineConfig());
+ new CoordinatorService(
+ nodeEngine, this, engineContext,
seaTunnelConfig.getEngineConfig());
monitorService.scheduleAtFixedRate(
this::printExecutionInfo,
0,
@@ -204,7 +213,8 @@ public class SeaTunnelServer
private void startWorker() {
taskExecutionService =
- new TaskExecutionService(classLoaderService, nodeEngine,
eventService);
+ new TaskExecutionService(
+ classLoaderService, nodeEngine, engineContext,
eventService);
nodeEngine.getMetricsRegistry().registerDynamicMetricsProvider(taskExecutionService);
taskExecutionService.start();
getSlotService();
@@ -243,6 +253,7 @@ public class SeaTunnelServer
}
MetadataProviderManager.closeProviders();
+ engineContext.close();
}
@Override
@@ -376,78 +387,15 @@ public class SeaTunnelServer
}
public void updateMetrics(Map<TaskLocation, SeaTunnelMetricsContext>
localMap) {
- if (localMap == null || localMap.isEmpty()) {
- return;
- }
- int partitionCount =
seaTunnelConfig.getEngineConfig().getJobMetricsPartitionCount();
-
- IMap<Long, Map<TaskLocation, SeaTunnelMetricsContext>> metricsImap =
-
getNodeEngine().getHazelcastInstance().getMap(Constant.IMAP_RUNNING_JOB_METRICS);
-
- Map<Long, Map<TaskLocation, SeaTunnelMetricsContext>> partitioned =
new HashMap<>();
- localMap.forEach(
- (key, value) -> {
- long partition = getMetricsImapPartition(key,
partitionCount);
- partitioned.computeIfAbsent(partition, k -> new
HashMap<>()).put(key, value);
- });
-
- partitioned
- .entrySet()
- .parallelStream()
- .forEach(
- entry -> {
- metricsImap.compute(
- entry.getKey(),
- (k, oldVal) -> {
- if (oldVal == null) oldVal = new
HashMap<>();
- oldVal.putAll(entry.getValue());
- return oldVal;
- });
- });
+ MetricsSnapshotStateStore metricsSnapshotStateStore =
+ engineContext.getStateStores().metricsSnapshotStore();
+ metricsSnapshotStateStore.merge(localMap);
}
public void removeMetrics(PipelineLocation pipelineLocation) {
- IMap<Long, Map<TaskLocation, SeaTunnelMetricsContext>> metricsImap =
-
getNodeEngine().getHazelcastInstance().getMap(Constant.IMAP_RUNNING_JOB_METRICS);
-
- Map<Long, List<TaskLocation>> partitionedTasks = new HashMap<>();
- for (Map.Entry<Long, Map<TaskLocation, SeaTunnelMetricsContext>> entry
:
- metricsImap.entrySet()) {
- long partition = entry.getKey();
- List<TaskLocation> tasksToRemove =
- entry.getValue().keySet().stream()
- .filter(
- t ->
- t.getTaskGroupLocation()
- .getPipelineLocation()
- .equals(pipelineLocation))
- .collect(Collectors.toList());
- if (!tasksToRemove.isEmpty()) {
- partitionedTasks.put(partition, tasksToRemove);
- }
- }
-
- partitionedTasks
- .entrySet()
- .parallelStream()
- .forEach(
- entry -> {
- long partition = entry.getKey();
- List<TaskLocation> tasks = entry.getValue();
- metricsImap.compute(
- partition,
- (k, oldVal) -> {
- if (oldVal != null) {
- tasks.forEach(oldVal::remove);
- if (oldVal.isEmpty()) return null;
- }
- return oldVal;
- });
- });
- }
-
- public static long getMetricsImapPartition(TaskLocation key, int
partitionCount) {
- return (key.hashCode() & 0x7FFFFFFF) % partitionCount;
+ MetricsSnapshotStateStore metricsSnapshotStateStore =
+ engineContext.getStateStores().metricsSnapshotStore();
+ metricsSnapshotStateStore.removePipeline(pipelineLocation);
}
public boolean isCoordinatorActive() {
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/TaskExecutionService.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/TaskExecutionService.java
index 5e011d5391..b3bc49dd6d 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/TaskExecutionService.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/TaskExecutionService.java
@@ -33,6 +33,7 @@ import
org.apache.seatunnel.engine.common.utils.PassiveCompletableFuture;
import org.apache.seatunnel.engine.common.utils.concurrent.CompletableFuture;
import org.apache.seatunnel.engine.core.classloader.ClassLoaderService;
import org.apache.seatunnel.engine.core.job.ConnectorJarIdentifier;
+import org.apache.seatunnel.engine.server.common.SeaTunnelEngineContext;
import
org.apache.seatunnel.engine.server.exception.TaskGroupContextNotFoundException;
import org.apache.seatunnel.engine.server.execution.ExecutionState;
import org.apache.seatunnel.engine.server.execution.ProgressState;
@@ -147,6 +148,9 @@ public class TaskExecutionService implements
DynamicMetricsProvider {
/** The NodeEngine implementation for this Hazelcast node. */
private final NodeEngineImpl nodeEngine;
+ /** Shared engine context exposing state-store abstractions. */
+ private final SeaTunnelEngineContext engineContext;
+
/** Service for managing class loaders for connector jars. */
private final ClassLoaderService classLoaderService;
@@ -225,10 +229,12 @@ public class TaskExecutionService implements
DynamicMetricsProvider {
public TaskExecutionService(
ClassLoaderService classLoaderService,
NodeEngineImpl nodeEngine,
+ SeaTunnelEngineContext engineContext,
EventService eventService) {
seaTunnelConfig = ConfigProvider.locateAndGetSeaTunnelConfig();
this.hzInstanceName = nodeEngine.getHazelcastInstance().getName();
this.nodeEngine = nodeEngine;
+ this.engineContext = engineContext;
this.classLoaderService = classLoaderService;
this.logger =
nodeEngine.getLoggingService().getLogger(TaskExecutionService.class);
@@ -538,7 +544,8 @@ public class TaskExecutionService implements
DynamicMetricsProvider {
.peek(
task -> {
TaskExecutionContext
taskExecutionContext =
- new TaskExecutionContext(task,
nodeEngine, this);
+ new TaskExecutionContext(
+ task, nodeEngine,
engineContext, this);
task.setTaskExecutionContext(taskExecutionContext);
taskExecutionContextMap.put(
task.getTaskID(),
taskExecutionContext);
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/monitor/CheckpointMonitorService.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/monitor/CheckpointMonitorService.java
index 44ec00b857..eb5b92553b 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/monitor/CheckpointMonitorService.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/monitor/CheckpointMonitorService.java
@@ -19,7 +19,6 @@ package org.apache.seatunnel.engine.server.checkpoint.monitor;
import org.apache.seatunnel.shade.com.google.common.base.Strings;
-import org.apache.seatunnel.engine.common.Constant;
import org.apache.seatunnel.engine.core.checkpoint.CheckpointHistoryEntry;
import org.apache.seatunnel.engine.core.checkpoint.CheckpointInfo;
import org.apache.seatunnel.engine.core.checkpoint.CheckpointOverview;
@@ -31,61 +30,41 @@ import
org.apache.seatunnel.engine.server.checkpoint.CheckpointCloseReason;
import org.apache.seatunnel.engine.server.checkpoint.CompletedCheckpoint;
import org.apache.seatunnel.engine.server.checkpoint.SubtaskStatistics;
import org.apache.seatunnel.engine.server.checkpoint.TaskStatistics;
+import org.apache.seatunnel.engine.server.common.SeaTunnelEngineContext;
+import
org.apache.seatunnel.engine.server.common.statestore.checkpoint.CheckpointOverviewStateStore;
-import com.hazelcast.core.EntryEvent;
-import com.hazelcast.map.IMap;
-import com.hazelcast.map.listener.EntryAddedListener;
-import com.hazelcast.map.listener.EntryExpiredListener;
-import com.hazelcast.map.listener.EntryRemovedListener;
-import com.hazelcast.map.listener.EntryUpdatedListener;
-import com.hazelcast.spi.impl.NodeEngine;
import lombok.extern.slf4j.Slf4j;
import java.util.ArrayList;
import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
-import java.util.Map;
import java.util.Objects;
import java.util.Optional;
-import java.util.Set;
-import java.util.UUID;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import java.util.stream.Collectors;
@Slf4j
public class CheckpointMonitorService {
- private final NodeEngine nodeEngine;
- private volatile IMap<Long, CheckpointOverview> overviewMap;
+ private final SeaTunnelEngineContext engineContext;
private final int maxHistorySize;
- private final AtomicLong overviewJobCount = new AtomicLong();
- private final AtomicLong inProgressCheckpointCount = new AtomicLong();
- private final AtomicLong retainedHistoryCount = new AtomicLong();
- private final Object overviewStatsLock = new Object();
- private final Set<Long> overviewDirtyJobIds =
ConcurrentHashMap.newKeySet();
- private final Map<Long, CheckpointOverviewStats> overviewStatsByJobId =
new HashMap<>();
- private final AtomicBoolean overviewStatsInitializing = new
AtomicBoolean(false);
- private volatile UUID overviewListenerId;
- public CheckpointMonitorService(NodeEngine nodeEngine, int maxHistorySize)
{
- this.nodeEngine = nodeEngine;
+ private volatile CheckpointOverviewStateStore overviewMap;
+
+ public CheckpointMonitorService(SeaTunnelEngineContext engineContext, int
maxHistorySize) {
+ this.engineContext = engineContext;
this.maxHistorySize = maxHistorySize;
}
- private IMap<Long, CheckpointOverview> getOverviewMap() {
+ private CheckpointOverviewStateStore getOverviewMap() {
if (overviewMap == null) {
synchronized (this) {
if (overviewMap == null) {
overviewMap =
- nodeEngine
- .getHazelcastInstance()
- .getMap(Constant.IMAP_CHECKPOINT_MONITOR);
- initOverviewStats();
+ engineContext
+ .getStateStores()
+ .auxiliary()
+ .checkpointOverviewStateStore();
}
}
}
@@ -259,186 +238,26 @@ public class CheckpointMonitorService {
}
public long getOverviewJobCount() {
- return overviewJobCount.get();
+ return getOverviewMap().getOverviewJobCount();
}
public long getInProgressCheckpointCount() {
- return inProgressCheckpointCount.get();
+ return getOverviewMap().getInProgressCheckpointCount();
}
public long getRetainedHistoryCount() {
- return retainedHistoryCount.get();
+ return getOverviewMap().getRetainedHistoryCount();
}
private void updateOverview(
long jobId, int pipelineId, Consumer<PipelineCheckpointOverview>
consumer) {
- getOverviewMap()
- .compute(
- jobId,
- (id, overview) -> {
- CheckpointOverview snapshot =
- overview == null ? new
CheckpointOverview(jobId) : overview;
- PipelineCheckpointOverview pipeline =
- snapshot.getOrCreatePipeline(pipelineId);
- consumer.accept(pipeline);
- snapshot.setUpdatedAt(System.currentTimeMillis());
- return snapshot;
- });
+ getOverviewMap().updateOverview(jobId, pipelineId, consumer);
}
private void removeInProgressIfExists(PipelineCheckpointOverview pipeline,
long checkpointId) {
pipeline.getInProgress().removeIf(cp -> cp.getCheckpointId() ==
checkpointId);
}
- private void initOverviewStats() {
- removeOverviewListener();
- overviewListenerId =
- overviewMap.addEntryListener(new
CheckpointOverviewEntryListener(), true);
- overviewStatsInitializing.set(true);
- overviewDirtyJobIds.clear();
-
- Map<Long, CheckpointOverviewStats> snapshotStats = new HashMap<>();
- overviewMap.forEach(
- (jobId, overview) -> snapshotStats.put(jobId,
toOverviewStats(overview)));
-
- Set<Long> dirtyJobIds = new HashSet<>(overviewDirtyJobIds);
- for (Long jobId : dirtyJobIds) {
- snapshotStats.put(jobId, toOverviewStats(overviewMap.get(jobId)));
- }
-
- synchronized (overviewStatsLock) {
- overviewStatsByJobId.clear();
- overviewJobCount.set(0L);
- inProgressCheckpointCount.set(0L);
- retainedHistoryCount.set(0L);
- snapshotStats.forEach(this::replaceOverviewStatsLocked);
- overviewStatsInitializing.set(false);
-
- Set<Long> postSnapshotDirtyJobIds = new
HashSet<>(overviewDirtyJobIds);
- overviewDirtyJobIds.clear();
- for (Long jobId : postSnapshotDirtyJobIds) {
- replaceOverviewStatsLocked(jobId,
toOverviewStats(overviewMap.get(jobId)));
- }
- }
- }
-
- private void removeOverviewListener() {
- if (overviewMap != null && overviewListenerId != null) {
- overviewMap.removeEntryListener(overviewListenerId);
- overviewListenerId = null;
- }
- overviewStatsInitializing.set(false);
- overviewDirtyJobIds.clear();
- synchronized (overviewStatsLock) {
- overviewStatsByJobId.clear();
- }
- overviewJobCount.set(0L);
- inProgressCheckpointCount.set(0L);
- retainedHistoryCount.set(0L);
- }
-
- private CheckpointOverviewStats toOverviewStats(CheckpointOverview
overview) {
- if (overview == null) {
- return CheckpointOverviewStats.EMPTY;
- }
- return new CheckpointOverviewStats(
- 1L, getInProgressCount(overview), getHistoryCount(overview));
- }
-
- private void replaceOverviewStatsLocked(Long jobId,
CheckpointOverviewStats stats) {
- CheckpointOverviewStats currentStats = overviewStatsByJobId.get(jobId);
- if (currentStats != null) {
- overviewJobCount.addAndGet(-currentStats.jobCount);
-
inProgressCheckpointCount.addAndGet(-currentStats.inProgressCheckpointCount);
- retainedHistoryCount.addAndGet(-currentStats.retainedHistoryCount);
- }
-
- if (stats.isEmpty()) {
- overviewStatsByJobId.remove(jobId);
- return;
- }
-
- overviewStatsByJobId.put(jobId, stats);
- overviewJobCount.addAndGet(stats.jobCount);
- inProgressCheckpointCount.addAndGet(stats.inProgressCheckpointCount);
- retainedHistoryCount.addAndGet(stats.retainedHistoryCount);
- }
-
- private long getInProgressCount(CheckpointOverview overview) {
- return overview.getPipelines().values().stream()
- .filter(Objects::nonNull)
- .mapToLong(pipelineOverview ->
pipelineOverview.getInProgress().size())
- .sum();
- }
-
- private long getHistoryCount(CheckpointOverview overview) {
- return overview.getPipelines().values().stream()
- .filter(Objects::nonNull)
- .mapToLong(pipelineOverview ->
pipelineOverview.getHistory().size())
- .sum();
- }
-
- private final class CheckpointOverviewEntryListener
- implements EntryAddedListener<Long, CheckpointOverview>,
- EntryUpdatedListener<Long, CheckpointOverview>,
- EntryRemovedListener<Long, CheckpointOverview>,
- EntryExpiredListener<Long, CheckpointOverview> {
-
- @Override
- public void entryAdded(EntryEvent<Long, CheckpointOverview> event) {
- replaceOverviewStats(event.getKey(), event.getValue());
- }
-
- @Override
- public void entryUpdated(EntryEvent<Long, CheckpointOverview> event) {
- replaceOverviewStats(event.getKey(), event.getValue());
- }
-
- @Override
- public void entryRemoved(EntryEvent<Long, CheckpointOverview> event) {
- replaceOverviewStats(event.getKey(), null);
- }
-
- @Override
- public void entryExpired(EntryEvent<Long, CheckpointOverview> event) {
- replaceOverviewStats(event.getKey(), null);
- }
- }
-
- private void replaceOverviewStats(Long jobId, CheckpointOverview overview)
{
- if (overviewStatsInitializing.get()) {
- overviewDirtyJobIds.add(jobId);
- return;
- }
- synchronized (overviewStatsLock) {
- if (overviewStatsInitializing.get()) {
- overviewDirtyJobIds.add(jobId);
- return;
- }
- replaceOverviewStatsLocked(jobId, toOverviewStats(overview));
- }
- }
-
- private static final class CheckpointOverviewStats {
- private static final CheckpointOverviewStats EMPTY =
- new CheckpointOverviewStats(0L, 0L, 0L);
-
- private final long jobCount;
- private final long inProgressCheckpointCount;
- private final long retainedHistoryCount;
-
- private CheckpointOverviewStats(
- long jobCount, long inProgressCheckpointCount, long
retainedHistoryCount) {
- this.jobCount = jobCount;
- this.inProgressCheckpointCount = inProgressCheckpointCount;
- this.retainedHistoryCount = retainedHistoryCount;
- }
-
- private boolean isEmpty() {
- return jobCount == 0L && inProgressCheckpointCount == 0L &&
retainedHistoryCount == 0L;
- }
- }
-
public static long calculateStateSize(CompletedCheckpoint checkpoint) {
return checkpoint.getTaskStatistics().values().stream()
.map(TaskStatistics::getSubtaskStats)
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/SeaTunnelEngineContext.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/SeaTunnelEngineContext.java
new file mode 100644
index 0000000000..18128fcab4
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/SeaTunnelEngineContext.java
@@ -0,0 +1,69 @@
+/*
+ * 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.server.common;
+
+import org.apache.seatunnel.engine.server.common.statestore.EngineStateStores;
+
+import java.util.Objects;
+
+/**
+ * Shared engine context propagated inside the engine.
+ *
+ * <p>This context is intended to reduce direct propagation of
infrastructure-specific runtime
+ * objects into engine services. At the current stage, it only exposes state
store bundles and
+ * serves as a small entry point for state-related abstractions.
+ */
+public final class SeaTunnelEngineContext implements AutoCloseable {
+
+ private final EngineStateStores stateStores;
+
+ private SeaTunnelEngineContext(Builder builder) {
+ this.stateStores = builder.stateStores;
+ }
+
+ public static Builder builder(EngineStateStores stateStores) {
+ return new Builder(stateStores);
+ }
+
+ /**
+ * Returns the grouped engine state stores.
+ *
+ * @return engine state stores
+ */
+ public EngineStateStores getStateStores() {
+ return stateStores;
+ }
+
+ @Override
+ public void close() {
+ stateStores.close();
+ }
+
+ /** Builder for {@link SeaTunnelEngineContext}. */
+ public static final class Builder {
+ private final EngineStateStores stateStores;
+
+ private Builder(EngineStateStores stateStores) {
+ this.stateStores = Objects.requireNonNull(stateStores,
"stateStores");
+ }
+
+ public SeaTunnelEngineContext build() {
+ return new SeaTunnelEngineContext(this);
+ }
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/AuthoritativeStateStores.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/AuthoritativeStateStores.java
new file mode 100644
index 0000000000..1e14e769a9
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/AuthoritativeStateStores.java
@@ -0,0 +1,38 @@
+/*
+ * 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.server.common.statestore;
+
+/**
+ * Bundle of authoritative control state that a new leader must trust during
leader handoff.
+ *
+ * <p>This layer groups states that are likely to live on top of a consensus
layer. The actual
+ * storage backend may still be a local store, but the responsibility for
deciding what is
+ * authoritative belongs more strongly to this group.
+ */
+public interface AuthoritativeStateStores extends AutoCloseable {
+
+ /**
+ * Releases resources owned by authoritative stores.
+ *
+ * <p>Implementations that only wrap non-closeable stores may keep the
default no-op behavior.
+ */
+ @Override
+ default void close() {
+ // no-op
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/AuxiliaryStateStores.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/AuxiliaryStateStores.java
new file mode 100644
index 0000000000..af7c8bdae5
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/AuxiliaryStateStores.java
@@ -0,0 +1,54 @@
+/*
+ * 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.server.common.statestore;
+
+import
org.apache.seatunnel.engine.server.common.statestore.checkpoint.CheckpointOverviewStateStore;
+import
org.apache.seatunnel.engine.server.common.statestore.metrics.MetricsSnapshotStateStore;
+
+/**
+ * Bundle of state that is closer to observability, recent history, or cleanup
than to failover
+ * correctness.
+ *
+ * <p>This group collects states that can often tolerate more staleness, or
for which limited loss
+ * is less likely to break system correctness immediately.
+ */
+public interface AuxiliaryStateStores extends AutoCloseable {
+ /**
+ * Returns the store for runtime task metrics snapshots.
+ *
+ * @return metrics snapshot store
+ */
+ MetricsSnapshotStateStore metricsSnapshotStore();
+
+ /**
+ * Returns the store for checkpoint overviews.
+ *
+ * @return checkpoint overview state store
+ */
+ CheckpointOverviewStateStore checkpointOverviewStateStore();
+
+ /**
+ * Releases resources owned by auxiliary stores.
+ *
+ * <p>Implementations that only wrap non-closeable stores may keep the
default no-op behavior.
+ */
+ @Override
+ default void close() {
+ // no-op
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/DefaultAuthoritativeStateStores.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/DefaultAuthoritativeStateStores.java
new file mode 100644
index 0000000000..f658f31d39
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/DefaultAuthoritativeStateStores.java
@@ -0,0 +1,21 @@
+/*
+ * 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.server.common.statestore;
+
+/** Default immutable implementation of {@link AuthoritativeStateStores}. */
+public class DefaultAuthoritativeStateStores implements
AuthoritativeStateStores {}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/DefaultAuxiliaryStateStores.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/DefaultAuxiliaryStateStores.java
new file mode 100644
index 0000000000..87017e8f3e
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/DefaultAuxiliaryStateStores.java
@@ -0,0 +1,66 @@
+/*
+ * 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.server.common.statestore;
+
+import
org.apache.seatunnel.engine.server.common.statestore.checkpoint.CheckpointOverviewStateStore;
+import
org.apache.seatunnel.engine.server.common.statestore.metrics.MetricsSnapshotStateStore;
+
+import java.util.Objects;
+
+/** Default immutable implementation of {@link AuxiliaryStateStores}. */
+public class DefaultAuxiliaryStateStores implements AuxiliaryStateStores {
+ private final MetricsSnapshotStateStore metricsSnapshotStore;
+ private final CheckpointOverviewStateStore checkpointOverviewStateStore;
+
+ public DefaultAuxiliaryStateStores(
+ MetricsSnapshotStateStore metricsSnapshotStore,
+ CheckpointOverviewStateStore checkpointOverviewStateStore) {
+ this.metricsSnapshotStore =
+ Objects.requireNonNull(metricsSnapshotStore,
"metricsSnapshotStore");
+ this.checkpointOverviewStateStore =
+ Objects.requireNonNull(
+ checkpointOverviewStateStore,
"checkpointOverviewStateStore");
+ }
+
+ @Override
+ public MetricsSnapshotStateStore metricsSnapshotStore() {
+ return metricsSnapshotStore;
+ }
+
+ @Override
+ public CheckpointOverviewStateStore checkpointOverviewStateStore() {
+ return checkpointOverviewStateStore;
+ }
+
+ @Override
+ public void close() {
+ closeIfPossible(checkpointOverviewStateStore);
+ closeIfPossible(metricsSnapshotStore);
+ }
+
+ private void closeIfPossible(Object store) {
+ if (!(store instanceof AutoCloseable)) {
+ return;
+ }
+ try {
+ ((AutoCloseable) store).close();
+ } catch (Exception e) {
+ throw new IllegalStateException("Failed to close auxiliary state
store", e);
+ }
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/EngineStateStoreNames.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/EngineStateStoreNames.java
new file mode 100644
index 0000000000..3ed6dfd244
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/EngineStateStoreNames.java
@@ -0,0 +1,31 @@
+/*
+ * 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.server.common.statestore;
+
+import lombok.Getter;
+
+/**
+ * Bundle of map names required to build a Hazelcast-based {@link
EngineStateStores} implementation.
+ */
+@Getter
+public class EngineStateStoreNames {
+ private EngineStateStoreNames() {}
+
+ public static final String RUNNING_JOB_METRICS =
"engine_runningJobMetrics";
+ public static final String CHECKPOINT_MONITOR =
"engine_checkpoint_monitor";
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/EngineStateStores.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/EngineStateStores.java
new file mode 100644
index 0000000000..e86f206126
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/EngineStateStores.java
@@ -0,0 +1,75 @@
+/*
+ * 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.server.common.statestore;
+
+import
org.apache.seatunnel.engine.server.common.statestore.checkpoint.CheckpointOverviewStateStore;
+import
org.apache.seatunnel.engine.server.common.statestore.metrics.MetricsSnapshotStateStore;
+
+/**
+ * Top-level bundle of state stores used directly by the engine.
+ *
+ * <p>This is the high-level port that lets engine code depend on state
semantics rather than
+ * concrete implementations such as {@code HazelcastRuntimeStateStore}.
+ *
+ * <p>Implementation construction is expected to stay in
Hazelcast/RocksDB-specific providers, while
+ * engine code depends only on this interface.
+ */
+public interface EngineStateStores extends AutoCloseable {
+ /**
+ * Returns the bundle of control state that must be treated as
authoritative during leader
+ * handoff.
+ *
+ * @return authoritative state stores
+ */
+ AuthoritativeStateStores authoritative();
+
+ /**
+ * Returns the bundle of auxiliary state used mainly for observability,
recent history, or
+ * cleanup.
+ *
+ * @return auxiliary state stores
+ */
+ AuxiliaryStateStores auxiliary();
+
+ /**
+ * Returns the store for runtime task metrics snapshots.
+ *
+ * @return metrics snapshot store
+ */
+ default MetricsSnapshotStateStore metricsSnapshotStore() {
+ return auxiliary().metricsSnapshotStore();
+ }
+
+ /**
+ * Returns the store for checkpoint overviews.
+ *
+ * @return checkpoint overview state store
+ */
+ default CheckpointOverviewStateStore checkpointOverviewStateStore() {
+ return auxiliary().checkpointOverviewStateStore();
+ }
+
+ /**
+ * Releases resources owned by the stores.
+ *
+ * <p>Hazelcast implementations are usually no-op, while RocksDB
implementations use this to
+ * close the underlying database resources.
+ */
+ @Override
+ void close();
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/IterableStateStore.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/IterableStateStore.java
new file mode 100644
index 0000000000..0ead0ef613
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/IterableStateStore.java
@@ -0,0 +1,57 @@
+/*
+ * 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.server.common.statestore;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Backend storage capability for full iteration.
+ *
+ * <p>Operations that read all values can have very different costs depending
on the backend, so
+ * this capability is separated from the base {@link StateStore}.
+ *
+ * <p>This is better treated as a backend SPI used internally by higher-level
stores such as history
+ * or metrics stores, rather than as an engine-facing contract.
+ *
+ * @param <K> key type
+ * @param <V> value type
+ */
+public interface IterableStateStore<K, V> extends StateStore<K, V> {
+
+ /** @return all stored entries */
+ Set<Map.Entry<K, V>> entrySet();
+
+ /** @return all stored values */
+ Collection<V> values();
+
+ /**
+ * Returns whether the store is empty.
+ *
+ * @return {@code true} if the store is empty
+ */
+ boolean isEmpty();
+
+ /**
+ * Returns the total number of entries.
+ *
+ * @return entry count
+ */
+ int size();
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/StateStore.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/StateStore.java
new file mode 100644
index 0000000000..60c09deabb
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/StateStore.java
@@ -0,0 +1,100 @@
+/*
+ * 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.server.common.statestore;
+
+/**
+ * Low-level backend storage SPI.
+ *
+ * <p>This is not meant to be the public state contract used directly by
engine code. Instead, it is
+ * the minimal key-value storage contract used to implement real backends such
as RocksDB-backed
+ * stores.
+ *
+ * <p>Rather than copying the full Hazelcast {@code IMap} surface, it keeps
only the minimum
+ * operations that remain meaningful when switching between backend
implementations.
+ *
+ * @param <K> key type
+ * @param <V> value type
+ */
+public interface StateStore<K, V> {
+
+ /**
+ * Retrieves the value for the given key.
+ *
+ * @param key key to look up
+ * @return stored value, or {@code null} if absent
+ */
+ V get(K key);
+
+ /**
+ * Stores a value for the given key.
+ *
+ * @param key key to store
+ * @param value value to store
+ */
+ void put(K key, V value);
+
+ /**
+ * Stores a value only when the key is currently absent.
+ *
+ * @param key key to store
+ * @param value value to store
+ * @return existing value if present, otherwise {@code null}
+ */
+ V putIfAbsent(K key, V value);
+
+ /**
+ * Removes the value for the given key.
+ *
+ * @param key key to remove
+ */
+ void remove(K key);
+
+ /**
+ * Returns whether the key exists.
+ *
+ * @param key key to check
+ * @return {@code true} if the key exists
+ */
+ boolean containsKey(K key);
+
+ /**
+ * Returns whether the store is empty.
+ *
+ * @return {@code true} if the store is empty
+ */
+ boolean isEmpty();
+
+ /**
+ * Returns the number of key-value pairs in the store.
+ *
+ * @return size of the store
+ */
+ int size();
+
+ /**
+ * Returns the default value when the key is absent.
+ *
+ * @param key key to look up
+ * @param defaultValue default value
+ * @return stored value or the default value
+ */
+ default V getOrDefault(K key, V defaultValue) {
+ V value = get(key);
+ return value != null ? value : defaultValue;
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/checkpoint/CheckpointOverviewStateStore.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/checkpoint/CheckpointOverviewStateStore.java
new file mode 100644
index 0000000000..342024ccfe
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/checkpoint/CheckpointOverviewStateStore.java
@@ -0,0 +1,67 @@
+/*
+ * 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.server.common.statestore.checkpoint;
+
+import org.apache.seatunnel.engine.core.checkpoint.CheckpointOverview;
+import org.apache.seatunnel.engine.core.checkpoint.PipelineCheckpointOverview;
+import org.apache.seatunnel.engine.server.common.statestore.StateStore;
+
+import java.util.function.Consumer;
+
+/**
+ * Store for checkpoint monitoring overview state.
+ *
+ * <p>This store keeps checkpoint overview data used by monitoring and REST
queries. It is
+ * operational and observability-oriented state rather than core failover
state.
+ */
+public interface CheckpointOverviewStateStore extends StateStore<Long,
CheckpointOverview> {
+
+ /**
+ * Updates the overview for the given job and pipeline.
+ *
+ * <p>If no overview exists for the job, a new one is created. If no
pipeline overview exists,
+ * it is created before applying the updater.
+ *
+ * @param jobId job identifier
+ * @param pipelineId pipeline identifier
+ * @param updater pipeline-level updater
+ */
+ void updateOverview(long jobId, int pipelineId,
Consumer<PipelineCheckpointOverview> updater);
+
+ /**
+ * Returns the number of jobs currently tracked by the overview store.
+ *
+ * @return tracked job count
+ */
+ long getOverviewJobCount();
+
+ /**
+ * Returns the total number of in-progress checkpoints across all tracked
jobs and pipelines.
+ *
+ * @return in-progress checkpoint count
+ */
+ long getInProgressCheckpointCount();
+
+ /**
+ * Returns the total number of retained checkpoint history entries across
all tracked jobs and
+ * pipelines.
+ *
+ * @return retained checkpoint history entry count
+ */
+ long getRetainedHistoryCount();
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/checkpoint/hazelcast/HazelcastCheckpointOverviewStateStore.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/checkpoint/hazelcast/HazelcastCheckpointOverviewStateStore.java
new file mode 100644
index 0000000000..0a7b9482be
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/checkpoint/hazelcast/HazelcastCheckpointOverviewStateStore.java
@@ -0,0 +1,282 @@
+/*
+ * 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.server.common.statestore.checkpoint.hazelcast;
+
+import org.apache.seatunnel.engine.core.checkpoint.CheckpointOverview;
+import org.apache.seatunnel.engine.core.checkpoint.PipelineCheckpointOverview;
+import
org.apache.seatunnel.engine.server.common.statestore.checkpoint.CheckpointOverviewStateStore;
+
+import com.hazelcast.core.EntryEvent;
+import com.hazelcast.map.IMap;
+import com.hazelcast.map.listener.EntryAddedListener;
+import com.hazelcast.map.listener.EntryExpiredListener;
+import com.hazelcast.map.listener.EntryRemovedListener;
+import com.hazelcast.map.listener.EntryUpdatedListener;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Consumer;
+
+/** Hazelcast-backed implementation of {@link CheckpointOverviewStateStore}. */
+public class HazelcastCheckpointOverviewStateStore
+ implements CheckpointOverviewStateStore, AutoCloseable {
+
+ private final IMap<Long, CheckpointOverview> overviewMap;
+ private final AtomicLong overviewJobCount = new AtomicLong();
+ private final AtomicLong inProgressCheckpointCount = new AtomicLong();
+ private final AtomicLong retainedHistoryCount = new AtomicLong();
+ private final Object overviewStatsLock = new Object();
+ private final Set<Long> dirtyJobIds = ConcurrentHashMap.newKeySet();
+ private final Map<Long, CheckpointOverviewStats> statsByJobId = new
HashMap<>();
+ private final AtomicBoolean statsInitializing = new AtomicBoolean(false);
+ private volatile UUID overviewListenerId;
+
+ public HazelcastCheckpointOverviewStateStore(IMap<Long,
CheckpointOverview> overviewMap) {
+ this.overviewMap = Objects.requireNonNull(overviewMap, "overviewMap");
+ initStats();
+ }
+
+ @Override
+ public CheckpointOverview get(Long jobId) {
+ return overviewMap.get(jobId);
+ }
+
+ @Override
+ public void put(Long jobId, CheckpointOverview overview) {
+ overviewMap.put(jobId, overview);
+ }
+
+ @Override
+ public CheckpointOverview putIfAbsent(Long jobId, CheckpointOverview
overview) {
+ return overviewMap.putIfAbsent(jobId, overview);
+ }
+
+ @Override
+ public void remove(Long jobId) {
+ overviewMap.remove(jobId);
+ }
+
+ @Override
+ public boolean containsKey(Long jobId) {
+ return overviewMap.containsKey(jobId);
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return overviewJobCount.get() == 0L;
+ }
+
+ @Override
+ public int size() {
+ return Math.toIntExact(overviewJobCount.get());
+ }
+
+ @Override
+ public void updateOverview(
+ long jobId, int pipelineId, Consumer<PipelineCheckpointOverview>
updater) {
+ overviewMap.compute(
+ jobId,
+ (id, overview) -> {
+ CheckpointOverview snapshot =
+ overview == null ? new CheckpointOverview(jobId) :
overview;
+ PipelineCheckpointOverview pipeline =
snapshot.getOrCreatePipeline(pipelineId);
+ updater.accept(pipeline);
+ snapshot.setUpdatedAt(System.currentTimeMillis());
+ return snapshot;
+ });
+ }
+
+ @Override
+ public long getOverviewJobCount() {
+ return overviewJobCount.get();
+ }
+
+ @Override
+ public long getInProgressCheckpointCount() {
+ return inProgressCheckpointCount.get();
+ }
+
+ @Override
+ public long getRetainedHistoryCount() {
+ return retainedHistoryCount.get();
+ }
+
+ @Override
+ public void close() {
+ removeOverviewListener();
+ }
+
+ private void initStats() {
+ removeOverviewListener();
+ overviewListenerId =
+ overviewMap.addEntryListener(new
CheckpointOverviewEntryListener(), true);
+ statsInitializing.set(true);
+ dirtyJobIds.clear();
+
+ Map<Long, CheckpointOverviewStats> snapshotStats = new HashMap<>();
+ overviewMap.forEach(
+ (jobId, overview) -> snapshotStats.put(jobId,
toOverviewStats(overview)));
+
+ Set<Long> dirtyJobIdsDuringSnapshot = new HashSet<>(dirtyJobIds);
+ for (Long jobId : dirtyJobIdsDuringSnapshot) {
+ snapshotStats.put(jobId, toOverviewStats(overviewMap.get(jobId)));
+ }
+
+ synchronized (overviewStatsLock) {
+ statsByJobId.clear();
+ overviewJobCount.set(0L);
+ inProgressCheckpointCount.set(0L);
+ retainedHistoryCount.set(0L);
+ snapshotStats.forEach(this::replaceOverviewStatsLocked);
+ statsInitializing.set(false);
+
+ Set<Long> postSnapshotDirtyJobIds = new HashSet<>(dirtyJobIds);
+ dirtyJobIds.clear();
+ for (Long jobId : postSnapshotDirtyJobIds) {
+ replaceOverviewStatsLocked(jobId,
toOverviewStats(overviewMap.get(jobId)));
+ }
+ }
+ }
+
+ private void removeOverviewListener() {
+ if (overviewListenerId != null) {
+ overviewMap.removeEntryListener(overviewListenerId);
+ overviewListenerId = null;
+ }
+ statsInitializing.set(false);
+ dirtyJobIds.clear();
+ synchronized (overviewStatsLock) {
+ statsByJobId.clear();
+ }
+ overviewJobCount.set(0L);
+ inProgressCheckpointCount.set(0L);
+ retainedHistoryCount.set(0L);
+ }
+
+ private CheckpointOverviewStats toOverviewStats(CheckpointOverview
overview) {
+ if (overview == null) {
+ return CheckpointOverviewStats.EMPTY;
+ }
+ return new CheckpointOverviewStats(
+ 1L, getInProgressCount(overview), getHistoryCount(overview));
+ }
+
+ private void replaceOverviewStatsLocked(Long jobId,
CheckpointOverviewStats stats) {
+ CheckpointOverviewStats currentStats = statsByJobId.get(jobId);
+ if (currentStats != null) {
+ overviewJobCount.addAndGet(-currentStats.jobCount);
+
inProgressCheckpointCount.addAndGet(-currentStats.inProgressCheckpointCount);
+ retainedHistoryCount.addAndGet(-currentStats.retainedHistoryCount);
+ }
+
+ if (stats.isEmpty()) {
+ statsByJobId.remove(jobId);
+ return;
+ }
+
+ statsByJobId.put(jobId, stats);
+ overviewJobCount.addAndGet(stats.jobCount);
+ inProgressCheckpointCount.addAndGet(stats.inProgressCheckpointCount);
+ retainedHistoryCount.addAndGet(stats.retainedHistoryCount);
+ }
+
+ private long getInProgressCount(CheckpointOverview overview) {
+ return overview.getPipelines().values().stream()
+ .filter(Objects::nonNull)
+ .mapToLong(pipelineOverview ->
pipelineOverview.getInProgress().size())
+ .sum();
+ }
+
+ private long getHistoryCount(CheckpointOverview overview) {
+ return overview.getPipelines().values().stream()
+ .filter(Objects::nonNull)
+ .map(PipelineCheckpointOverview::getHistory)
+ .filter(Objects::nonNull)
+ .mapToLong(Collection::size)
+ .sum();
+ }
+
+ private void replaceOverviewStats(Long jobId, CheckpointOverview overview)
{
+ if (statsInitializing.get()) {
+ dirtyJobIds.add(jobId);
+ return;
+ }
+ synchronized (overviewStatsLock) {
+ if (statsInitializing.get()) {
+ dirtyJobIds.add(jobId);
+ return;
+ }
+ replaceOverviewStatsLocked(jobId, toOverviewStats(overview));
+ }
+ }
+
+ private final class CheckpointOverviewEntryListener
+ implements EntryAddedListener<Long, CheckpointOverview>,
+ EntryUpdatedListener<Long, CheckpointOverview>,
+ EntryRemovedListener<Long, CheckpointOverview>,
+ EntryExpiredListener<Long, CheckpointOverview> {
+
+ @Override
+ public void entryAdded(EntryEvent<Long, CheckpointOverview> event) {
+ replaceOverviewStats(event.getKey(), event.getValue());
+ }
+
+ @Override
+ public void entryUpdated(EntryEvent<Long, CheckpointOverview> event) {
+ replaceOverviewStats(event.getKey(), event.getValue());
+ }
+
+ @Override
+ public void entryRemoved(EntryEvent<Long, CheckpointOverview> event) {
+ replaceOverviewStats(event.getKey(), null);
+ }
+
+ @Override
+ public void entryExpired(EntryEvent<Long, CheckpointOverview> event) {
+ replaceOverviewStats(event.getKey(), null);
+ }
+ }
+
+ private static final class CheckpointOverviewStats {
+ private static final CheckpointOverviewStats EMPTY =
+ new CheckpointOverviewStats(0L, 0L, 0L);
+
+ private final long jobCount;
+ private final long inProgressCheckpointCount;
+ private final long retainedHistoryCount;
+
+ private CheckpointOverviewStats(
+ long jobCount, long inProgressCheckpointCount, long
retainedHistoryCount) {
+ this.jobCount = jobCount;
+ this.inProgressCheckpointCount = inProgressCheckpointCount;
+ this.retainedHistoryCount = retainedHistoryCount;
+ }
+
+ private boolean isEmpty() {
+ return jobCount == 0L && inProgressCheckpointCount == 0L &&
retainedHistoryCount == 0L;
+ }
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/hazelcast/HazelcastEngineStateStores.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/hazelcast/HazelcastEngineStateStores.java
new file mode 100644
index 0000000000..6e5a3a9bde
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/hazelcast/HazelcastEngineStateStores.java
@@ -0,0 +1,100 @@
+/*
+ * 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.server.common.statestore.hazelcast;
+
+import
org.apache.seatunnel.engine.server.common.statestore.AuthoritativeStateStores;
+import
org.apache.seatunnel.engine.server.common.statestore.AuxiliaryStateStores;
+import
org.apache.seatunnel.engine.server.common.statestore.DefaultAuthoritativeStateStores;
+import
org.apache.seatunnel.engine.server.common.statestore.DefaultAuxiliaryStateStores;
+import org.apache.seatunnel.engine.server.common.statestore.EngineStateStores;
+import
org.apache.seatunnel.engine.server.common.statestore.checkpoint.CheckpointOverviewStateStore;
+import
org.apache.seatunnel.engine.server.common.statestore.checkpoint.hazelcast.HazelcastCheckpointOverviewStateStore;
+import
org.apache.seatunnel.engine.server.common.statestore.metrics.MetricsSnapshotStateStore;
+import
org.apache.seatunnel.engine.server.common.statestore.metrics.hazelcast.HazelcastMetricsSnapshotStateStore;
+
+import com.hazelcast.spi.impl.NodeEngine;
+
+import java.util.Objects;
+
+import static
org.apache.seatunnel.engine.server.common.statestore.EngineStateStoreNames.CHECKPOINT_MONITOR;
+import static
org.apache.seatunnel.engine.server.common.statestore.EngineStateStoreNames.RUNNING_JOB_METRICS;
+
+/**
+ * {@link EngineStateStores} implementation backed by Hazelcast.
+ *
+ * <p>Engine code is not expected to reference this implementation directly.
It is intended to be
+ * created only during bootstrap and injected through interfaces.
+ */
+public class HazelcastEngineStateStores implements EngineStateStores {
+
+ private final NodeEngine nodeEngine;
+ private final int metricsPartitionCount;
+ private volatile AuthoritativeStateStores authoritativeStateStores;
+ private volatile AuxiliaryStateStores auxiliaryStateStores;
+
+ public HazelcastEngineStateStores(NodeEngine nodeEngine, int
metricsPartitionCount) {
+ Objects.requireNonNull(nodeEngine, "nodeEngine");
+ this.nodeEngine = nodeEngine;
+ this.metricsPartitionCount = metricsPartitionCount;
+ }
+
+ private void ensureInitialized() {
+ if (authoritativeStateStores != null && auxiliaryStateStores != null) {
+ return;
+ }
+ synchronized (this) {
+ if (authoritativeStateStores != null && auxiliaryStateStores !=
null) {
+ return;
+ }
+
+ MetricsSnapshotStateStore metricsSnapshotStore =
+ new HazelcastMetricsSnapshotStateStore(
+
nodeEngine.getHazelcastInstance().getMap(RUNNING_JOB_METRICS),
+ metricsPartitionCount);
+ CheckpointOverviewStateStore checkpointOverviewStateStore =
+ new HazelcastCheckpointOverviewStateStore(
+
nodeEngine.getHazelcastInstance().getMap(CHECKPOINT_MONITOR));
+ this.authoritativeStateStores = new
DefaultAuthoritativeStateStores();
+ this.auxiliaryStateStores =
+ new DefaultAuxiliaryStateStores(
+ metricsSnapshotStore,
checkpointOverviewStateStore);
+ }
+ }
+
+ @Override
+ public AuthoritativeStateStores authoritative() {
+ ensureInitialized();
+ return authoritativeStateStores;
+ }
+
+ @Override
+ public AuxiliaryStateStores auxiliary() {
+ ensureInitialized();
+ return auxiliaryStateStores;
+ }
+
+ @Override
+ public void close() {
+ if (auxiliaryStateStores != null) {
+ auxiliaryStateStores.close();
+ }
+ if (authoritativeStateStores != null) {
+ authoritativeStateStores.close();
+ }
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/metrics/MetricsSnapshotStateStore.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/metrics/MetricsSnapshotStateStore.java
new file mode 100644
index 0000000000..03f01ec5ef
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/metrics/MetricsSnapshotStateStore.java
@@ -0,0 +1,113 @@
+/*
+ * 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.server.common.statestore.metrics;
+
+import org.apache.seatunnel.engine.server.common.statestore.StateStore;
+import org.apache.seatunnel.engine.server.dag.physical.PipelineLocation;
+import org.apache.seatunnel.engine.server.execution.TaskLocation;
+import org.apache.seatunnel.engine.server.metrics.SeaTunnelMetricsContext;
+
+import java.util.Collections;
+import java.util.Map;
+
+/**
+ * Store for runtime task metrics snapshots.
+ *
+ * <p>Here, {@code merge()} does not mean merging the internals of {@link
SeaTunnelMetricsContext}.
+ * It means batch-applying the latest snapshots for multiple tasks.
+ *
+ * <p>In the current SeaTunnel metrics reporting flow, each task sends its
latest {@link
+ * SeaTunnelMetricsContext} to the coordinator, and the coordinator replaces
the snapshot by task
+ * key. The Hazelcast implementation uses {@code compute()} internally to
safely update the outer
+ * partition map, while the RocksDB implementation uses task location directly
as the key, so
+ * task-level overwrite preserves the same storage meaning. Even when a
backend uses internal
+ * partition buckets, the exposed {@link StateStore} contract remains
task-based, so {@link #size()}
+ * must report the number of task snapshots rather than backend-specific
buckets.
+ */
+public interface MetricsSnapshotStateStore
+ extends StateStore<TaskLocation, SeaTunnelMetricsContext> {
+
+ /**
+ * Applies metrics snapshots for multiple tasks in a single batch.
+ *
+ * @param snapshot snapshot batch to apply
+ */
+ void merge(Map<TaskLocation, SeaTunnelMetricsContext> snapshot);
+
+ /**
+ * Stores a single task snapshot by delegating to {@link #merge(Map)}.
+ *
+ * @param taskLocation task location to store
+ * @param metricsContext metrics snapshot to store
+ */
+ @Override
+ default void put(TaskLocation taskLocation, SeaTunnelMetricsContext
metricsContext) {
+ merge(Collections.singletonMap(taskLocation, metricsContext));
+ }
+
+ /**
+ * Checks whether a task snapshot exists.
+ *
+ * @param taskLocation task location to check
+ * @return {@code true} if a snapshot exists for the task
+ */
+ @Override
+ default boolean containsKey(TaskLocation taskLocation) {
+ return get(taskLocation) != null;
+ }
+
+ /**
+ * Conditional insertion is intentionally not exposed for metrics
snapshots because the current
+ * engine model treats them as latest-snapshot overwrites.
+ *
+ * @param taskLocation task location to store
+ * @param metricsContext metrics snapshot to store
+ * @return never returns normally
+ */
+ @Override
+ default SeaTunnelMetricsContext putIfAbsent(
+ TaskLocation taskLocation, SeaTunnelMetricsContext metricsContext)
{
+ throw new UnsupportedOperationException(
+ "Metrics snapshots are updated through merge semantics rather
than putIfAbsent.");
+ }
+
+ /**
+ * Removes all task metrics belonging to a specific pipeline.
+ *
+ * @param pipelineLocation pipeline location to remove
+ */
+ void removePipeline(PipelineLocation pipelineLocation);
+
+ /**
+ * Checks whether any task snapshot exists for a specific pipeline.
+ *
+ * @param pipelineLocation pipeline location to check
+ * @return {@code true} if any snapshot exists for the pipeline
+ */
+ boolean containsPipeline(PipelineLocation pipelineLocation);
+
+ /**
+ * Returns the number of active backend partition buckets currently
holding task snapshots.
+ *
+ * <p>This value is observability-oriented and may differ from {@link
#size()}, which reports
+ * logical task snapshot count.
+ *
+ * @return active backend partition bucket count
+ */
+ int activePartitionKeyCount();
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/metrics/hazelcast/HazelcastMetricsSnapshotStateStore.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/metrics/hazelcast/HazelcastMetricsSnapshotStateStore.java
new file mode 100644
index 0000000000..be4e3f83eb
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/metrics/hazelcast/HazelcastMetricsSnapshotStateStore.java
@@ -0,0 +1,322 @@
+/*
+ * 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.server.common.statestore.metrics.hazelcast;
+
+import
org.apache.seatunnel.engine.server.common.statestore.metrics.MetricsSnapshotStateStore;
+import org.apache.seatunnel.engine.server.dag.physical.PipelineLocation;
+import org.apache.seatunnel.engine.server.execution.TaskLocation;
+import org.apache.seatunnel.engine.server.metrics.SeaTunnelMetricsContext;
+
+import com.hazelcast.core.EntryEvent;
+import com.hazelcast.map.IMap;
+import com.hazelcast.map.listener.EntryAddedListener;
+import com.hazelcast.map.listener.EntryRemovedListener;
+import com.hazelcast.map.listener.EntryUpdatedListener;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
+
+/** Implementation backed by a partitioned Hazelcast metrics {@link IMap}. */
+public class HazelcastMetricsSnapshotStateStore
+ implements MetricsSnapshotStateStore, AutoCloseable {
+
+ private final IMap<Long, Map<TaskLocation, SeaTunnelMetricsContext>>
metricsImap;
+ private final int partitionCount;
+ private final AtomicLong activePartitionKeyCount = new AtomicLong();
+ private final AtomicLong taskSnapshotCount = new AtomicLong();
+ private final Object metricsStatsLock = new Object();
+ private final Set<Long> dirtyPartitionKeys = ConcurrentHashMap.newKeySet();
+ private final Map<Long, RunningJobMetricsStats> statsByPartitionKey = new
HashMap<>();
+ private final AtomicBoolean statsInitializing = new AtomicBoolean(false);
+ private volatile UUID metricsListenerId;
+
+ public HazelcastMetricsSnapshotStateStore(
+ IMap<Long, Map<TaskLocation, SeaTunnelMetricsContext>> metricsImap,
+ int partitionCount) {
+ this.metricsImap = metricsImap;
+ this.partitionCount = partitionCount;
+ initStats();
+ }
+
+ @Override
+ public void merge(Map<TaskLocation, SeaTunnelMetricsContext> snapshot) {
+ if (snapshot == null || snapshot.isEmpty()) {
+ return;
+ }
+
+ Map<Long, Map<TaskLocation, SeaTunnelMetricsContext>> partitioned =
new HashMap<>();
+ snapshot.forEach(
+ (key, value) -> {
+ long partition = partition(key);
+ partitioned.computeIfAbsent(partition, k -> new
HashMap<>()).put(key, value);
+ });
+
+ partitioned
+ .entrySet()
+ .parallelStream()
+ .forEach(
+ entry -> {
+ metricsImap.compute(
+ entry.getKey(),
+ (k, oldVal) -> {
+ if (oldVal == null) oldVal = new
HashMap<>();
+ oldVal.putAll(entry.getValue());
+ return oldVal;
+ });
+ });
+ }
+
+ @Override
+ public SeaTunnelMetricsContext get(TaskLocation taskLocation) {
+ Map<TaskLocation, SeaTunnelMetricsContext> partitionMap =
+ metricsImap.get(partition(taskLocation));
+ if (partitionMap == null) {
+ return null;
+ }
+ return partitionMap.get(taskLocation);
+ }
+
+ @Override
+ public void remove(final TaskLocation taskLocation) {
+ metricsImap.compute(
+ partition(taskLocation),
+ (ignored, current) -> {
+ if (current == null) {
+ return null;
+ }
+ Map<TaskLocation, SeaTunnelMetricsContext> updated = new
HashMap<>(current);
+ updated.remove(taskLocation);
+ return updated.isEmpty() ? null : updated;
+ });
+ }
+
+ @Override
+ public void removePipeline(final PipelineLocation pipelineLocation) {
+ Map<Long, List<TaskLocation>> partitionedTasks = new HashMap<>();
+ for (Map.Entry<Long, Map<TaskLocation, SeaTunnelMetricsContext>> entry
:
+ metricsImap.entrySet()) {
+ long partition = entry.getKey();
+ List<TaskLocation> tasksToRemove =
+ entry.getValue().keySet().stream()
+ .filter(
+ t ->
+ t.getTaskGroupLocation()
+ .getPipelineLocation()
+ .equals(pipelineLocation))
+ .collect(Collectors.toList());
+ if (!tasksToRemove.isEmpty()) {
+ partitionedTasks.put(partition, tasksToRemove);
+ }
+ }
+
+ partitionedTasks
+ .entrySet()
+ .parallelStream()
+ .forEach(
+ entry -> {
+ long partition = entry.getKey();
+ List<TaskLocation> tasks = entry.getValue();
+ metricsImap.compute(
+ partition,
+ (k, oldVal) -> {
+ if (oldVal != null) {
+ tasks.forEach(oldVal::remove);
+ if (oldVal.isEmpty()) return null;
+ }
+ return oldVal;
+ });
+ });
+ }
+
+ @Override
+ public boolean containsPipeline(PipelineLocation pipelineLocation) {
+ for (Map<TaskLocation, SeaTunnelMetricsContext> partitionMap :
metricsImap.values()) {
+ boolean found =
+ partitionMap.keySet().stream()
+ .anyMatch(
+ taskLocation ->
+ pipelineLocation.equals(
+ taskLocation
+
.getTaskGroupLocation()
+
.getPipelineLocation()));
+ if (found) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public int size() {
+ return Math.toIntExact(taskSnapshotCount.get());
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return taskSnapshotCount.get() == 0L;
+ }
+
+ @Override
+ public int activePartitionKeyCount() {
+ return Math.toIntExact(activePartitionKeyCount.get());
+ }
+
+ private long partition(TaskLocation taskLocation) {
+ return (taskLocation.hashCode() & Integer.MAX_VALUE) % partitionCount;
+ }
+
+ @Override
+ public void close() {
+ removeMetricsListener();
+ }
+
+ private void initStats() {
+ removeMetricsListener();
+ metricsListenerId = metricsImap.addEntryListener(new
MetricsEntryListener(), true);
+ statsInitializing.set(true);
+ dirtyPartitionKeys.clear();
+
+ Map<Long, RunningJobMetricsStats> snapshotStats = new HashMap<>();
+ metricsImap.forEach(
+ (partitionKey, metrics) ->
+ snapshotStats.put(partitionKey,
toRunningJobMetricsStats(metrics)));
+
+ Set<Long> dirtyKeysDuringSnapshot = new HashSet<>(dirtyPartitionKeys);
+ for (Long partitionKey : dirtyKeysDuringSnapshot) {
+ snapshotStats.put(
+ partitionKey,
toRunningJobMetricsStats(metricsImap.get(partitionKey)));
+ }
+
+ synchronized (metricsStatsLock) {
+ statsByPartitionKey.clear();
+ activePartitionKeyCount.set(0L);
+ taskSnapshotCount.set(0L);
+ snapshotStats.forEach(this::replaceRunningJobMetricsStatsLocked);
+ statsInitializing.set(false);
+
+ Set<Long> postSnapshotDirtyKeys = new
HashSet<>(dirtyPartitionKeys);
+ dirtyPartitionKeys.clear();
+ for (Long partitionKey : postSnapshotDirtyKeys) {
+ replaceRunningJobMetricsStatsLocked(
+ partitionKey,
toRunningJobMetricsStats(metricsImap.get(partitionKey)));
+ }
+ }
+ }
+
+ private RunningJobMetricsStats toRunningJobMetricsStats(
+ Map<TaskLocation, SeaTunnelMetricsContext> metrics) {
+ if (metrics == null || metrics.isEmpty()) {
+ return RunningJobMetricsStats.EMPTY;
+ }
+ return new RunningJobMetricsStats(1L, metrics.size());
+ }
+
+ private void replaceRunningJobMetricsStatsLocked(
+ Long partitionKey, RunningJobMetricsStats stats) {
+ RunningJobMetricsStats currentStats =
statsByPartitionKey.get(partitionKey);
+ if (currentStats != null) {
+ activePartitionKeyCount.addAndGet(-currentStats.partitionKeyCount);
+ taskSnapshotCount.addAndGet(-currentStats.taskSnapshotCount);
+ }
+
+ if (stats.isEmpty()) {
+ statsByPartitionKey.remove(partitionKey);
+ return;
+ }
+
+ statsByPartitionKey.put(partitionKey, stats);
+ activePartitionKeyCount.addAndGet(stats.partitionKeyCount);
+ taskSnapshotCount.addAndGet(stats.taskSnapshotCount);
+ }
+
+ private void replaceRunningJobMetricsStats(
+ Long partitionKey, Map<TaskLocation, SeaTunnelMetricsContext>
metrics) {
+ if (statsInitializing.get()) {
+ dirtyPartitionKeys.add(partitionKey);
+ return;
+ }
+ synchronized (metricsStatsLock) {
+ if (statsInitializing.get()) {
+ dirtyPartitionKeys.add(partitionKey);
+ return;
+ }
+ replaceRunningJobMetricsStatsLocked(partitionKey,
toRunningJobMetricsStats(metrics));
+ }
+ }
+
+ private void removeMetricsListener() {
+ if (metricsListenerId != null) {
+ metricsImap.removeEntryListener(metricsListenerId);
+ metricsListenerId = null;
+ }
+ statsInitializing.set(false);
+ dirtyPartitionKeys.clear();
+ synchronized (metricsStatsLock) {
+ statsByPartitionKey.clear();
+ }
+ activePartitionKeyCount.set(0L);
+ taskSnapshotCount.set(0L);
+ }
+
+ private final class MetricsEntryListener
+ implements EntryAddedListener<Long, Map<TaskLocation,
SeaTunnelMetricsContext>>,
+ EntryUpdatedListener<Long, Map<TaskLocation,
SeaTunnelMetricsContext>>,
+ EntryRemovedListener<Long, Map<TaskLocation,
SeaTunnelMetricsContext>> {
+
+ @Override
+ public void entryAdded(EntryEvent<Long, Map<TaskLocation,
SeaTunnelMetricsContext>> event) {
+ replaceRunningJobMetricsStats(event.getKey(), event.getValue());
+ }
+
+ @Override
+ public void entryUpdated(
+ EntryEvent<Long, Map<TaskLocation, SeaTunnelMetricsContext>>
event) {
+ replaceRunningJobMetricsStats(event.getKey(), event.getValue());
+ }
+
+ @Override
+ public void entryRemoved(
+ EntryEvent<Long, Map<TaskLocation, SeaTunnelMetricsContext>>
event) {
+ replaceRunningJobMetricsStats(event.getKey(), null);
+ }
+ }
+
+ private static final class RunningJobMetricsStats {
+ private static final RunningJobMetricsStats EMPTY = new
RunningJobMetricsStats(0L, 0L);
+
+ private final long partitionKeyCount;
+ private final long taskSnapshotCount;
+
+ private RunningJobMetricsStats(long partitionKeyCount, long
taskSnapshotCount) {
+ this.partitionKeyCount = partitionKeyCount;
+ this.taskSnapshotCount = taskSnapshotCount;
+ }
+
+ private boolean isEmpty() {
+ return partitionKeyCount == 0L && taskSnapshotCount == 0L;
+ }
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/execution/TaskExecutionContext.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/execution/TaskExecutionContext.java
index 74ffb7aad9..ad79765930 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/execution/TaskExecutionContext.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/execution/TaskExecutionContext.java
@@ -17,32 +17,33 @@
package org.apache.seatunnel.engine.server.execution;
-import org.apache.seatunnel.engine.common.Constant;
-import org.apache.seatunnel.engine.server.SeaTunnelServer;
import org.apache.seatunnel.engine.server.TaskExecutionService;
+import org.apache.seatunnel.engine.server.common.SeaTunnelEngineContext;
import org.apache.seatunnel.engine.server.metrics.SeaTunnelMetricsContext;
import org.apache.seatunnel.engine.server.utils.NodeEngineUtil;
import com.hazelcast.cluster.Address;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.logging.ILogger;
-import com.hazelcast.map.IMap;
import com.hazelcast.spi.impl.NodeEngineImpl;
import com.hazelcast.spi.impl.operationservice.Operation;
import com.hazelcast.spi.impl.operationservice.impl.InvocationFuture;
-import java.util.HashMap;
-
public class TaskExecutionContext {
private final Task task;
private final NodeEngineImpl nodeEngine;
+ private final SeaTunnelEngineContext engineContext;
private final TaskExecutionService taskExecutionService;
public TaskExecutionContext(
- Task task, NodeEngineImpl nodeEngine, TaskExecutionService
taskExecutionService) {
+ Task task,
+ NodeEngineImpl nodeEngine,
+ SeaTunnelEngineContext engineContext,
+ TaskExecutionService taskExecutionService) {
this.task = task;
this.nodeEngine = nodeEngine;
+ this.engineContext = engineContext;
this.taskExecutionService = taskExecutionService;
}
@@ -59,18 +60,10 @@ public class TaskExecutionContext {
}
public SeaTunnelMetricsContext getOrCreateMetricsContext(TaskLocation
taskLocation) {
- IMap<Long, HashMap<TaskLocation, SeaTunnelMetricsContext>> map =
-
nodeEngine.getHazelcastInstance().getMap(Constant.IMAP_RUNNING_JOB_METRICS);
- int partitionCount =
- taskExecutionService
- .getSeaTunnelConfig()
- .getEngineConfig()
- .getJobMetricsPartitionCount();
- long partition = SeaTunnelServer.getMetricsImapPartition(taskLocation,
partitionCount);
- HashMap<TaskLocation, SeaTunnelMetricsContext> centralMap =
map.get(partition);
- return centralMap == null || centralMap.get(taskLocation) == null
- ? new SeaTunnelMetricsContext()
- : centralMap.get(taskLocation);
+ return engineContext
+ .getStateStores()
+ .metricsSnapshotStore()
+ .getOrDefault(taskLocation, new SeaTunnelMetricsContext());
}
public <T> T getTask() {
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/EngineStateStoreMetricExports.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/EngineStateStoreMetricExports.java
index c8a8697039..0369818ea9 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/EngineStateStoreMetricExports.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/EngineStateStoreMetricExports.java
@@ -18,6 +18,7 @@
package org.apache.seatunnel.engine.server.telemetry.metrics.exports;
import org.apache.seatunnel.engine.common.Constant;
+import
org.apache.seatunnel.engine.server.common.statestore.EngineStateStoreNames;
import org.apache.seatunnel.engine.server.telemetry.metrics.AbstractCollector;
import com.hazelcast.core.HazelcastInstanceNotActiveException;
@@ -39,11 +40,11 @@ public class EngineStateStoreMetricExports extends
AbstractCollector {
Constant.IMAP_RUNNING_JOB_STATE,
Constant.IMAP_STATE_TIMESTAMPS,
Constant.IMAP_OWNED_SLOT_PROFILES,
- Constant.IMAP_RUNNING_JOB_METRICS,
+ EngineStateStoreNames.RUNNING_JOB_METRICS,
Constant.IMAP_FINISHED_JOB_STATE,
Constant.IMAP_FINISHED_JOB_METRICS,
Constant.IMAP_FINISHED_JOB_VERTEX_INFO,
- Constant.IMAP_CHECKPOINT_MONITOR,
+ EngineStateStoreNames.CHECKPOINT_MONITOR,
Constant.IMAP_CONNECTOR_JAR_REF_COUNTERS,
Constant.IMAP_CHECKPOINT_ID,
Constant.IMAP_PENDING_PIPELINE_CLEANUP);
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServicePipelineCleanupTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServicePipelineCleanupTest.java
index 07f32459d1..e8218e46ba 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServicePipelineCleanupTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServicePipelineCleanupTest.java
@@ -19,6 +19,7 @@ package org.apache.seatunnel.engine.server;
import org.apache.seatunnel.engine.common.Constant;
import org.apache.seatunnel.engine.core.job.PipelineStatus;
+import
org.apache.seatunnel.engine.server.common.statestore.metrics.MetricsSnapshotStateStore;
import org.apache.seatunnel.engine.server.dag.physical.PipelineLocation;
import org.apache.seatunnel.engine.server.execution.TaskGroupLocation;
import org.apache.seatunnel.engine.server.execution.TaskLocation;
@@ -252,14 +253,9 @@ class CoordinatorServicePipelineCleanupTest extends
AbstractSeaTunnelServerTest
}
private boolean hasMetricsForPipeline(PipelineLocation pipelineLocation) {
- IMap<Long, Map<TaskLocation, SeaTunnelMetricsContext>> metricsIMap =
-
nodeEngine.getHazelcastInstance().getMap(Constant.IMAP_RUNNING_JOB_METRICS);
- return metricsIMap.entrySet().stream()
- .flatMap(entry -> entry.getValue().keySet().stream())
- .anyMatch(
- taskLocation ->
- pipelineLocation.equals(
-
taskLocation.getTaskGroupLocation().getPipelineLocation()));
+ MetricsSnapshotStateStore metricsStore =
+
server.getEngineContext().getStateStores().metricsSnapshotStore();
+ return metricsStore.containsPipeline(pipelineLocation);
}
private void awaitCoordinatorActive(CoordinatorService coordinatorService)
{
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceTest.java
index a2ab949248..59d86896a7 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceTest.java
@@ -35,6 +35,8 @@ import org.apache.seatunnel.engine.core.job.JobDAGInfo;
import org.apache.seatunnel.engine.core.job.JobImmutableInformation;
import org.apache.seatunnel.engine.core.job.JobInfo;
import org.apache.seatunnel.engine.core.job.PipelineStatus;
+import org.apache.seatunnel.engine.server.common.SeaTunnelEngineContext;
+import
org.apache.seatunnel.engine.server.common.statestore.metrics.MetricsSnapshotStateStore;
import org.apache.seatunnel.engine.server.dag.physical.PhysicalPlan;
import org.apache.seatunnel.engine.server.dag.physical.PipelineLocation;
import org.apache.seatunnel.engine.server.dag.physical.SubPlan;
@@ -525,9 +527,11 @@ public class CoordinatorServiceTest {
Mockito.when(nodeEngine.getLogger(Mockito.any(Class.class))).thenReturn(logger);
Mockito.when(nodeEngine.getHazelcastInstance()).thenReturn(hazelcastInstance);
Mockito.when(hazelcastInstance.getMap(Mockito.anyString())).thenReturn(map);
+ SeaTunnelEngineContext engineContext =
Mockito.mock(SeaTunnelEngineContext.class);
+ Mockito.when(server.getEngineContext()).thenReturn(engineContext);
CoordinatorService coordinatorService =
- new CoordinatorService(nodeEngine, server, engineConfig);
+ new CoordinatorService(nodeEngine, server,
server.getEngineContext(), engineConfig);
stopCoordinatorSchedulers(coordinatorService);
return coordinatorService;
}
@@ -920,12 +924,12 @@ public class CoordinatorServiceTest {
"batch_fake_to_console.conf",
"test_cleanup_metrics_imap");
CoordinatorService coordinatorService =
jobInformation.coordinatorService;
- IMap<Long, HashMap<TaskLocation, SeaTunnelMetricsContext>> metricsImap
=
- coordinatorService.getMetricsImap();
- await().atMost(60000, TimeUnit.MILLISECONDS)
- .untilAsserted(() ->
Assertions.assertFalse(metricsImap.isEmpty()));
- await().atMost(60000, TimeUnit.MILLISECONDS)
- .untilAsserted(() ->
Assertions.assertTrue(metricsImap.isEmpty()));
+ MetricsSnapshotStateStore metricsSnapshotStateStore =
+ coordinatorService.getMetricsSnapshotStateStore();
+ await().atMost(10000, TimeUnit.MILLISECONDS)
+ .untilAsserted(() ->
Assertions.assertFalse(metricsSnapshotStateStore.isEmpty()));
+ await().atMost(10000, TimeUnit.MILLISECONDS)
+ .untilAsserted(() ->
Assertions.assertTrue(metricsSnapshotStateStore.isEmpty()));
jobInformation.coordinatorService.clearCoordinatorService();
jobInformation.coordinatorServiceTest.shutdown();
@@ -941,12 +945,12 @@ public class CoordinatorServiceTest {
"batch_fake_to_console.conf",
"test_cleanup_metrics_imap_with_partition_config");
CoordinatorService coordinatorService =
jobInformation.coordinatorService;
- IMap<Long, HashMap<TaskLocation, SeaTunnelMetricsContext>> metricsImap
=
- coordinatorService.getMetricsImap();
- await().atMost(60000, TimeUnit.MILLISECONDS)
- .untilAsserted(() ->
Assertions.assertFalse(metricsImap.isEmpty()));
- await().atMost(60000, TimeUnit.MILLISECONDS)
- .untilAsserted(() ->
Assertions.assertTrue(metricsImap.isEmpty()));
+ MetricsSnapshotStateStore metricsSnapshotStateStore =
+ coordinatorService.getMetricsSnapshotStateStore();
+ await().atMost(10000, TimeUnit.MILLISECONDS)
+ .untilAsserted(() ->
Assertions.assertFalse(metricsSnapshotStateStore.isEmpty()));
+ await().atMost(10000, TimeUnit.MILLISECONDS)
+ .untilAsserted(() ->
Assertions.assertTrue(metricsSnapshotStateStore.isEmpty()));
jobInformation.coordinatorService.clearCoordinatorService();
jobInformation.coordinatorServiceTest.shutdown();
@@ -971,8 +975,8 @@ public class CoordinatorServiceTest {
taskLocation.setTaskID(i);
localMap.put(taskLocation, new SeaTunnelMetricsContext());
}
- IMap<Long, HashMap<TaskLocation, SeaTunnelMetricsContext>>
metricsImap =
- server1.getCoordinatorService().getMetricsImap();
+ MetricsSnapshotStateStore metricsSnapshotStateStore =
+
server1.getCoordinatorService().getMetricsSnapshotStateStore();
CompletableFuture.runAsync(
() -> {
try {
@@ -988,8 +992,15 @@ public class CoordinatorServiceTest {
throw new CompletionException(e);
}
});
- await().atMost(60000, TimeUnit.MILLISECONDS)
- .untilAsserted(() -> Assertions.assertEquals(10,
metricsImap.size()));
+ await().atMost(10000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ 10,
+
metricsSnapshotStateStore.activePartitionKeyCount()));
+ await().atMost(10000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () -> Assertions.assertEquals(100,
metricsSnapshotStateStore.size()));
} finally {
instance1.shutdown();
setDefaultConfigFile();
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/common/statestore/checkpoint/hazelcast/HazelcastCheckpointOverviewStateStoreTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/common/statestore/checkpoint/hazelcast/HazelcastCheckpointOverviewStateStoreTest.java
new file mode 100644
index 0000000000..17254bc969
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/common/statestore/checkpoint/hazelcast/HazelcastCheckpointOverviewStateStoreTest.java
@@ -0,0 +1,133 @@
+/*
+ * 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.server.common.statestore.checkpoint.hazelcast;
+
+import org.apache.seatunnel.engine.core.checkpoint.CheckpointHistoryEntry;
+import org.apache.seatunnel.engine.core.checkpoint.CheckpointInfo;
+import org.apache.seatunnel.engine.core.checkpoint.CheckpointOverview;
+import org.apache.seatunnel.engine.core.checkpoint.CheckpointStatus;
+import org.apache.seatunnel.engine.core.checkpoint.CheckpointType;
+import org.apache.seatunnel.engine.core.checkpoint.InProgressCheckpoint;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import com.hazelcast.config.Config;
+import com.hazelcast.core.Hazelcast;
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.map.IMap;
+
+import java.util.concurrent.TimeUnit;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class HazelcastCheckpointOverviewStateStoreTest {
+
+ private static HazelcastInstance hazelcastInstance;
+
+ @BeforeAll
+ static void beforeAll() {
+ Config config = new Config();
+ config.setClusterName("HazelcastCheckpointOverviewStateStoreTest-" +
System.nanoTime());
+ hazelcastInstance = Hazelcast.newHazelcastInstance(config);
+ }
+
+ @AfterAll
+ static void afterAll() {
+ if (hazelcastInstance != null) {
+ hazelcastInstance.shutdown();
+ }
+ }
+
+ @Test
+ void countAccessorsShouldUseLogicalOverviewCounts() {
+ IMap<Long, CheckpointOverview> iMap =
+ hazelcastInstance.getMap("checkpoint-overview-counts");
+ iMap.clear();
+ HazelcastCheckpointOverviewStateStore store =
+ new HazelcastCheckpointOverviewStateStore(iMap);
+
+ store.updateOverview(
+ 1L,
+ 1,
+ pipeline -> {
+ pipeline.getInProgress()
+ .add(
+ new InProgressCheckpoint(
+ 101L,
CheckpointType.CHECKPOINT_TYPE, 10L, 1, 2));
+ pipeline.getInProgress()
+ .add(
+ new InProgressCheckpoint(
+ 102L,
CheckpointType.CHECKPOINT_TYPE, 20L, 1, 2));
+ pipeline.addHistory(historyEntry(1L, 1, 201L), 8);
+ });
+ store.updateOverview(1L, 2, pipeline ->
pipeline.addHistory(historyEntry(1L, 2, 202L), 8));
+ store.updateOverview(
+ 2L,
+ 1,
+ pipeline -> {
+ pipeline.getInProgress()
+ .add(
+ new InProgressCheckpoint(
+ 103L,
CheckpointType.CHECKPOINT_TYPE, 30L, 1, 2));
+ pipeline.addHistory(historyEntry(2L, 1, 203L), 8);
+ });
+
+ awaitOverviewStats(store, 2L, 3L, 3L, 2);
+
+ store.remove(1L);
+
+ awaitOverviewStats(store, 1L, 1L, 1L, 1);
+ }
+
+ private static void awaitOverviewStats(
+ HazelcastCheckpointOverviewStateStore store,
+ long expectedOverviewJobCount,
+ long expectedInProgressCheckpointCount,
+ long expectedRetainedHistoryCount,
+ int expectedSize) {
+ await().atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ assertEquals(expectedOverviewJobCount,
store.getOverviewJobCount());
+ assertEquals(
+ expectedInProgressCheckpointCount,
+ store.getInProgressCheckpointCount());
+ assertEquals(
+ expectedRetainedHistoryCount,
store.getRetainedHistoryCount());
+ assertEquals(expectedSize, store.size());
+ });
+ }
+
+ private static CheckpointHistoryEntry historyEntry(
+ long jobId, int pipelineId, long checkpointId) {
+ return CheckpointHistoryEntry.builder()
+ .jobId(jobId)
+ .pipelineId(pipelineId)
+ .checkpointInfo(
+ CheckpointInfo.builder()
+ .checkpointId(checkpointId)
+ .checkpointType(CheckpointType.CHECKPOINT_TYPE)
+ .status(CheckpointStatus.COMPLETED)
+ .triggerTimestamp(checkpointId)
+ .build())
+ .build();
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/common/statestore/metrics/hazelcast/HazelcastMetricsSnapshotStateStoreTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/common/statestore/metrics/hazelcast/HazelcastMetricsSnapshotStateStoreTest.java
new file mode 100644
index 0000000000..7ef1ada64e
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/common/statestore/metrics/hazelcast/HazelcastMetricsSnapshotStateStoreTest.java
@@ -0,0 +1,203 @@
+/*
+ * 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.server.common.statestore.metrics.hazelcast;
+
+import org.apache.seatunnel.engine.server.dag.physical.PipelineLocation;
+import org.apache.seatunnel.engine.server.execution.TaskGroupLocation;
+import org.apache.seatunnel.engine.server.execution.TaskLocation;
+import org.apache.seatunnel.engine.server.metrics.SeaTunnelMetricsContext;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import com.hazelcast.config.Config;
+import com.hazelcast.core.Hazelcast;
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.map.IMap;
+
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+class HazelcastMetricsSnapshotStateStoreTest {
+
+ private static HazelcastInstance hazelcastInstance;
+ private static final String METRIC_NAME = "test.metric";
+
+ @BeforeAll
+ static void beforeAll() {
+ Config config = new Config();
+ config.setClusterName("HazelcastMetricsSnapshotStateStoreTest-" +
System.nanoTime());
+ hazelcastInstance = Hazelcast.newHazelcastInstance(config);
+ }
+
+ @AfterAll
+ static void afterAll() {
+ if (hazelcastInstance != null) {
+ hazelcastInstance.shutdown();
+ }
+ }
+
+ @Test
+ void mergeShouldStoreAndOverwriteSnapshots() {
+ IMap<Long, Map<TaskLocation, SeaTunnelMetricsContext>> iMap =
+ hazelcastInstance.getMap("metrics-snapshot-merge");
+ iMap.clear();
+ HazelcastMetricsSnapshotStateStore store = new
HazelcastMetricsSnapshotStateStore(iMap, 8);
+
+ TaskLocation taskOne = taskLocation(1L, 10, 100L, 0L, 0);
+ TaskLocation taskTwo = taskLocation(1L, 11, 101L, 0L, 0);
+ SeaTunnelMetricsContext metricsOne = metricsContextWithCounterValue(1);
+ SeaTunnelMetricsContext metricsTwo = metricsContextWithCounterValue(2);
+ SeaTunnelMetricsContext updatedMetricsOne =
metricsContextWithCounterValue(3);
+
+ Map<TaskLocation, SeaTunnelMetricsContext> initialSnapshot = new
LinkedHashMap<>();
+ initialSnapshot.put(taskOne, metricsOne);
+ initialSnapshot.put(taskTwo, metricsTwo);
+ store.merge(initialSnapshot);
+
+ assertCounterValue(1, store.get(taskOne));
+ assertCounterValue(2, store.get(taskTwo));
+ awaitSize(store, 2);
+
+ store.merge(singletonSnapshot(taskOne, updatedMetricsOne));
+
+ assertCounterValue(3, store.get(taskOne));
+ assertCounterValue(2, store.get(taskTwo));
+ awaitSize(store, 2);
+ }
+
+ @Test
+ void removeShouldDeleteSingleTaskSnapshot() {
+ IMap<Long, Map<TaskLocation, SeaTunnelMetricsContext>> iMap =
+ hazelcastInstance.getMap("metrics-snapshot-remove");
+ iMap.clear();
+ HazelcastMetricsSnapshotStateStore store = new
HazelcastMetricsSnapshotStateStore(iMap, 8);
+
+ TaskLocation taskOne = taskLocation(2L, 20, 200L, 0L, 0);
+ TaskLocation taskTwo = taskLocation(2L, 21, 201L, 0L, 0);
+ SeaTunnelMetricsContext metricsOne =
metricsContextWithCounterValue(10);
+ SeaTunnelMetricsContext metricsTwo =
metricsContextWithCounterValue(20);
+
+ Map<TaskLocation, SeaTunnelMetricsContext> snapshot = new
LinkedHashMap<>();
+ snapshot.put(taskOne, metricsOne);
+ snapshot.put(taskTwo, metricsTwo);
+ store.merge(snapshot);
+
+ store.remove(taskOne);
+
+ assertNull(store.get(taskOne));
+ assertCounterValue(20, store.get(taskTwo));
+ awaitSize(store, 1);
+
+ store.remove(taskTwo);
+
+ assertNull(store.get(taskTwo));
+ awaitSize(store, 0);
+ }
+
+ @Test
+ void removePipelineShouldDeleteOnlyMatchingPipelineSnapshots() {
+ IMap<Long, Map<TaskLocation, SeaTunnelMetricsContext>> iMap =
+ hazelcastInstance.getMap("metrics-snapshot-remove-pipeline");
+ iMap.clear();
+ HazelcastMetricsSnapshotStateStore store = new
HazelcastMetricsSnapshotStateStore(iMap, 8);
+
+ PipelineLocation pipelineToRemove = new PipelineLocation(3L, 30);
+ PipelineLocation pipelineToKeep = new PipelineLocation(3L, 31);
+ TaskLocation removedOne = taskLocation(3L, 30, 300L, 0L, 0);
+ TaskLocation removedTwo = taskLocation(3L, 30, 301L, 0L, 0);
+ TaskLocation kept = taskLocation(3L, 31, 302L, 0L, 0);
+ SeaTunnelMetricsContext removedOneMetrics =
metricsContextWithCounterValue(100);
+ SeaTunnelMetricsContext removedTwoMetrics =
metricsContextWithCounterValue(200);
+ SeaTunnelMetricsContext keptMetrics =
metricsContextWithCounterValue(300);
+
+ Map<TaskLocation, SeaTunnelMetricsContext> snapshot = new
LinkedHashMap<>();
+ snapshot.put(removedOne, removedOneMetrics);
+ snapshot.put(removedTwo, removedTwoMetrics);
+ snapshot.put(kept, keptMetrics);
+ store.merge(snapshot);
+
+ store.removePipeline(pipelineToRemove);
+
+ assertNull(store.get(removedOne));
+ assertNull(store.get(removedTwo));
+ assertCounterValue(300, store.get(kept));
+ awaitSize(store, 1);
+ assertEquals(pipelineToKeep,
kept.getTaskGroupLocation().getPipelineLocation());
+ }
+
+ @Test
+ void sizeShouldCountTaskSnapshotsInsteadOfPartitionBuckets() {
+ IMap<Long, Map<TaskLocation, SeaTunnelMetricsContext>> iMap =
+ hazelcastInstance.getMap("metrics-snapshot-size");
+ iMap.clear();
+ HazelcastMetricsSnapshotStateStore store = new
HazelcastMetricsSnapshotStateStore(iMap, 1);
+
+ TaskLocation taskOne = taskLocation(4L, 40, 400L, 0L, 0);
+ TaskLocation taskTwo = taskLocation(4L, 40, 401L, 0L, 1);
+
+ Map<TaskLocation, SeaTunnelMetricsContext> snapshot = new
LinkedHashMap<>();
+ snapshot.put(taskOne, metricsContextWithCounterValue(1));
+ snapshot.put(taskTwo, metricsContextWithCounterValue(2));
+ store.merge(snapshot);
+
+ assertEquals(1, iMap.size());
+ awaitSize(store, 2);
+ }
+
+ private static void awaitSize(HazelcastMetricsSnapshotStateStore store,
int expectedSize) {
+ await().atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(() -> assertEquals(expectedSize, store.size()));
+ }
+
+ private static Map<TaskLocation, SeaTunnelMetricsContext>
singletonSnapshot(
+ TaskLocation taskLocation, SeaTunnelMetricsContext metricsContext)
{
+ Map<TaskLocation, SeaTunnelMetricsContext> snapshot = new HashMap<>();
+ snapshot.put(taskLocation, metricsContext);
+ return snapshot;
+ }
+
+ private static SeaTunnelMetricsContext metricsContextWithCounterValue(long
value) {
+ SeaTunnelMetricsContext metricsContext = new SeaTunnelMetricsContext();
+ metricsContext.counter(METRIC_NAME).inc(value);
+ return metricsContext;
+ }
+
+ private static void assertCounterValue(long expected,
SeaTunnelMetricsContext metricsContext) {
+ assertEquals(expected, metricsContext.counter(METRIC_NAME).getCount());
+ }
+
+ private static TaskLocation taskLocation(
+ long jobId,
+ int pipelineId,
+ long taskGroupId,
+ long taskInGroupIndex,
+ int parallelismIndex) {
+ return new TaskLocation(
+ new TaskGroupLocation(jobId, pipelineId, taskGroupId),
+ taskInGroupIndex,
+ parallelismIndex);
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/EngineStateStoreLogicalMetricExportsTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/EngineStateStoreLogicalMetricExportsTest.java
index dbbfa1dcd5..7cb201cdac 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/EngineStateStoreLogicalMetricExportsTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/EngineStateStoreLogicalMetricExportsTest.java
@@ -32,6 +32,7 @@ import
org.apache.seatunnel.engine.server.SeaTunnelServerStarter;
import org.apache.seatunnel.engine.server.TestUtils;
import org.apache.seatunnel.engine.server.checkpoint.CheckpointCloseReason;
import org.apache.seatunnel.engine.server.checkpoint.CompletedCheckpoint;
+import
org.apache.seatunnel.engine.server.common.statestore.EngineStateStoreNames;
import org.apache.seatunnel.engine.server.execution.TaskGroupLocation;
import org.apache.seatunnel.engine.server.execution.TaskLocation;
import org.apache.seatunnel.engine.server.master.JobHistoryService.JobState;
@@ -187,7 +188,7 @@ class EngineStateStoreLogicalMetricExportsTest {
private void seedRunningJobMetrics() {
IMap<Long, HashMap<TaskLocation, SeaTunnelMetricsContext>> metricsMap =
- instance.getMap(Constant.IMAP_RUNNING_JOB_METRICS);
+ instance.getMap(EngineStateStoreNames.RUNNING_JOB_METRICS);
HashMap<TaskLocation, SeaTunnelMetricsContext> partitionZero = new
HashMap<>();
partitionZero.put(new TaskLocation(new TaskGroupLocation(1L, 1, 1L),
0, 0), metricCtx());
partitionZero.put(new TaskLocation(new TaskGroupLocation(1L, 1, 2L),
0, 0), metricCtx());
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/EngineStateStoreMetricExportsTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/EngineStateStoreMetricExportsTest.java
index f9390dd993..611be9dac8 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/EngineStateStoreMetricExportsTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/EngineStateStoreMetricExportsTest.java
@@ -20,6 +20,7 @@ package
org.apache.seatunnel.engine.server.telemetry.metrics.exports;
import org.apache.seatunnel.engine.common.Constant;
import org.apache.seatunnel.engine.server.SeaTunnelServerStarter;
import org.apache.seatunnel.engine.server.TestUtils;
+import
org.apache.seatunnel.engine.server.common.statestore.EngineStateStoreNames;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
@@ -110,11 +111,11 @@ class EngineStateStoreMetricExportsTest {
Constant.IMAP_RUNNING_JOB_STATE,
Constant.IMAP_STATE_TIMESTAMPS,
Constant.IMAP_OWNED_SLOT_PROFILES,
- Constant.IMAP_RUNNING_JOB_METRICS,
+ EngineStateStoreNames.RUNNING_JOB_METRICS,
Constant.IMAP_FINISHED_JOB_STATE,
Constant.IMAP_FINISHED_JOB_METRICS,
Constant.IMAP_FINISHED_JOB_VERTEX_INFO,
- Constant.IMAP_CHECKPOINT_MONITOR,
+ EngineStateStoreNames.CHECKPOINT_MONITOR,
Constant.IMAP_CONNECTOR_JAR_REF_COUNTERS,
Constant.IMAP_CHECKPOINT_ID,
Constant.IMAP_PENDING_PIPELINE_CLEANUP)),
