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-12188-271f73324dc4ef26f391925d6251009098717850 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit efc81bbb10102f0f4d8c8dec5040be4564221560 Author: Goutam Adwant <[email protected]> AuthorDate: Tue Sep 8 05:28:59 2026 +0000 [Test][Zeta] Bound finished DAG store benchmark verification (#12188) Signed-off-by: goutamadwant <[email protected]> --- docs/en/engines/zeta/benchmark.md | 16 +- docs/zh/engines/zeta/benchmark.md | 14 +- .../benchmark/IMapDagStorageBenchmark.java | 6 +- .../imap/IMapDagStorageBenchmarkWorkload.java | 24 +- .../imap/IMapDagStorageBenchmarkWorkloadTest.java | 262 +++++++++++++++++++++ .../imap/IMapDagStoragePersistenceTest.java | 120 ++++++++++ 6 files changed, 432 insertions(+), 10 deletions(-) diff --git a/docs/en/engines/zeta/benchmark.md b/docs/en/engines/zeta/benchmark.md index 27ff7193bc..ed760947cf 100644 --- a/docs/en/engines/zeta/benchmark.md +++ b/docs/en/engines/zeta/benchmark.md @@ -249,7 +249,7 @@ java -jar seatunnel-benchmarks/target/benchmarks.jar \ ### Run the IMap DAG Storage Benchmarks ```bash -java -jar seatunnel-benchmarks/target/benchmarks.jar IMapDagStorageBenchmark +java -jar seatunnel-benchmarks/target/benchmarks.jar IMapDagStorageBenchmark -foe true ``` `finishedJobDagStore` writes a fixed batch of 100 unique production `JobDAGInfo` values through @@ -257,6 +257,20 @@ the finished-job DAG IMap and its file-backed MapStore. `finishedJobDagLoad` evi reloads it through MapStore. `pipelineCount=1|10|100` controls the exact number of code-built source-to-sink pipelines in each DAG, and `storedDagCount=0|100` controls retained storage pressure. +Store teardown checks the first, middle, and last cached values and deletes the batch outside +timing. After the final measurement in each fork, those samples are evicted and reloaded from +MapStore before deletion. Intermediate iterations do not replay the WAL for verification: +full-WAL scans allocate more as writes and deletion records accumulate, affecting later samples +even though teardown is not timed. WAL history still grows through the unchanged writes and +deletes; this is not a storage compaction or steady-state benchmark. + +Only the final batch has a persistence read-back in each benchmark fork. Dedicated tests cover +complete batch round trips. Local-file read-back establishes visible persisted contents, not +crash durability. Results from the former per-iteration reload fixture are not directly comparable +to this corrected baseline and must not be reported as a production speedup. +Use `-foe true` so a verification failure rejects the run. Allocation/GC profiler results may +include untimed teardown, including the terminal reload; they are not write-only measurements. + ```bash java -jar seatunnel-benchmarks/target/benchmarks.jar \ 'IMapDagStorageBenchmark.finishedJobDagLoad$' \ diff --git a/docs/zh/engines/zeta/benchmark.md b/docs/zh/engines/zeta/benchmark.md index 9338be814e..754e20f96e 100644 --- a/docs/zh/engines/zeta/benchmark.md +++ b/docs/zh/engines/zeta/benchmark.md @@ -237,7 +237,7 @@ java -jar seatunnel-benchmarks/target/benchmarks.jar \ ### 运行 IMap DAG Storage Benchmark ```bash -java -jar seatunnel-benchmarks/target/benchmarks.jar IMapDagStorageBenchmark +java -jar seatunnel-benchmarks/target/benchmarks.jar IMapDagStorageBenchmark -foe true ``` `finishedJobDagStore` 通过已完成作业 DAG IMap 及其文件型 MapStore 固定写入 100 个唯一的 @@ -245,6 +245,18 @@ java -jar seatunnel-benchmarks/target/benchmarks.jar IMapDagStorageBenchmark `pipelineCount=1|10|100` 控制代码构建的 Source-to-Sink Pipeline 数量， `storedDagCount=0|100` 控制已有存储压力。 +Store 的 TearDown 在计时之外检查缓存中的首个、中间和最后一个值，并删除本批数据。 +每个 Fork 的最后一次测量结束后，会先逐出这些样本，再通过 MapStore 重新加载并校验， +然后删除数据。中间迭代不再为校验重放 WAL：随着写入和删除记录累积，全量 WAL 扫描的 +分配量会增长，即使 TearDown 不计时，也会影响后续样本。原有写入和删除操作保持不变， +WAL 历史仍会增长；这不是存储压缩或稳态 Benchmark。 + +每个 Benchmark Fork 仅对最后一批数据进行持久化回读，独立测试覆盖完整批次的往返校验。 +本地文件回读只能证明持久化内容可见，不能证明崩溃后的持久性。旧版每次迭代都回读的 +结果不能与修正后的基线直接比较，也不能将差异描述为生产性能提升。 +使用 `-foe true` 确保校验失败时拒绝本次运行。分配量和 GC Profiler 的结果可能包含不计时的 +TearDown，包括最后的回读，因此不能将它们视为仅包含写入的指标。 + ```bash java -jar seatunnel-benchmarks/target/benchmarks.jar \ 'IMapDagStorageBenchmark.finishedJobDagLoad$' \ diff --git a/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/IMapDagStorageBenchmark.java b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/IMapDagStorageBenchmark.java index cf93d88b13..161cad7ddf 100644 --- a/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/IMapDagStorageBenchmark.java +++ b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/IMapDagStorageBenchmark.java @@ -68,8 +68,10 @@ public class IMapDagStorageBenchmark extends BenchmarkBase { * <p>The workload builds a real {@code JobDAGInfo} containing 1, 10, or 100 source-to-sink * pipelines before measurement and starts with 0 or 100 retained DAGs. The timed phase writes * that payload under 100 unique job IDs using the production history TTL, exercising IMap - * serialization, FileMapStore, and WAL append. Fixture construction, durable sample reloads, - * and deletion of the measured phase are not timed. + * serialization, FileMapStore, and WAL append. Fixture construction, sample checks, and + * deletion of the measured phase are not timed. Intermediate iterations check cached values; + * only the final measurement reloads samples from MapStore, avoiding full-WAL verification + * scans between measurements. This read-back checks persisted contents, not crash durability. * * <p>{@link Mode#SingleShotTime} fixes each growth phase at 100 DAGs. {@link * OperationsPerInvocation} normalizes the phase duration to one persisted DAG and prevents a diff --git a/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/storage/imap/IMapDagStorageBenchmarkWorkload.java b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/storage/imap/IMapDagStorageBenchmarkWorkload.java index d57823579e..f94b109523 100644 --- a/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/storage/imap/IMapDagStorageBenchmarkWorkload.java +++ b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/storage/imap/IMapDagStorageBenchmarkWorkload.java @@ -28,6 +28,8 @@ import org.openjdk.jmh.annotations.Scope; import org.openjdk.jmh.annotations.Setup; import org.openjdk.jmh.annotations.State; import org.openjdk.jmh.annotations.TearDown; +import org.openjdk.jmh.infra.IterationParams; +import org.openjdk.jmh.runner.IterationType; import com.hazelcast.map.IMap; @@ -60,10 +62,13 @@ public class IMapDagStorageBenchmarkWorkload { private long[] storeKeys; private int storedDagCountInIteration; private boolean loadExecuted; + private int measurementIteration; + private boolean lastMeasurementIteration; /** Constructs an exact number of production source-to-sink DAG pipelines. */ @Setup(Level.Trial) public void setUp(SeaTunnelStorageEnvironmentContext environment) { + measurementIteration = 0; finishedJobDagMap = environment .getServer() @@ -85,7 +90,10 @@ public class IMapDagStorageBenchmarkWorkload { /** Builds a fresh, fixed-size key set before each measured DAG-store phase. */ @Setup(Level.Iteration) - public void prepareStoreIteration() { + public void prepareStoreIteration(IterationParams iterationParams) { + lastMeasurementIteration = + iterationParams.getType() == IterationType.MEASUREMENT + && ++measurementIteration == iterationParams.getCount(); storeKeys = new long[STORE_OPERATIONS_PER_INVOCATION]; for (int index = 0; index < STORE_OPERATIONS_PER_INVOCATION; index++) { storeKeys[index] = Long.MAX_VALUE - sequence.incrementAndGet(); @@ -111,7 +119,7 @@ public class IMapDagStorageBenchmarkWorkload { } } - /** Validates durable samples and removes the fixed store batch outside measured time. */ + /** Checks sampled values and removes the fixed store batch outside measured time. */ @TearDown(Level.Iteration) public void cleanStoreIteration() { if (storedDagCountInIteration == 0) { @@ -157,15 +165,19 @@ public class IMapDagStorageBenchmarkWorkload { sampledKeys.add(storeKeys[0]); sampledKeys.add(storeKeys[STORE_OPERATIONS_PER_INVOCATION / 2]); sampledKeys.add(storeKeys[STORE_OPERATIONS_PER_INVOCATION - 1]); - for (long sampledKey : sampledKeys) { - finishedJobDagMap.evict(sampledKey); + // MapStore reloads replay the full, growing WAL. Only do this after the last + // measurement so verification allocations cannot interfere with a later sample. + if (lastMeasurementIteration) { + for (long sampledKey : sampledKeys) { + finishedJobDagMap.evict(sampledKey); + } + finishedJobDagMap.loadAll(sampledKeys, true); } - finishedJobDagMap.loadAll(sampledKeys, true); for (long sampledKey : sampledKeys) { JobDAGInfo stored = finishedJobDagMap.get(sampledKey); if (!finishedJobDag.equals(stored)) { throw new IllegalStateException( - "The JobDAGInfo append was not durably persisted for key " + sampledKey); + "The stored JobDAGInfo did not match for key " + sampledKey); } } } diff --git a/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/storage/imap/IMapDagStorageBenchmarkWorkloadTest.java b/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/storage/imap/IMapDagStorageBenchmarkWorkloadTest.java new file mode 100644 index 0000000000..dacd20dc29 --- /dev/null +++ b/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/storage/imap/IMapDagStorageBenchmarkWorkloadTest.java @@ -0,0 +1,262 @@ +/* + * 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.benchmark.storage.imap; + +import org.apache.seatunnel.benchmark.dag.JobDagFixtureFactory; +import org.apache.seatunnel.benchmark.storage.SeaTunnelStorageEnvironmentContext; +import org.apache.seatunnel.engine.common.Constant; +import org.apache.seatunnel.engine.core.job.JobDAGInfo; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.ValueSource; +import org.openjdk.jmh.infra.IterationParams; +import org.openjdk.jmh.runner.IterationType; +import org.openjdk.jmh.runner.options.TimeValue; + +import com.hazelcast.map.IMap; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anySet; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.RETURNS_DEEP_STUBS; +import static org.mockito.Mockito.clearInvocations; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class IMapDagStorageBenchmarkWorkloadTest { + + private final Map<Long, JobDAGInfo> cached = new HashMap<>(); + private final Map<Long, JobDAGInfo> persisted = new HashMap<>(); + private final List<Long> writtenKeys = new ArrayList<>(); + private IMap<Long, JobDAGInfo> map; + private IMapDagStorageBenchmarkWorkload workload; + + @BeforeEach + @SuppressWarnings("unchecked") + void setUp() { + map = mock(IMap.class); + SeaTunnelStorageEnvironmentContext environment = + mock(SeaTunnelStorageEnvironmentContext.class, RETURNS_DEEP_STUBS); + when(environment + .getServer() + .getNodeEngine() + .getHazelcastInstance() + .<Long, JobDAGInfo>getMap(Constant.IMAP_FINISHED_JOB_VERTEX_INFO)) + .thenReturn(map); + when(environment.storageConfig().getEngineConfig().getHistoryJobExpireMinutes()) + .thenReturn(60); + doAnswer( + invocation -> { + Long key = invocation.getArgument(0); + JobDAGInfo value = invocation.getArgument(1); + persisted.put(key, value); + return cached.put(key, value); + }) + .when(map) + .put(anyLong(), any(JobDAGInfo.class)); + doAnswer( + invocation -> { + Long key = invocation.getArgument(0); + JobDAGInfo value = invocation.getArgument(1); + writtenKeys.add(key); + persisted.put(key, value); + return cached.put(key, value); + }) + .when(map) + .put(anyLong(), any(JobDAGInfo.class), eq(60L), eq(TimeUnit.MINUTES)); + when(map.get(anyLong())).thenAnswer(invocation -> cached.get(invocation.getArgument(0))); + when(map.evict(anyLong())) + .thenAnswer(invocation -> cached.remove(invocation.getArgument(0)) != null); + doAnswer( + invocation -> { + Iterable<Long> keys = invocation.getArgument(0); + for (Long key : keys) { + if (persisted.containsKey(key)) { + cached.put(key, persisted.get(key)); + } + } + return null; + }) + .when(map) + .loadAll(anySet(), eq(true)); + doAnswer( + invocation -> { + cached.remove(invocation.getArgument(0)); + persisted.remove(invocation.getArgument(0)); + return null; + }) + .when(map) + .delete(anyLong()); + + workload = new IMapDagStorageBenchmarkWorkload(); + workload.pipelineCount = 1; + workload.storedDagCount = 100; + workload.setUp(environment); + clearInvocations(map); + } + + @ParameterizedTest + @ValueSource(ints = {1, 20}) + void boundsIntermediateVerificationWithoutReloadingHistory(int batches) { + Set<Long> retainedKeys = new HashSet<>(persisted.keySet()); + for (int batch = 0; batch < batches; batch++) { + storeBatch(IterationType.MEASUREMENT, batches + 1); + workload.cleanStoreIteration(); + assertEquals(retainedKeys, persisted.keySet()); + } + assertEquals(batches * 100, new HashSet<>(writtenKeys).size()); + verify(map, times(batches * 3)).get(anyLong()); + verify(map, times(batches * 100)).delete(anyLong()); + verify(map, never()).evict(anyLong()); + verify(map, never()).loadAll(anySet(), anyBoolean()); + verify(map, times(batches * 100)) + .put(anyLong(), any(JobDAGInfo.class), eq(60L), eq(TimeUnit.MINUTES)); + } + + @ParameterizedTest + @CsvSource({"0,1", "3,1", "3,5"}) + void reloadsOnlyAfterTheFinalMeasurement(int warmups, int measurements) { + for (int index = 0; index < warmups; index++) { + storeBatch(IterationType.WARMUP, warmups); + workload.cleanStoreIteration(); + } + for (int index = 0; index < measurements - 1; index++) { + storeBatch(IterationType.MEASUREMENT, measurements); + workload.cleanStoreIteration(); + } + verify(map, never()).evict(anyLong()); + verify(map, never()).loadAll(anySet(), anyBoolean()); + + long lastKey = storeBatch(IterationType.MEASUREMENT, measurements); + workload.cleanStoreIteration(); + Set<Long> samples = new HashSet<>(); + samples.add(lastKey + 99); + samples.add(lastKey + 49); + samples.add(lastKey); + verify(map).loadAll(samples, true); + for (Long key : samples) { + verify(map).evict(key); + } + assertEquals(101, persisted.size()); + } + + @ParameterizedTest + @CsvSource({"0,false", "50,false", "99,false", "0,true", "50,true", "99,true"}) + void rejectsInvalidCachedSamplesAndStillDeletesTheBatch(int sampleIndex, boolean missing) { + storeBatch(IterationType.MEASUREMENT, 2); + Long key = writtenKeys.get(sampleIndex); + if (missing) { + cached.remove(key); + } else { + cached.put(key, JobDagFixtureFactory.create(2)); + } + IllegalStateException failure = + assertThrows(IllegalStateException.class, workload::cleanStoreIteration); + assertTrue(failure.getMessage().contains("did not match for key " + key)); + assertEquals(101, persisted.size()); + verify(map, times(100)).delete(anyLong()); + verify(map, never()).loadAll(anySet(), anyBoolean()); + } + + @Test + void finalReadBackDetectsMissingPersistenceEvenWhenTheCachedValueIsCorrect() { + storeBatch(IterationType.MEASUREMENT, 1); + Long key = writtenKeys.get(0); + persisted.remove(key); + assertTrue(cached.containsKey(key)); + + IllegalStateException failure = + assertThrows(IllegalStateException.class, workload::cleanStoreIteration); + assertTrue(failure.getMessage().contains("did not match for key " + key)); + verify(map, times(100)).delete(anyLong()); + assertEquals(101, persisted.size()); + } + + @Test + void cleansAcknowledgedWritesAfterAnIncompleteBatch() { + doAnswer( + invocation -> { + if (writtenKeys.size() == 6) { + throw new IllegalStateException("write failed"); + } + Long key = invocation.getArgument(0); + JobDAGInfo value = invocation.getArgument(1); + writtenKeys.add(key); + persisted.put(key, value); + return cached.put(key, value); + }) + .when(map) + .put(anyLong(), any(JobDAGInfo.class), eq(60L), eq(TimeUnit.MINUTES)); + workload.prepareStoreIteration(iteration(IterationType.MEASUREMENT, 1)); + assertEquals( + "write failed", + assertThrows(IllegalStateException.class, workload::storeFinishedJobDagBatch) + .getMessage()); + assertTrue( + assertThrows(IllegalStateException.class, workload::cleanStoreIteration) + .getMessage() + .contains("expected entry count")); + verify(map, times(6)).delete(anyLong()); + assertEquals(101, persisted.size()); + verify(map, never()).loadAll(anySet(), anyBoolean()); + } + + @Test + void loadInvocationsStillReloadAndValidateWithoutDeletingData() { + workload.prepareStoreIteration(iteration(IterationType.MEASUREMENT, 1)); + for (int index = 0; index < 2; index++) { + workload.prepareInvocation(); + workload.loadFinishedJobDag(); + workload.cleanInvocation(); + } + workload.cleanStoreIteration(); + verify(map, times(2)).loadAll(anySet(), eq(true)); + verify(map, times(4)).evict(anyLong()); + verify(map, times(2)).get(anyLong()); + verify(map, never()).delete(anyLong()); + } + + private long storeBatch(IterationType type, int count) { + workload.prepareStoreIteration(iteration(type, count)); + return workload.storeFinishedJobDagBatch(); + } + + private static IterationParams iteration(IterationType type, int count) { + return new IterationParams(type, count, TimeValue.seconds(1), 1); + } +} diff --git a/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/storage/imap/IMapDagStoragePersistenceTest.java b/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/storage/imap/IMapDagStoragePersistenceTest.java new file mode 100644 index 0000000000..c937718694 --- /dev/null +++ b/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/storage/imap/IMapDagStoragePersistenceTest.java @@ -0,0 +1,120 @@ +/* + * 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.benchmark.storage.imap; + +import org.apache.seatunnel.benchmark.dag.JobDagFixtureFactory; +import org.apache.seatunnel.benchmark.storage.SeaTunnelStorageEnvironmentContext; +import org.apache.seatunnel.engine.common.Constant; +import org.apache.seatunnel.engine.core.job.JobDAGInfo; +import org.apache.seatunnel.engine.imap.storage.file.common.WALReader; +import org.apache.seatunnel.engine.imap.storage.file.config.FileConfiguration; +import org.apache.seatunnel.engine.serializer.protobuf.ProtoStuffSerializer; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; + +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.openjdk.jmh.infra.IterationParams; +import org.openjdk.jmh.runner.IterationType; +import org.openjdk.jmh.runner.options.TimeValue; + +import java.util.Collections; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** Exercises complete DAG batches through the benchmark's actual file-backed MapStore. */ +class IMapDagStoragePersistenceTest { + + @ParameterizedTest + @CsvSource({"1,0", "1,100", "10,0", "10,100", "100,0", "100,100"}) + @Timeout(value = 3, unit = TimeUnit.MINUTES) + void persistsCompleteBatchesAndDeletionRecords(int pipelines, int retainedDags) + throws Exception { + SeaTunnelStorageEnvironmentContext environment = new SeaTunnelStorageEnvironmentContext(); + try { + environment.setUp(); + IMapDagStorageBenchmarkWorkload workload = new IMapDagStorageBenchmarkWorkload(); + workload.pipelineCount = pipelines; + workload.storedDagCount = retainedDags; + workload.setUp(environment); + JobDAGInfo expected = JobDagFixtureFactory.create(pipelines); + String clusterName = + environment + .getServer() + .getNodeEngine() + .getHazelcastInstance() + .getConfig() + .getClusterName(); + Path walRoot = + new Path( + environment + .imapDirectory() + .resolve(clusterName) + .resolve(Constant.IMAP_FINISHED_JOB_VERTEX_INFO) + .toUri()); + try (FileSystem fileSystem = FileSystem.newInstanceLocal(new Configuration())) { + WALReader reader = + new WALReader( + fileSystem, FileConfiguration.HDFS, new ProtoStuffSerializer()); + Set<Object> retainedKeys = + new HashSet<>(reader.loadAllData(walRoot, Collections.emptySet()).keySet()); + assertEquals(retainedDags + 1, retainedKeys.size()); + for (int batch = 0; batch < 2; batch++) { + workload.prepareStoreIteration( + new IterationParams( + IterationType.MEASUREMENT, 2, TimeValue.seconds(1), 1)); + workload.prepareInvocation(); + long lastKey = workload.storeFinishedJobDagBatch(); + workload.cleanInvocation(); + + // Read the WAL independently of IMap's cache, once for the entire batch. + Map<Object, Object> persisted = + reader.loadAllData(walRoot, Collections.emptySet()); + assertEquals(retainedDags + 101, persisted.size()); + for (long key = lastKey; key < lastKey + 100; key++) { + assertEquals(expected, persisted.get(key), "Missing or invalid DAG " + key); + } + workload.cleanStoreIteration(); + Map<Object, Object> afterDeletion = + reader.loadAllData(walRoot, Collections.emptySet()); + assertEquals(retainedKeys, afterDeletion.keySet()); + assertTrue(afterDeletion.values().stream().allMatch(expected::equals)); + for (long key = lastKey; key < lastKey + 100; key++) { + assertFalse(afterDeletion.containsKey(key)); + } + } + for (int invocation = 0; invocation < 2; invocation++) { + workload.prepareInvocation(); + workload.loadFinishedJobDag(); + workload.cleanInvocation(); + } + } + } finally { + environment.tearDown(); + } + } +}
