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