This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 570eb97611 [Fix][Zeta] Keep IMap entry expiration times across
restarts in the file map store (#12973)
570eb97611 is described below
commit 570eb976113567efc3cbaa8160dbf8c1cab3c2ac
Author: Doyeon Kim <[email protected]>
AuthorDate: Sun Oct 25 16:21:48 2026 +0000
[Fix][Zeta] Keep IMap entry expiration times across restarts in the file
map store (#12973)
---
docs/en/engines/zeta/state-storage-and-recovery.md | 3 +-
docs/zh/engines/zeta/state-storage-and-recovery.md | 3 +-
.../engine/e2e/ClusterFaultToleranceIT.java | 106 +++++++++++++++++++
.../engine/server/persistence/FileMapStore.java | 52 +++++++--
.../server/persistence/FileMapStoreFactory.java | 6 +-
.../server/persistence/FileMapStoreTest.java | 76 ++++++++++++-
.../engine/imap/storage/api/ExpirableValue.java | 70 ++++++++++++
.../engine/imap/storage/api/IMapStorage.java | 47 +++++++++
.../engine/imap/storage/file/IMapFileStorage.java | 36 ++++++-
.../imap/storage/file/bean/IMapFileData.java | 9 ++
.../engine/imap/storage/file/common/WALReader.java | 51 ++++++++-
.../imap/storage/file/IMapFileStorageTest.java | 25 +++++
.../file/common/WALReaderAndWriterTest.java | 117 +++++++++++++++++++++
13 files changed, 578 insertions(+), 23 deletions(-)
diff --git a/docs/en/engines/zeta/state-storage-and-recovery.md
b/docs/en/engines/zeta/state-storage-and-recovery.md
index b0a7460cb1..c60b69539f 100644
--- a/docs/en/engines/zeta/state-storage-and-recovery.md
+++ b/docs/en/engines/zeta/state-storage-and-recovery.md
@@ -240,7 +240,8 @@ seatunnel:
|---|---|
| Remove from `finished-job-state` IMap | Yes |
| Remove from `finished-job-metrics` IMap | Yes |
-| Delete MapStore persistence files for that job | Yes (after IMap eviction) |
+| Keep the job removed after a cluster restart (IMap persistence enabled) |
Yes (the expiration time is persisted with the entry, so an expired job is not
loaded again and a job that has not expired yet keeps its remaining time) |
+| Delete MapStore persistence files for that job | **No** (the expired records
stay in the IMap storage files) |
| Delete checkpoint storage directories | **No** |
| Delete savepoint directories | **No** |
| Delete WAL files | **No** (only indirectly via compaction) |
diff --git a/docs/zh/engines/zeta/state-storage-and-recovery.md
b/docs/zh/engines/zeta/state-storage-and-recovery.md
index 840dd109f0..98af1592d8 100644
--- a/docs/zh/engines/zeta/state-storage-and-recovery.md
+++ b/docs/zh/engines/zeta/state-storage-and-recovery.md
@@ -231,7 +231,8 @@ seatunnel:
|---|---|
| 从 `finished-job-state` IMap 中移除 | 是 |
| 从 `finished-job-metrics` IMap 中移除 | 是 |
-| 删除该作业的 MapStore 持久化文件 | 是（IMap 驱逐后） |
+| 集群重启后作业仍保持已移除（启用 IMap 持久化时） | 是（过期时间随条目一起持久化，已过期的作业不会被重新加载，尚未过期的作业保留剩余时间） |
+| 删除该作业的 MapStore 持久化文件 | **否**（过期记录仍保留在 IMap 存储文件中） |
| 删除 Checkpoint 存储目录 | **否** |
| 删除 Savepoint 目录 | **否** |
| 删除 WAL 文件 | **否**（仅通过压缩间接处理）|
diff --git
a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/ClusterFaultToleranceIT.java
b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/ClusterFaultToleranceIT.java
index 675d89279e..f99cca0950 100644
---
a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/ClusterFaultToleranceIT.java
+++
b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/ClusterFaultToleranceIT.java
@@ -25,6 +25,7 @@ import org.apache.seatunnel.common.utils.FileUtils;
import org.apache.seatunnel.engine.client.SeaTunnelClient;
import org.apache.seatunnel.engine.client.job.ClientJobExecutionEnvironment;
import org.apache.seatunnel.engine.client.job.ClientJobProxy;
+import org.apache.seatunnel.engine.common.Constant;
import org.apache.seatunnel.engine.common.config.ConfigProvider;
import org.apache.seatunnel.engine.common.config.JobConfig;
import org.apache.seatunnel.engine.common.config.SeaTunnelConfig;
@@ -39,6 +40,7 @@ import
org.testcontainers.shaded.org.apache.commons.lang3.tuple.ImmutablePair;
import com.hazelcast.client.config.ClientConfig;
import com.hazelcast.config.Config;
+import com.hazelcast.core.EntryView;
import com.hazelcast.instance.impl.HazelcastInstanceImpl;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
@@ -951,6 +953,110 @@ public class ClusterFaultToleranceIT {
}
}
+ /**
+ * A finished job is kept in the history maps with a TTL of {@code
history-job-expire-minutes}.
+ * Hazelcast does not delete expired entries from the map store, so the
job must keep its
+ * expiration time after the cluster restarts from the IMap file storage
instead of being loaded
+ * back without one and never expiring.
+ */
+ @Test
+ public void testFinishedJobKeepsExpirationTimeAfterAllNodeDown() throws
Exception {
+ String testCaseName =
"testFinishedJobKeepsExpirationTimeAfterAllNodeDown";
+ String testClusterName =
+
"ClusterFaultToleranceIT_testFinishedJobKeepsExpirationTimeAfterAllNodeDown_"
+ + System.currentTimeMillis();
+ HazelcastInstanceImpl node = null;
+ SeaTunnelClient engineClient = null;
+
+ try {
+ String yaml =
+ "hazelcast:\n"
+ + " network:\n"
+ + " join:\n"
+ + " tcp-ip:\n"
+ + " enabled: true\n"
+ + " member-list:\n"
+ + " - localhost\n"
+ + " port:\n"
+ + " auto-increment: true\n"
+ + " port-count: 100\n"
+ + " port: 5801\n"
+ + " map:\n"
+ + " engine*:\n"
+ + " map-store:\n"
+ + " enabled: true\n"
+ + " initial-mode: EAGER\n"
+ + " factory-class-name:
org.apache.seatunnel.engine.server.persistence.FileMapStoreFactory\n"
+ + " properties:\n"
+ + " type: hdfs\n"
+ + " namespace: /tmp/seatunnel/imap\n"
+ + " clusterName: "
+ + TestUtils.getClusterName(testClusterName)
+ + "\n"
+ + " fs.defaultFS: file:///\n";
+ Config hazelcastConfig = Config.loadFromString(yaml);
+
hazelcastConfig.setClusterName(TestUtils.getClusterName(testClusterName));
+ SeaTunnelConfig seaTunnelConfig =
ConfigProvider.locateAndGetSeaTunnelConfig();
+ seaTunnelConfig.setHazelcastConfig(hazelcastConfig);
+
seaTunnelConfig.getEngineConfig().getHttpConfig().setEnabled(false);
+ node =
SeaTunnelServerStarter.createHazelcastInstance(seaTunnelConfig);
+
+ Common.setDeployMode(DeployMode.CLIENT);
+ ImmutablePair<String, String> testResources =
+ createTestResources(testCaseName, JobMode.BATCH, 100, 1);
+ JobConfig jobConfig = new JobConfig();
+ jobConfig.setName(testCaseName);
+
+ ClientConfig clientConfig =
ConfigProvider.locateAndGetClientConfig();
+
clientConfig.setClusterName(TestUtils.getClusterName(testClusterName));
+ engineClient = new SeaTunnelClient(clientConfig);
+ ClientJobProxy clientJobProxy =
+ engineClient
+ .createExecutionContext(
+ testResources.getRight(), jobConfig,
seaTunnelConfig)
+ .execute();
+ Assertions.assertEquals(JobStatus.FINISHED,
clientJobProxy.waitForJobComplete());
+ Long jobId = clientJobProxy.getJobId();
+
+ HazelcastInstanceImpl finalNode = node;
+ Awaitility.await()
+ .atMost(60, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertNotNull(
+ finalNode
+
.getMap(Constant.IMAP_FINISHED_JOB_STATE)
+ .getEntryView(jobId)));
+ long expirationTime =
+ node.getMap(Constant.IMAP_FINISHED_JOB_STATE)
+ .getEntryView(jobId)
+ .getExpirationTime();
+ Assertions.assertNotEquals(Long.MAX_VALUE, expirationTime);
+
+ engineClient.close();
+ engineClient = null;
+ node.shutdown();
+ node =
SeaTunnelServerStarter.createHazelcastInstance(seaTunnelConfig);
+
+ EntryView<Object, Object> restored =
+
node.getMap(Constant.IMAP_FINISHED_JOB_STATE).getEntryView(jobId);
+ Assertions.assertNotNull(restored);
+ // Hazelcast keeps record times at second granularity.
+ long drift = Math.abs(restored.getExpirationTime() -
expirationTime);
+ Assertions.assertTrue(
+ drift <= TimeUnit.SECONDS.toMillis(2),
+ "expiration time moved by " + drift + " ms across the
restart");
+ } finally {
+ if (engineClient != null) {
+ engineClient.close();
+ }
+
+ if (node != null) {
+ node.shutdown();
+ }
+ }
+ }
+
@Test
@Disabled
public void testStreamJobRestoreFromOssInAllNodeDown() throws Exception {
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/persistence/FileMapStore.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/persistence/FileMapStore.java
index 6230ff80c4..16c20e9330 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/persistence/FileMapStore.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/persistence/FileMapStore.java
@@ -20,13 +20,15 @@ package org.apache.seatunnel.engine.server.persistence;
import org.apache.seatunnel.shade.com.google.common.collect.Maps;
import org.apache.seatunnel.engine.common.utils.FactoryUtil;
+import org.apache.seatunnel.engine.imap.storage.api.ExpirableValue;
import org.apache.seatunnel.engine.imap.storage.api.IMapStorage;
import org.apache.seatunnel.engine.imap.storage.api.IMapStorageFactory;
import
org.apache.seatunnel.engine.server.common.statestore.EngineStateStoreNames;
import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.map.EntryLoader.MetadataAwareValue;
+import com.hazelcast.map.EntryStore;
import com.hazelcast.map.MapLoaderLifecycleSupport;
-import com.hazelcast.map.MapStore;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
@@ -36,8 +38,16 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
+/**
+ * Persists engine IMaps to {@link IMapStorage}.
+ *
+ * <p>It is an {@link EntryStore} so that Hazelcast passes the expiration time
of every stored
+ * entry. Hazelcast does not delete an entry from the map store when it
expires, so the expiration
+ * time is stored with the entry and handed back on load; otherwise an entry
put with a TTL would be
+ * loaded again after a restart and never expire.
+ */
@Slf4j
-public class FileMapStore implements MapStore<Object, Object>,
MapLoaderLifecycleSupport {
+public class FileMapStore implements EntryStore<Object, Object>,
MapLoaderLifecycleSupport {
private IMapStorage mapStorage;
@@ -78,13 +88,19 @@ public class FileMapStore implements MapStore<Object,
Object>, MapLoaderLifecycl
}
@Override
- public void store(Object key, Object value) {
- mapStorage.store(key, value);
+ public void store(Object key, MetadataAwareValue<Object> value) {
+ mapStorage.store(key, value.getValue(), value.getExpirationTime());
}
@Override
- public void storeAll(Map<Object, Object> map) {
- mapStorage.storeAll(map);
+ public void storeAll(Map<Object, MetadataAwareValue<Object>> map) {
+ Map<Object, ExpirableValue> values = new HashMap<>(map.size());
+ map.forEach(
+ (key, value) ->
+ values.put(
+ key,
+ new ExpirableValue(value.getValue(),
value.getExpirationTime())));
+ mapStorage.storeAllWithExpirationTime(values);
}
@Override
@@ -99,16 +115,30 @@ public class FileMapStore implements MapStore<Object,
Object>, MapLoaderLifecycl
@SneakyThrows
@Override
- public Object load(Object key) {
+ public MetadataAwareValue<Object> load(Object key) {
return null;
}
+ /**
+ * Loads the requested keys with the expiration times they were stored
with, so entries put with
+ * a TTL keep expiring after a restart. Keys that are missing or already
expired are left out,
+ * and Hazelcast does not load them.
+ */
@SneakyThrows
@Override
- public Map<Object, Object> loadAll(Collection<Object> keys) {
- Map<Object, Object> allMap = mapStorage.loadAll();
- Map<Object, Object> retMap = new HashMap<>();
- keys.forEach(key -> retMap.put(key, allMap.get(key)));
+ public Map<Object, MetadataAwareValue<Object>> loadAll(Collection<Object>
keys) {
+ Map<Object, ExpirableValue> allMap =
mapStorage.loadAllWithExpirationTime();
+ Map<Object, MetadataAwareValue<Object>> retMap = new HashMap<>();
+ keys.forEach(
+ key -> {
+ ExpirableValue value = allMap.get(key);
+ if (value != null) {
+ retMap.put(
+ key,
+ new MetadataAwareValue<>(
+ value.getValue(),
value.getExpirationTime()));
+ }
+ });
return Collections.unmodifiableMap(retMap);
}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/persistence/FileMapStoreFactory.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/persistence/FileMapStoreFactory.java
index 2022e0fcc5..8ae25228ea 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/persistence/FileMapStoreFactory.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/persistence/FileMapStoreFactory.java
@@ -17,14 +17,16 @@
package org.apache.seatunnel.engine.server.persistence;
+import com.hazelcast.map.EntryLoader.MetadataAwareValue;
import com.hazelcast.map.MapLoader;
import com.hazelcast.map.MapStoreFactory;
import java.util.Properties;
-public class FileMapStoreFactory implements MapStoreFactory<Object, Object> {
+public class FileMapStoreFactory implements MapStoreFactory<Object,
MetadataAwareValue<Object>> {
@Override
- public MapLoader<Object, Object> newMapStore(String mapName, Properties
properties) {
+ public MapLoader<Object, MetadataAwareValue<Object>> newMapStore(
+ String mapName, Properties properties) {
properties.setProperty("businessName", mapName);
return new FileMapStore();
}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/persistence/FileMapStoreTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/persistence/FileMapStoreTest.java
index a1067c4f58..d6d2491fdb 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/persistence/FileMapStoreTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/persistence/FileMapStoreTest.java
@@ -17,6 +17,7 @@
package org.apache.seatunnel.engine.server.persistence;
+import org.apache.seatunnel.engine.common.Constant;
import org.apache.seatunnel.engine.imap.storage.file.common.FileConstants;
import
org.apache.seatunnel.engine.server.common.statestore.EngineStateStoreNames;
@@ -26,12 +27,22 @@ import org.junit.jupiter.api.condition.EnabledOnOs;
import org.junit.jupiter.api.condition.OS;
import org.junit.jupiter.api.io.TempDir;
+import com.hazelcast.config.Config;
+import com.hazelcast.config.JoinConfig;
+import com.hazelcast.config.MapConfig;
+import com.hazelcast.config.MapStoreConfig;
+import com.hazelcast.core.EntryView;
+import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.map.EntryLoader.MetadataAwareValue;
+import com.hazelcast.map.IMap;
import java.nio.file.Path;
import java.util.Collections;
import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+import static org.awaitility.Awaitility.await;
import static org.mockito.Mockito.mock;
/**
@@ -116,12 +127,75 @@ public class FileMapStoreTest {
props,
EngineStateStoreNames.RUNNING_JOB_METRICS));
Assertions.assertEquals(Collections.emptySet(), store.loadAllKeys());
- Assertions.assertDoesNotThrow(() -> store.store("k", "v"));
+ Assertions.assertDoesNotThrow(() -> store.store("k", new
MetadataAwareValue<>("v")));
Assertions.assertDoesNotThrow(() -> store.delete("k"));
store.destroy();
}
+ /**
+ * Hazelcast does not delete an entry from the map store when its TTL
expires. Verifies that
+ * after a restart an expired entry is not loaded again, and an entry put
with a TTL keeps its
+ * expiration time instead of never expiring.
+ */
+ @Test
+ public void testEntriesKeepExpirationTimeAcrossRestart() {
+ String clusterName = "file-map-store-expiration-" + System.nanoTime();
+ String mapName = Constant.IMAP_FINISHED_JOB_STATE;
+ long expirationTimeBeforeRestart;
+
+ HazelcastInstance writer =
Hazelcast.newHazelcastInstance(hazelcastConfig(clusterName));
+ try {
+ IMap<Long, String> map = writer.getMap(mapName);
+ map.put(1L, "expired", 1, TimeUnit.SECONDS);
+ map.put(2L, "expiring", 10, TimeUnit.MINUTES);
+ map.put(3L, "never-expiring");
+ expirationTimeBeforeRestart =
map.getEntryView(2L).getExpirationTime();
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
!map.containsKey(1L));
+ } finally {
+ writer.shutdown();
+ }
+
+ HazelcastInstance restarted =
Hazelcast.newHazelcastInstance(hazelcastConfig(clusterName));
+ try {
+ IMap<Long, String> map = restarted.getMap(mapName);
+
+ Assertions.assertNull(map.getEntryView(1L));
+ Assertions.assertEquals(2, map.size());
+
+ EntryView<Long, String> expiring = map.getEntryView(2L);
+ Assertions.assertEquals("expiring", expiring.getValue());
+ // Hazelcast keeps record times at second granularity.
+ long drift = Math.abs(expiring.getExpirationTime() -
expirationTimeBeforeRestart);
+ Assertions.assertTrue(
+ drift <= TimeUnit.SECONDS.toMillis(2),
+ "expiration time moved by " + drift + " ms across the
restart");
+
+ EntryView<Long, String> neverExpiring = map.getEntryView(3L);
+ Assertions.assertEquals("never-expiring",
neverExpiring.getValue());
+ Assertions.assertEquals(Long.MAX_VALUE,
neverExpiring.getExpirationTime());
+ } finally {
+ restarted.shutdown();
+ }
+ }
+
+ private Config hazelcastConfig(String clusterName) {
+ MapStoreConfig mapStoreConfig =
+ new MapStoreConfig()
+ .setEnabled(true)
+
.setInitialLoadMode(MapStoreConfig.InitialLoadMode.EAGER)
+
.setFactoryClassName(FileMapStoreFactory.class.getName())
+
.setProperties(buildLocalFsProperties(tempDir.toString()));
+
+ Config config = new Config();
+ config.setClusterName(clusterName);
+ JoinConfig join = config.getNetworkConfig().getJoin();
+ join.getMulticastConfig().setEnabled(false);
+ join.getAutoDetectionConfig().setEnabled(false);
+ config.addMapConfig(new
MapConfig("engine*").setMapStoreConfig(mapStoreConfig));
+ return config;
+ }
+
private Properties buildLocalFsProperties(String namespace) {
Properties props = new Properties();
// "hdfs" is the factoryIdentifier of IMapFileStorageFactory — covers
local fs via
diff --git
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-api/src/main/java/org/apache/seatunnel/engine/imap/storage/api/ExpirableValue.java
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-api/src/main/java/org/apache/seatunnel/engine/imap/storage/api/ExpirableValue.java
new file mode 100644
index 0000000000..efe12174b9
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-api/src/main/java/org/apache/seatunnel/engine/imap/storage/api/ExpirableValue.java
@@ -0,0 +1,70 @@
+/*
+ * 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.imap.storage.api;
+
+import java.util.Objects;
+
+/** A stored value together with the time at which its entry expires. */
+public final class ExpirableValue {
+
+ /** Expiration time of an entry that never expires. */
+ public static final long NO_EXPIRATION_TIME = Long.MAX_VALUE;
+
+ private final Object value;
+
+ private final long expirationTime;
+
+ public ExpirableValue(Object value) {
+ this(value, NO_EXPIRATION_TIME);
+ }
+
+ /**
+ * @param value stored value
+ * @param expirationTime epoch milliseconds at which the entry expires, or
{@link
+ * #NO_EXPIRATION_TIME}
+ */
+ public ExpirableValue(Object value, long expirationTime) {
+ this.value = Objects.requireNonNull(value, "value");
+ this.expirationTime = expirationTime;
+ }
+
+ public Object getValue() {
+ return value;
+ }
+
+ public long getExpirationTime() {
+ return expirationTime;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof ExpirableValue)) {
+ return false;
+ }
+ ExpirableValue that = (ExpirableValue) o;
+ return expirationTime == that.expirationTime && Objects.equals(value,
that.value);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(value, expirationTime);
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-api/src/main/java/org/apache/seatunnel/engine/imap/storage/api/IMapStorage.java
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-api/src/main/java/org/apache/seatunnel/engine/imap/storage/api/IMapStorage.java
index f4c6a8f0bb..34c31425fc 100644
---
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-api/src/main/java/org/apache/seatunnel/engine/imap/storage/api/IMapStorage.java
+++
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-api/src/main/java/org/apache/seatunnel/engine/imap/storage/api/IMapStorage.java
@@ -22,6 +22,7 @@ package org.apache.seatunnel.engine.imap.storage.api;
import java.io.IOException;
import java.util.Collection;
+import java.util.HashMap;
import java.util.Map;
import java.util.Set;
@@ -47,6 +48,36 @@ public interface IMapStorage {
*/
public Set<Object> storeAll(Map<Object, Object> map);
+ /**
+ * Store a key-value pair that expires at the given time.
+ *
+ * <p>Storages that do not persist expiration times store the pair without
it, so the entry
+ * never expires after it is loaded again.
+ *
+ * @param key storage key
+ * @param value storage value
+ * @param expirationTime epoch milliseconds at which the entry expires, or
{@link
+ * ExpirableValue#NO_EXPIRATION_TIME}
+ * @return storage status, true is success, false is fail
+ */
+ default boolean store(Object key, Object value, long expirationTime) {
+ return store(key, value);
+ }
+
+ /**
+ * Store key-value pairs together with the times at which they expire.
+ *
+ * @param map storage key to value and expiration time
+ * @return if some key-value pair is not stored, return this keys; if all
key-value pair is
+ * stored, return empty set.
+ * @see #store(Object, Object, long)
+ */
+ default Set<Object> storeAllWithExpirationTime(Map<Object, ExpirableValue>
map) {
+ Map<Object, Object> values = new HashMap<>(map.size());
+ map.forEach((key, value) -> values.put(key, value.getValue()));
+ return storeAll(values);
+ }
+
/**
* Delete a key in the map storage.
*
@@ -66,6 +97,22 @@ public interface IMapStorage {
public Map<Object, Object> loadAll() throws IOException;
+ /**
+ * Load all entries together with the times at which they expire. Entries
that have already
+ * expired are not returned.
+ *
+ * <p>Storages that do not persist expiration times return every entry
with {@link
+ * ExpirableValue#NO_EXPIRATION_TIME}.
+ *
+ * @return storage key to value and expiration time
+ */
+ default Map<Object, ExpirableValue> loadAllWithExpirationTime() throws
IOException {
+ Map<Object, Object> values = loadAll();
+ Map<Object, ExpirableValue> result = new HashMap<>(values.size());
+ values.forEach((key, value) -> result.put(key, new
ExpirableValue(value)));
+ return result;
+ }
+
public Set<Object> loadAllKeys();
public void destroy(boolean deleteAllFileFlag);
diff --git
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/IMapFileStorage.java
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/IMapFileStorage.java
index ab125ba5c8..64b1692397 100644
---
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/IMapFileStorage.java
+++
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/IMapFileStorage.java
@@ -20,6 +20,7 @@
package org.apache.seatunnel.engine.imap.storage.file;
+import org.apache.seatunnel.engine.imap.storage.api.ExpirableValue;
import org.apache.seatunnel.engine.imap.storage.api.IMapStorage;
import
org.apache.seatunnel.engine.imap.storage.api.exception.IMapStorageException;
import org.apache.seatunnel.engine.imap.storage.file.bean.IMapFileData;
@@ -169,9 +170,14 @@ public class IMapFileStorage implements IMapStorage {
@Override
public boolean store(Object key, Object value) {
+ return store(key, value, ExpirableValue.NO_EXPIRATION_TIME);
+ }
+
+ @Override
+ public boolean store(Object key, Object value, long expirationTime) {
IMapFileData data;
try {
- data = parseToIMapFileData(key, value);
+ data = parseToIMapFileData(key, value, expirationTime);
} catch (IOException e) {
log.error("parse to IMapFileData error, key is {}, value is {}",
key, value, e);
return false;
@@ -183,12 +189,21 @@ public class IMapFileStorage implements IMapStorage {
@Override
public Set<Object> storeAll(Map<Object, Object> map) {
+ Map<Object, ExpirableValue> values = new HashMap<>(map.size());
+ map.forEach((key, value) -> values.put(key, new
ExpirableValue(value)));
+ return storeAllWithExpirationTime(values);
+ }
+
+ @Override
+ public Set<Object> storeAllWithExpirationTime(Map<Object, ExpirableValue>
map) {
Map<Long, Object> requestMap = new HashMap<>(map.size());
Set<Object> failures = new HashSet<>();
map.forEach(
(key, value) -> {
try {
- IMapFileData data = parseToIMapFileData(key, value);
+ IMapFileData data =
+ parseToIMapFileData(
+ key, value.getValue(),
value.getExpirationTime());
long requestId = sendToDisruptorQueue(data,
WALEventType.APPEND);
requestMap.put(requestId, key);
} catch (IOException e) {
@@ -241,6 +256,17 @@ public class IMapFileStorage implements IMapStorage {
}
}
+ @Override
+ public Map<Object, ExpirableValue> loadAllWithExpirationTime() {
+ try {
+ WALReader reader = new WALReader(fs, fileConfiguration,
serializer);
+ return reader.loadAllDataWithExpirationTime(
+ new Path(businessRootPath), new HashSet<>());
+ } catch (IOException e) {
+ throw new IMapStorageException("load all data error", e);
+ }
+ }
+
@Override
public Set<Object> loadAllKeys() {
try {
@@ -283,7 +309,8 @@ public class IMapFileStorage implements IMapStorage {
}
}
- private IMapFileData parseToIMapFileData(Object key, Object value) throws
IOException {
+ private IMapFileData parseToIMapFileData(Object key, Object value, long
expirationTime)
+ throws IOException {
return IMapFileData.builder()
.key(serializer.serialize(key))
.keyClassName(key.getClass().getName())
@@ -291,6 +318,9 @@ public class IMapFileStorage implements IMapStorage {
.valueClassName(value.getClass().getName())
.timestamp(System.currentTimeMillis())
.deleted(false)
+ // Leave the field unset for entries that never expire, as
older releases did.
+ .expirationTime(
+ expirationTime == ExpirableValue.NO_EXPIRATION_TIME ?
null : expirationTime)
.build();
}
diff --git
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/bean/IMapFileData.java
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/bean/IMapFileData.java
index e8920f86fb..46e3164e19 100644
---
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/bean/IMapFileData.java
+++
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/bean/IMapFileData.java
@@ -44,6 +44,15 @@ public class IMapFileData implements Serializable,
Comparable<IMapFileData> {
private long timestamp;
+ /**
+ * Epoch milliseconds at which the entry expires, or {@code null} if it
never expires.
+ *
+ * <p>ProtoStuff numbers fields by declaration order, so this field must
stay last to keep WAL
+ * files written before it was added readable. Those records read it as
{@code null}, and
+ * releases that do not know this field skip it.
+ */
+ private Long expirationTime;
+
@Override
public int compareTo(IMapFileData o) {
return IMapDataComparator.compare(
diff --git
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/common/WALReader.java
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/common/WALReader.java
index f69d0cfb38..8be32aa388 100644
---
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/common/WALReader.java
+++
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/common/WALReader.java
@@ -22,6 +22,7 @@ package org.apache.seatunnel.engine.imap.storage.file.common;
import org.apache.seatunnel.shade.org.apache.commons.lang3.ClassUtils;
+import org.apache.seatunnel.engine.imap.storage.api.ExpirableValue;
import
org.apache.seatunnel.engine.imap.storage.api.exception.IMapStorageException;
import org.apache.seatunnel.engine.imap.storage.file.bean.IMapFileData;
import org.apache.seatunnel.engine.imap.storage.file.config.FileConfiguration;
@@ -39,6 +40,7 @@ import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
+import java.util.function.BiFunction;
import static
org.apache.seatunnel.engine.imap.storage.file.common.LatestMutationAccumulator.SerializedKey;
@@ -85,40 +87,81 @@ public class WALReader {
this.fileReader.initialize(fs, serializer);
}
+ /** Returns the keys whose latest mutation is neither a delete nor
expired. */
public Set<Object> loadAllKeys(Path parentPath) throws IOException {
LatestMutationAccumulator accumulator =
readLatestMutations(parentPath, null);
+ long now = System.currentTimeMillis();
Set<Object> result = new HashSet<>(accumulator.size());
Iterator<IMapFileData> mutations = accumulator.iterator();
while (mutations.hasNext()) {
IMapFileData mutation = mutations.next();
mutations.remove();
- if (!mutation.isDeleted()) {
+ if (isLive(mutation, now)) {
result.add(deserializeData(mutation.getKey(),
mutation.getKeyClassName()));
}
}
return result;
}
+ /** Returns the latest value of each key that is neither deleted nor
expired. */
public Map<Object, Object> loadAllData(Path parentPath, Set<Object>
searchKeys)
throws IOException {
+ return loadLiveData(parentPath, searchKeys, (value, mutation) ->
value);
+ }
+
+ /**
+ * Returns the latest value of each key that is neither deleted nor
expired, together with the
+ * time at which it expires.
+ */
+ public Map<Object, ExpirableValue> loadAllDataWithExpirationTime(
+ Path parentPath, Set<Object> searchKeys) throws IOException {
+ return loadLiveData(
+ parentPath,
+ searchKeys,
+ (value, mutation) ->
+ new ExpirableValue(
+ value,
+ mutation.getExpirationTime() == null
+ ? ExpirableValue.NO_EXPIRATION_TIME
+ : mutation.getExpirationTime()));
+ }
+
+ private <V> Map<Object, V> loadLiveData(
+ Path parentPath,
+ Set<Object> searchKeys,
+ BiFunction<Object, IMapFileData, V> valueMapper)
+ throws IOException {
Set<SerializedKey> serializedSearchKeys =
serializeSearchKeys(searchKeys);
LatestMutationAccumulator accumulator =
readLatestMutations(parentPath, serializedSearchKeys);
- Map<Object, Object> result = new HashMap<>(accumulator.size());
+ long now = System.currentTimeMillis();
+ Map<Object, V> result = new HashMap<>(accumulator.size());
Iterator<IMapFileData> mutations = accumulator.iterator();
while (mutations.hasNext()) {
IMapFileData mutation = mutations.next();
mutations.remove();
- if (mutation.isDeleted()) {
+ if (!isLive(mutation, now)) {
continue;
}
Object key = deserializeData(mutation.getKey(),
mutation.getKeyClassName());
Object value = deserializeData(mutation.getValue(),
mutation.getValueClassName());
- result.put(key, value);
+ result.put(key, valueMapper.apply(value, mutation));
}
return result;
}
+ /**
+ * An entry stays in the WAL after it expires because Hazelcast does not
delete expired entries
+ * from the map store, so its latest put is skipped here instead of being
loaded again.
+ */
+ private static boolean isLive(IMapFileData mutation, long now) {
+ if (mutation.isDeleted()) {
+ return false;
+ }
+ Long expirationTime = mutation.getExpirationTime();
+ return expirationTime == null || expirationTime > now;
+ }
+
private LatestMutationAccumulator readLatestMutations(
Path parentPath, Set<SerializedKey> serializedSearchKeys) throws
IOException {
LatestMutationAccumulator accumulator = new
LatestMutationAccumulator(serializedSearchKeys);
diff --git
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/IMapFileStorageTest.java
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/IMapFileStorageTest.java
index a1aebb5e47..73409654b7 100644
---
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/IMapFileStorageTest.java
+++
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/IMapFileStorageTest.java
@@ -20,6 +20,7 @@
package org.apache.seatunnel.engine.imap.storage.file;
+import org.apache.seatunnel.engine.imap.storage.api.ExpirableValue;
import org.apache.seatunnel.engine.imap.storage.file.common.FileConstants;
import org.apache.hadoop.conf.Configuration;
@@ -33,6 +34,7 @@ import org.junit.jupiter.api.condition.EnabledOnOs;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -108,6 +110,29 @@ public class IMapFileStorageTest {
STORAGE.deleteAll(keys);
}
+ @Test
+ void testStoreWithExpirationTime() {
+ long expirationTime = System.currentTimeMillis() +
TimeUnit.MINUTES.toMillis(10);
+ STORAGE.store("expiring", 1L, expirationTime);
+ STORAGE.store("expired", 2L, System.currentTimeMillis() - 1);
+ Map<Object, ExpirableValue> batch = new HashMap<>();
+ batch.put("batch-expiring", new ExpirableValue(3L, expirationTime));
+ batch.put("batch-never-expiring", new ExpirableValue(4L));
+
Assertions.assertTrue(STORAGE.storeAllWithExpirationTime(batch).isEmpty());
+
+ Map<Object, ExpirableValue> loaded =
STORAGE.loadAllWithExpirationTime();
+ Assertions.assertEquals(new ExpirableValue(1L, expirationTime),
loaded.get("expiring"));
+ Assertions.assertEquals(
+ new ExpirableValue(3L, expirationTime),
loaded.get("batch-expiring"));
+ Assertions.assertEquals(new ExpirableValue(4L),
loaded.get("batch-never-expiring"));
+ Assertions.assertFalse(loaded.containsKey("expired"));
+ Assertions.assertFalse(STORAGE.loadAll().containsKey("expired"));
+ Assertions.assertFalse(STORAGE.loadAllKeys().contains("expired"));
+
+ STORAGE.deleteAll(
+ Arrays.asList("expiring", "expired", "batch-expiring",
"batch-never-expiring"));
+ }
+
@Test
void testStoreArray() {
Long[] data = new Long[10];
diff --git
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/common/WALReaderAndWriterTest.java
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/common/WALReaderAndWriterTest.java
index 2eeb243669..d4c187f181 100644
---
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/common/WALReaderAndWriterTest.java
+++
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/common/WALReaderAndWriterTest.java
@@ -23,6 +23,7 @@ package org.apache.seatunnel.engine.imap.storage.file.common;
import org.apache.seatunnel.engine.common.job.JobResult;
import org.apache.seatunnel.engine.common.job.JobStatus;
import org.apache.seatunnel.engine.common.job.JobStatusData;
+import org.apache.seatunnel.engine.imap.storage.api.ExpirableValue;
import
org.apache.seatunnel.engine.imap.storage.api.exception.IMapStorageException;
import org.apache.seatunnel.engine.imap.storage.file.bean.IMapFileData;
import org.apache.seatunnel.engine.imap.storage.file.config.FileConfiguration;
@@ -44,9 +45,11 @@ import org.junit.jupiter.api.condition.EnabledOnOs;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.TimeUnit;
import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.condition.OS.LINUX;
@@ -194,6 +197,95 @@ public class WALReaderAndWriterTest {
reader.loadAllData(path, Collections.emptySet()));
}
+ @Test
+ void shouldSkipExpiredEntriesAndReturnExpirationTimes() throws IOException
{
+ Path path = new Path(PARENT_PATH, "expiration");
+ long now = System.currentTimeMillis();
+ long liveExpirationTime = now + TimeUnit.MINUTES.toMillis(10);
+ IMapFileData expired = mutation("expired", "value-a", 10, false);
+ expired.setExpirationTime(now - 1);
+ IMapFileData live = mutation("live", "value-b", 10, false);
+ live.setExpirationTime(liveExpirationTime);
+ // Records written before the expiration time was stored have none.
+ IMapFileData neverExpiring = mutation("never-expiring", "value-c", 10,
false);
+ writeWalFile(new Path(path, "region/expiration-wal.txt"), expired,
live, neverExpiring);
+
+ WALReader reader = new WALReader(FS, FileConfiguration.HDFS,
SERIALIZER);
+
+ Assertions.assertEquals(
+ new HashSet<>(Arrays.asList("live", "never-expiring")),
reader.loadAllKeys(path));
+ Map<Object, Object> expectedData = new HashMap<>();
+ expectedData.put("live", "value-b");
+ expectedData.put("never-expiring", "value-c");
+ Assertions.assertEquals(expectedData, reader.loadAllData(path,
Collections.emptySet()));
+ Map<Object, ExpirableValue> expectedDataWithExpirationTime = new
HashMap<>();
+ expectedDataWithExpirationTime.put(
+ "live", new ExpirableValue("value-b", liveExpirationTime));
+ expectedDataWithExpirationTime.put("never-expiring", new
ExpirableValue("value-c"));
+ Assertions.assertEquals(
+ expectedDataWithExpirationTime,
+ reader.loadAllDataWithExpirationTime(path,
Collections.emptySet()));
+ }
+
+ @Test
+ void shouldNotFallBackToOlderValueWhenLatestValueExpired() throws
IOException {
+ Path path = new Path(PARENT_PATH, "latest-expired");
+ IMapFileData expired = mutation("key", "new", 20, false);
+ expired.setExpirationTime(System.currentTimeMillis() - 1);
+ writeWalFile(
+ new Path(path, "region/latest-expired-wal.txt"),
+ mutation("key", "old", 10, false),
+ expired);
+
+ WALReader reader = new WALReader(FS, FileConfiguration.HDFS,
SERIALIZER);
+
+ Assertions.assertEquals(
+ Collections.emptyMap(), reader.loadAllData(path,
Collections.emptySet()));
+ Assertions.assertEquals(Collections.emptySet(),
reader.loadAllKeys(path));
+ }
+
+ @Test
+ void shouldLoadRecordsWrittenBeforeExpirationTimeWasStored() throws
IOException {
+ Path path = new Path(PARENT_PATH, "pre-expiration-time");
+ Path file = new Path(path, "region/pre-expiration-time-wal.txt");
+ FS.mkdirs(file.getParent());
+ try (FSDataOutputStream output = FS.create(file, true)) {
+ output.write(
+ WALDataUtils.wrapperBytes(
+
SERIALIZER.serialize(preExpirationTimeRecord("key", "value", 10))));
+ }
+
+ WALReader reader = new WALReader(FS, FileConfiguration.HDFS,
SERIALIZER);
+
+ Assertions.assertEquals(
+ Collections.singletonMap("key", new ExpirableValue("value")),
+ reader.loadAllDataWithExpirationTime(path,
Collections.emptySet()));
+ }
+
+ @Test
+ void shouldWriteRecordsWithoutExpirationTimeInThePreviousFormat() throws
IOException {
+ Assertions.assertArrayEquals(
+ SERIALIZER.serialize(preExpirationTimeRecord("key", "value",
10)),
+ SERIALIZER.serialize(mutation("key", "value", 10, false)));
+ }
+
+ @Test
+ void shouldKeepRecordsWithExpirationTimeReadableByPreviousReleases()
throws IOException {
+ IMapFileData record = mutation("key", "value", 10, false);
+ record.setExpirationTime(1_000L);
+
+ PreExpirationTimeIMapFileData read =
+ SERIALIZER.deserialize(
+ SERIALIZER.serialize(record),
PreExpirationTimeIMapFileData.class);
+
+ Assertions.assertFalse(read.deleted);
+ Assertions.assertArrayEquals(record.getKey(), read.key);
+ Assertions.assertEquals(record.getKeyClassName(), read.keyClassName);
+ Assertions.assertArrayEquals(record.getValue(), read.value);
+ Assertions.assertEquals(record.getValueClassName(),
read.valueClassName);
+ Assertions.assertEquals(record.getTimestamp(), read.timestamp);
+ }
+
private static IMapFileData mutation(String key, String value, long
timestamp, boolean deleted)
throws IOException {
return IMapFileData.builder()
@@ -219,6 +311,31 @@ public class WALReaderAndWriterTest {
return WALDataUtils.wrapperBytes(SERIALIZER.serialize(mutation));
}
+ private static PreExpirationTimeIMapFileData preExpirationTimeRecord(
+ String key, String value, long timestamp) throws IOException {
+ PreExpirationTimeIMapFileData record = new
PreExpirationTimeIMapFileData();
+ record.key = SERIALIZER.serialize(key);
+ record.keyClassName = String.class.getName();
+ record.value = SERIALIZER.serialize(value);
+ record.valueClassName = String.class.getName();
+ record.timestamp = timestamp;
+ return record;
+ }
+
+ /**
+ * Field layout of {@link IMapFileData} before {@code expirationTime} was
added. ProtoStuff
+ * numbers fields by declaration order, so this class reads and writes the
same bytes as earlier
+ * releases did.
+ */
+ public static class PreExpirationTimeIMapFileData {
+ private boolean deleted;
+ private byte[] key;
+ private String keyClassName;
+ private byte[] value;
+ private String valueClassName;
+ private long timestamp;
+ }
+
@Test
public void testReplayKeepsTombstoneForSameKeyAndTimestamp() throws
Exception {
String key = "deleted-key";
