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-12026-65936d6e08aa34f5cfa967dec0e4cdf96b9d101e in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit c6b363af5c18702102dcc5536f6adf47b0ccb067 Author: zhiweiniu <[email protected]> AuthorDate: Tue Sep 1 15:37:34 2026 +0000 [Feature][Zeta] Add checkpoint completion benchmark (#12026) --- docs/en/engines/zeta/benchmark.md | 23 +- docs/zh/engines/zeta/benchmark.md | 21 +- .../benchmark/CheckpointBenchmarkTrigger.java | 69 ++++ .../benchmark/CheckpointingTimeBenchmark.java | 66 ++++ .../CheckpointingTimeBenchmarkPipeline.java | 295 ++++++++++++++++ .../SeaTunnelCheckpointEnvironmentContext.java | 380 +++++++++++++++++++++ .../connector/source/BenchmarkSourceReader.java | 4 +- .../benchmark/engine-checkpoint.yaml.template | 42 +++ .../hazelcast-checkpoint-master.yaml.template | 50 +++ .../hazelcast-checkpoint-worker.yaml.template | 38 +++ .../benchmark/source-sink-checkpoint.conf.template | 49 +++ .../benchmark/CheckpointingTimeBenchmarkTest.java | 72 ++++ 12 files changed, 1104 insertions(+), 5 deletions(-) diff --git a/docs/en/engines/zeta/benchmark.md b/docs/en/engines/zeta/benchmark.md index b20db41fa6..e093286ea7 100644 --- a/docs/en/engines/zeta/benchmark.md +++ b/docs/en/engines/zeta/benchmark.md @@ -15,12 +15,14 @@ not replace a production proof of concept. ## How It Works -The `seatunnel-benchmarks` module provides two types of tests: +The `seatunnel-benchmarks` module provides three types of tests: - `SeaTunnelRowBenchmark` measures hot paths such as row creation, access, copying, projection, and size calculation. - `SeaTunnelPipelineBenchmark` starts an embedded single-node Zeta cluster and runs complete bounded jobs through the normal client and configuration parser APIs. +- `CheckpointingTimeBenchmark` keeps one streaming job running and measures the time required to + complete explicitly triggered regular checkpoints. The MiniCluster starts during JMH Trial setup and is outside the measurement. Job submission, scheduling, Source, Transform, Sink, and job completion are inside the JMH measurement. @@ -149,6 +151,23 @@ java -jar seatunnel-benchmarks/target/benchmarks.jar SeaTunnelRowBenchmark \ For a quick functional validation, add `-f 1 -wi 0 -i 1 -r 1s` to shorten the run. A single un-warmed sample is not valid performance evidence. +### Run the Checkpoint Benchmark + +```bash +java -jar seatunnel-benchmarks/target/benchmarks.jar CheckpointingTimeBenchmark +``` + +The benchmark runs both `recordSize=1b` and `recordSize=1kb`. `checkpointSingleInput` uses a +controlled input rate and equal Source/Sink parallelism. Its dedicated JMH environment starts an +isolated two-node Zeta cluster with separate master and worker roles and one streaming job per trial. +The master has no worker slots, the worker executes the pipeline, and IMap backup count is zero. A +separate checkpoint engine configuration (not the shared benchmark engine configuration) enables +the `engine*` MapStore with the HDFS file storage implementation on the local filesystem, and stores +checkpoints through the HDFS checkpoint plugin in local mode. Every invocation explicitly triggers +one regular checkpoint and waits until Zeta has persisted and completed it. The score is checkpoint +completion time in `s/op`, so lower is better. Job startup, workload ramp-up, persistence +verification, and job shutdown are outside the measured invocation. + ### Read Workflow Reports The scheduled and manually triggered `Benchmarks` workflow runs each selected benchmark on Java 8 @@ -234,7 +253,7 @@ These conditions reject incomplete output and prove that Transform work reached | Field | Description | |---|---| -| `Score` | Processed rows per second for a Pipeline benchmark; higher is better. Row microbenchmarks retain `ops/ms`. | +| `Score` | Processed rows per second for a Pipeline benchmark; higher is better. Row microbenchmarks retain `ops/ms`, while the checkpoint benchmark reports `s/op` and lower is better. | | `Error` | Uncertainty calculated from samples inside this JMH run. | | `Cnt` | Aggregated measurement samples, not processed rows. | | `Units` | Unit of the score. | diff --git a/docs/zh/engines/zeta/benchmark.md b/docs/zh/engines/zeta/benchmark.md index e4f91bc13d..294e4a799c 100644 --- a/docs/zh/engines/zeta/benchmark.md +++ b/docs/zh/engines/zeta/benchmark.md @@ -13,11 +13,13 @@ title: Zeta 基准测试 ## 工作原理 -`seatunnel-benchmarks` 提供两类测试： +`seatunnel-benchmarks` 提供三类测试： - `SeaTunnelRowBenchmark`：测试 Row 创建、读取、复制、投影和大小计算等热点代码。 - `SeaTunnelPipelineBenchmark`：启动单节点嵌入式 Zeta 集群，并通过正常的 Client 和配置 解析 API 运行完整的有界作业。 +- `CheckpointingTimeBenchmark`：保持一个流式作业运行，并测量显式触发普通 Checkpoint 的 + 完成耗时。 MiniCluster 在每个 JMH Trial 的 Setup 阶段启动，不计入测量。作业提交、调度、Source、 Transform、Sink 和作业完成都计入 JMH 测量。 @@ -143,6 +145,21 @@ java -jar seatunnel-benchmarks/target/benchmarks.jar SeaTunnelRowBenchmark \ 快速功能验证时可以增加 `-f 1 -wi 0 -i 1 -r 1s` 缩短运行时间。没有预热且只有一个样本的 结果不能用于性能结论。 +### 运行 Checkpoint 基准测试 + +```bash +java -jar seatunnel-benchmarks/target/benchmarks.jar CheckpointingTimeBenchmark +``` + +该测试覆盖 `recordSize=1b` 和 `recordSize=1kb`。`checkpointSingleInput` 使用受控输入速率 +以及相同的 Source/Sink 并行度。专用 JMH 环境会在每个 Trial 中启动 master/worker 角色 +分离的双节点 Zeta 集群和一个流式作业。master 不提供 worker slot，pipeline 只在 worker +执行，IMap backup count 为 0。该环境使用一份独立的 Checkpoint Engine 配置（不复用普通 +Benchmark 的 Engine 配置），为 `engine*` 开启基于本地文件系统的 HDFS MapStore，并通过 +HDFS Checkpoint 插件的 local 模式保存状态。每次 invocation 显式触发一个普通 Checkpoint， +并等待 Zeta 完成持久化。Score 使用 `s/op`，数值越低越好；作业启动、负载建立、持久化 +校验和作业关闭不计入 invocation 时间。 + ### 查看 Workflow 报告 定时或手动触发的 `Benchmarks` workflow 会在 Java 8 和 Java 11 上运行所选 benchmark。 @@ -220,7 +237,7 @@ Linux runner 使用 async-profiler 的 `ctimer` 事件进行 CPU profiling，不 | 字段 | 说明 | |---|---| -| `Score` | Pipeline Benchmark 每秒处理的行数，越大越好；Row 微基准仍使用 `ops/ms`。 | +| `Score` | Pipeline Benchmark 每秒处理的行数，越大越好；Row 微基准使用 `ops/ms`；Checkpoint Benchmark 使用 `s/op`，越低越好。 | | `Error` | 根据本次 JMH 运行内部样本计算的不确定性。 | | `Cnt` | 参与聚合的 Measurement 样本数，不是处理行数。 | | `Units` | Score 的单位。 | diff --git a/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/CheckpointBenchmarkTrigger.java b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/CheckpointBenchmarkTrigger.java new file mode 100644 index 0000000000..ab754e7b0a --- /dev/null +++ b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/CheckpointBenchmarkTrigger.java @@ -0,0 +1,69 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.benchmark; + +import org.apache.seatunnel.common.utils.ReflectionUtils; +import org.apache.seatunnel.engine.common.utils.PassiveCompletableFuture; +import org.apache.seatunnel.engine.common.utils.concurrent.CompletableFuture; +import org.apache.seatunnel.engine.core.checkpoint.CheckpointType; +import org.apache.seatunnel.engine.server.checkpoint.CheckpointCoordinator; +import org.apache.seatunnel.engine.server.checkpoint.CompletedCheckpoint; +import org.apache.seatunnel.engine.server.checkpoint.PendingCheckpoint; + +import java.time.Instant; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.apache.seatunnel.engine.core.checkpoint.CheckpointType.CHECKPOINT_TYPE; + +/** Benchmark-only test bridge for explicitly triggering a regular checkpoint. */ +final class CheckpointBenchmarkTrigger { + + private CheckpointBenchmarkTrigger() {} + + @SuppressWarnings("unchecked") + static PassiveCompletableFuture<CompletedCheckpoint> trigger( + CheckpointCoordinator coordinator) { + // This bridge bypasses the coordinator trigger lock, so callers must keep periodic + // checkpoints effectively disabled and use a single JMH thread to prevent concurrent + // checkpoint triggers. + CompletableFuture<PendingCheckpoint> pendingCheckpoint = + (CompletableFuture<PendingCheckpoint>) + ReflectionUtils.invoke( + coordinator, + "createPendingCheckpoint", + new Class<?>[] {long.class, CheckpointType.class}, + new Object[] {Instant.now().toEpochMilli(), CHECKPOINT_TYPE}); + ReflectionUtils.invoke( + coordinator, + "startTriggerPendingCheckpoint", + new Class<?>[] {CompletableFuture.class}, + new Object[] {pendingCheckpoint}); + return pendingCheckpoint.join().getCompletableFuture(); + } + + static boolean hasPendingCheckpoint(CheckpointCoordinator coordinator) { + AtomicInteger pendingCounter = + (AtomicInteger) + ReflectionUtils.getField(coordinator, "pendingCounter") + .orElseThrow( + () -> + new IllegalStateException( + "Checkpoint pending counter is unavailable")); + return pendingCounter.get() > 0; + } +} diff --git a/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/CheckpointingTimeBenchmark.java b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/CheckpointingTimeBenchmark.java new file mode 100644 index 0000000000..a0a273a448 --- /dev/null +++ b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/CheckpointingTimeBenchmark.java @@ -0,0 +1,66 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.benchmark; + +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Threads; +import org.openjdk.jmh.runner.Runner; +import org.openjdk.jmh.runner.RunnerException; +import org.openjdk.jmh.runner.options.Options; +import org.openjdk.jmh.runner.options.OptionsBuilder; +import org.openjdk.jmh.runner.options.VerboseMode; + +import static java.util.concurrent.TimeUnit.SECONDS; + +/** Measures regular checkpoint completion time for one input pipeline. */ +@BenchmarkMode(Mode.AverageTime) +@OutputTimeUnit(SECONDS) +@Threads(1) +@Fork( + value = 3, + jvmArgsAppend = { + "-Xms4g", + "-Xmx4g", + "-XX:+UseG1GC", + "-XX:+AlwaysPreTouch", + "-XX:+DisableExplicitGC", + "-XX:ActiveProcessorCount=4", + "-Djava.net.preferIPv4Stack=true" + }) +public class CheckpointingTimeBenchmark extends BenchmarkBase { + + public static void main(String[] args) throws RunnerException { + Options options = + new OptionsBuilder() + .verbosity(VerboseMode.NORMAL) + .include(CheckpointingTimeBenchmark.class.getCanonicalName()) + .build(); + + new Runner(options).run(); + } + + @Benchmark + public void checkpointSingleInput(CheckpointingTimeBenchmarkPipeline pipeline) + throws Exception { + pipeline.triggerCheckpoint(); + } +} diff --git a/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/CheckpointingTimeBenchmarkPipeline.java b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/CheckpointingTimeBenchmarkPipeline.java new file mode 100644 index 0000000000..4383212022 --- /dev/null +++ b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/CheckpointingTimeBenchmarkPipeline.java @@ -0,0 +1,295 @@ +/* + * 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; + +import org.apache.seatunnel.engine.client.job.ClientJobExecutionEnvironment; +import org.apache.seatunnel.engine.client.job.ClientJobProxy; +import org.apache.seatunnel.engine.common.config.JobConfig; +import org.apache.seatunnel.engine.common.job.JobStatus; +import org.apache.seatunnel.engine.common.utils.PassiveCompletableFuture; +import org.apache.seatunnel.engine.core.checkpoint.CheckpointHistoryEntry; +import org.apache.seatunnel.engine.core.checkpoint.CheckpointInfo; +import org.apache.seatunnel.engine.core.checkpoint.CheckpointStatus; +import org.apache.seatunnel.engine.server.SeaTunnelServer; +import org.apache.seatunnel.engine.server.checkpoint.CheckpointCoordinator; +import org.apache.seatunnel.engine.server.checkpoint.CompletedCheckpoint; +import org.apache.seatunnel.engine.server.checkpoint.monitor.CheckpointMonitorService; +import org.apache.seatunnel.engine.server.dag.physical.SubPlan; +import org.apache.seatunnel.engine.server.master.JobMaster; + +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Param; +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 java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.time.Duration; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +/** Long-running source-to-sink pipeline whose checkpoint completion time is measured. */ +@State(Scope.Thread) +public class CheckpointingTimeBenchmarkPipeline { + + private static final String JOB_TEMPLATE = + loadTemplate("/benchmark/source-sink-checkpoint.conf.template"); + private static final Duration START_TIMEOUT = Duration.ofMinutes(2); + private static final Duration CHECKPOINT_TIMEOUT = Duration.ofMinutes(2); + private static final Duration CANCEL_TIMEOUT = Duration.ofSeconds(30); + private static final String DEBLOATING_RECORD_SIZE_VALUE = "1b"; + private static final String UNALIGNED_RECORD_SIZE_VALUE = "1kb"; + private static final long SOURCE_RATE_PER_SECOND = 10_000L; + private static final int PIPELINE_PARALLELISM = 4; + + @Param({"1b", "1kb"}) + public String recordSize; + + private SeaTunnelCheckpointEnvironmentContext environment; + private ClientJobProxy jobProxy; + private CheckpointCoordinator checkpointCoordinator; + private CheckpointMonitorService checkpointMonitorService; + private int pipelineId; + private boolean checkpointCompleted; + + /** Submits the streaming pipeline after the checkpoint benchmark cluster has started. */ + @Setup(Level.Trial) + public void setUp(SeaTunnelCheckpointEnvironmentContext environment) throws Exception { + this.environment = environment; + try { + Path jobConfigFile = environment.getClusterHome().resolve("checkpoint-benchmark.conf"); + Path resultPath = environment.getClusterHome().resolve("checkpoint-results"); + Files.write( + jobConfigFile, + createCheckpointJobConfig(resultPath).getBytes(StandardCharsets.UTF_8)); + JobConfig jobConfig = new JobConfig(); + jobConfig.setName("checkpoint-benchmark-" + UUID.randomUUID()); + ClientJobExecutionEnvironment executionEnvironment = + environment + .getClient() + .createExecutionContext( + jobConfigFile.toString(), + jobConfig, + environment.getMasterConfig()); + jobProxy = executionEnvironment.execute(); + waitUntilRunning(); + resolveCheckpointControl(); + + // The MASTER node has no worker slots, so RUNNING proves execution is on WORKER. + Thread.sleep(2_000L); + } catch (Exception setupFailure) { + try { + tearDown(); + } catch (Exception cleanupFailure) { + setupFailure.addSuppressed(cleanupFailure); + } + throw setupFailure; + } + } + + /** Stops the streaming pipeline after checking checkpoint and IMap persistence. */ + @TearDown(Level.Trial) + public void tearDown() throws Exception { + try { + if (checkpointCompleted) { + verifyCheckpointWasPersisted(); + } + if (jobProxy != null && !jobProxy.getJobStatus().isEndState()) { + jobProxy.cancelJob(); + waitUntilCanceled(); + } + } finally { + environment = null; + jobProxy = null; + checkpointCoordinator = null; + checkpointMonitorService = null; + checkpointCompleted = false; + } + } + + /** Triggers a regular checkpoint and waits until the coordinator records its completion. */ + public void triggerCheckpoint() throws Exception { + if (checkpointCoordinator == null || checkpointMonitorService == null) { + throw new IllegalStateException("Checkpoint benchmark pipeline has not been started"); + } + long previousCheckpointId = latestCompletedCheckpointId(); + PassiveCompletableFuture<CompletedCheckpoint> checkpoint = + CheckpointBenchmarkTrigger.trigger(checkpointCoordinator); + checkpoint.get(CHECKPOINT_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS); + + long deadline = System.nanoTime() + CHECKPOINT_TIMEOUT.toNanos(); + while (System.nanoTime() < deadline) { + if (latestCompletedCheckpointId() > previousCheckpointId + && !CheckpointBenchmarkTrigger.hasPendingCheckpoint(checkpointCoordinator)) { + checkpointCompleted = true; + return; + } + Thread.sleep(1L); + } + throw new IllegalStateException("Timed out waiting for benchmark checkpoint to complete"); + } + + String createCheckpointJobConfig(Path resultPath) { + return renderTemplate( + JOB_TEMPLATE, + "payload_size", + selectedRecordSize().getBytes(), + "source_rate_per_second", + SOURCE_RATE_PER_SECOND, + "pipeline_parallelism", + PIPELINE_PARALLELISM, + "result_path", + resultPath.toAbsolutePath()); + } + + private void waitUntilRunning() throws InterruptedException { + long deadline = System.nanoTime() + START_TIMEOUT.toNanos(); + while (System.nanoTime() < deadline) { + JobStatus status = jobProxy.getJobStatus(); + if (status == JobStatus.RUNNING) { + return; + } + if (status.isEndState()) { + throw new IllegalStateException( + "Checkpoint benchmark job ended during setup with " + status); + } + Thread.sleep(100L); + } + throw new IllegalStateException("Timed out waiting for checkpoint benchmark job to run"); + } + + private void waitUntilCanceled() throws InterruptedException { + long deadline = System.nanoTime() + CANCEL_TIMEOUT.toNanos(); + while (System.nanoTime() < deadline) { + if (jobProxy.getJobStatus().isEndState()) { + return; + } + Thread.sleep(10L); + } + throw new IllegalStateException("Timed out canceling checkpoint benchmark job"); + } + + private void resolveCheckpointControl() { + SeaTunnelServer server = + environment + .getMasterInstance() + .node + .getNodeEngine() + .getService(SeaTunnelServer.SERVICE_NAME); + if (!server.isMasterNode()) { + throw new IllegalStateException("Checkpoint benchmark master node is not active"); + } + JobMaster jobMaster = server.getCoordinatorService().getJobMaster(jobProxy.getJobId()); + if (jobMaster == null) { + throw new IllegalStateException("Checkpoint benchmark JobMaster is unavailable"); + } + List<SubPlan> pipelines = jobMaster.getPhysicalPlan().getPipelineList(); + if (pipelines.size() != 1) { + throw new IllegalStateException( + "Checkpoint benchmark requires one pipeline but found " + pipelines.size()); + } + pipelineId = pipelines.get(0).getPipelineId(); + checkpointCoordinator = + jobMaster.getCheckpointManager().getCheckpointCoordinator(pipelineId); + checkpointMonitorService = server.getCheckpointMonitorService(); + } + + private long latestCompletedCheckpointId() { + return checkpointMonitorService + .getHistory(jobProxy.getJobId(), pipelineId, 1, CheckpointStatus.COMPLETED).stream() + .map(CheckpointHistoryEntry::getCheckpointInfo) + .mapToLong(CheckpointInfo::getCheckpointId) + .findFirst() + .orElse(-1L); + } + + private void verifyCheckpointWasPersisted() throws Exception { + verifyPersistenceDirectory( + environment.getCheckpointStorageDirectory(), ".ser", "checkpoint"); + verifyPersistenceDirectory(environment.getMapStoreDirectory(), null, "MapStore"); + } + + private static void verifyPersistenceDirectory( + Path directory, String fileSuffix, String storageName) throws Exception { + if (!Files.isDirectory(directory)) { + throw new IllegalStateException( + storageName + " storage directory was not created: " + directory); + } + try (Stream<Path> files = Files.walk(directory)) { + if (files.noneMatch( + path -> + Files.isRegularFile(path) + && (fileSuffix == null + || path.getFileName() + .toString() + .endsWith(fileSuffix)))) { + throw new IllegalStateException( + "No persisted " + storageName + " state was found under " + directory); + } + } + } + + private SeaTunnelCheckpointEnvironmentContext.MemorySize selectedRecordSize() { + if (DEBLOATING_RECORD_SIZE_VALUE.equals(recordSize)) { + return SeaTunnelCheckpointEnvironmentContext.DEBLOATING_RECORD_SIZE; + } + if (UNALIGNED_RECORD_SIZE_VALUE.equals(recordSize)) { + return SeaTunnelCheckpointEnvironmentContext.UNALIGNED_RECORD_SIZE; + } + throw new IllegalArgumentException("Unsupported checkpoint record size: " + recordSize); + } + + private static String loadTemplate(String resourceName) { + InputStream input = + CheckpointingTimeBenchmarkPipeline.class.getResourceAsStream(resourceName); + if (input == null) { + throw new IllegalStateException("Benchmark template was not found: " + resourceName); + } + try (BufferedReader reader = + new BufferedReader(new InputStreamReader(input, StandardCharsets.UTF_8))) { + return reader.lines().collect(Collectors.joining("\n", "", "\n")); + } catch (IOException e) { + throw new IllegalStateException("Could not read benchmark template " + resourceName, e); + } + } + + private static String renderTemplate(String template, Object... replacements) { + String rendered = template; + for (int index = 0; index < replacements.length; index += 2) { + rendered = + rendered.replace( + "{{" + replacements[index] + "}}", + String.valueOf(replacements[index + 1])); + } + if (rendered.contains("{{")) { + throw new IllegalStateException( + "Checkpoint benchmark template contains an unresolved placeholder"); + } + return rendered; + } +} diff --git a/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/SeaTunnelCheckpointEnvironmentContext.java b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/SeaTunnelCheckpointEnvironmentContext.java new file mode 100644 index 0000000000..9a19889edc --- /dev/null +++ b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/SeaTunnelCheckpointEnvironmentContext.java @@ -0,0 +1,380 @@ +/* + * 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; + +import org.apache.seatunnel.common.config.Common; +import org.apache.seatunnel.common.config.DeployMode; +import org.apache.seatunnel.engine.client.SeaTunnelClient; +import org.apache.seatunnel.engine.common.config.ConfigProvider; +import org.apache.seatunnel.engine.common.config.SeaTunnelClientConfig; +import org.apache.seatunnel.engine.common.config.SeaTunnelConfig; +import org.apache.seatunnel.engine.server.SeaTunnelServerStarter; + +import org.openjdk.jmh.annotations.Level; +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 com.hazelcast.client.config.ClientConfig; +import com.hazelcast.config.Config; +import com.hazelcast.config.YamlConfigBuilder; +import com.hazelcast.instance.impl.HazelcastInstanceImpl; + +import java.io.BufferedReader; +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.time.Duration; +import java.util.Collections; +import java.util.Comparator; +import java.util.List; +import java.util.Locale; +import java.util.UUID; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +/** JMH environment that owns a dedicated two-node Zeta master/worker cluster and its client. */ +@State(Scope.Thread) +public class SeaTunnelCheckpointEnvironmentContext { + + public static final MemorySize DEBLOATING_RECORD_SIZE = MemorySize.parse("1b"); + public static final MemorySize UNALIGNED_RECORD_SIZE = MemorySize.parse("1kb"); + + private static final String ENGINE_CONFIG_TEMPLATE = + loadTemplate("/benchmark/engine-checkpoint.yaml.template"); + private static final String HAZELCAST_MASTER_CONFIG_TEMPLATE = + loadTemplate("/benchmark/hazelcast-checkpoint-master.yaml.template"); + private static final String HAZELCAST_WORKER_CONFIG_TEMPLATE = + loadTemplate("/benchmark/hazelcast-checkpoint-worker.yaml.template"); + private static final Duration START_TIMEOUT = Duration.ofMinutes(2); + private static final int WORKER_SLOT_COUNT = 12; + + private Path clusterHome; + private Path checkpointStorageDirectory; + private Path mapStoreDirectory; + private SeaTunnelConfig masterConfig; + private HazelcastInstanceImpl masterInstance; + private HazelcastInstanceImpl workerInstance; + private SeaTunnelClient client; + private String previousSeaTunnelHome; + private String previousUppercaseSeaTunnelHome; + private String previousSeaTunnelConfig; + private String previousCommonSeaTunnelHome; + private DeployMode previousDeployMode; + + /** Starts a dedicated master, worker, and client for checkpoint benchmark pipelines. */ + @Setup(Level.Trial) + public void setUp() throws Exception { + try { + prepareClusterHome(); + String clusterName = "seatunnel-checkpoint-benchmark-" + UUID.randomUUID(); + writeEngineConfig(); + + masterConfig = createMasterConfig(clusterName); + masterInstance = SeaTunnelServerStarter.createMasterHazelcastInstance(masterConfig); + + int masterPort = + masterInstance.getCluster().getLocalMember().getSocketAddress().getPort(); + SeaTunnelConfig workerConfig = + createWorkerConfig(clusterName, "127.0.0.1:" + masterPort); + workerInstance = SeaTunnelServerStarter.createWorkerHazelcastInstance(workerConfig); + waitUntilClusterFormed(); + + ClientConfig clientConfig = new SeaTunnelClientConfig(); + clientConfig.setClusterName(clusterName); + clientConfig + .getNetworkConfig() + .setAddresses(Collections.singletonList("127.0.0.1:" + masterPort)); + clientConfig + .getConnectionStrategyConfig() + .getConnectionRetryConfig() + .setClusterConnectTimeoutMillis(30_000L); + client = new SeaTunnelClient(clientConfig); + } catch (Exception setupFailure) { + try { + tearDown(); + } catch (Exception cleanupFailure) { + setupFailure.addSuppressed(cleanupFailure); + } + throw setupFailure; + } + } + + /** Stops the client and both benchmark cluster nodes. */ + @TearDown(Level.Trial) + public void tearDown() throws Exception { + Exception failure = null; + try { + if (client != null) { + client.close(); + } + } catch (Exception cleanupFailure) { + failure = cleanupFailure; + } finally { + if (workerInstance != null) { + workerInstance.shutdown(); + } + if (masterInstance != null) { + masterInstance.shutdown(); + } + restoreEnvironment(); + try { + deleteRecursively(clusterHome); + } catch (Exception deletionFailure) { + if (failure == null) { + failure = deletionFailure; + } else { + failure.addSuppressed(deletionFailure); + } + } + clearState(); + } + if (failure != null) { + throw failure; + } + } + + private void prepareClusterHome() throws IOException { + clusterHome = Files.createTempDirectory("seatunnel-checkpoint-benchmark-"); + Path persistenceDirectory = clusterHome.resolve("checkpoint-persistence"); + checkpointStorageDirectory = persistenceDirectory.resolve("checkpoints"); + mapStoreDirectory = persistenceDirectory.resolve("imap"); + + previousSeaTunnelHome = System.getProperty("seatunnel.home"); + previousUppercaseSeaTunnelHome = System.getProperty("SEATUNNEL_HOME"); + previousSeaTunnelConfig = System.getProperty("seatunnel.config"); + previousCommonSeaTunnelHome = Common.getSeaTunnelHome(); + previousDeployMode = Common.getDeployMode(); + System.setProperty("seatunnel.home", clusterHome.toString()); + System.setProperty("SEATUNNEL_HOME", clusterHome.toString()); + Common.setSeaTunnelHome(clusterHome.toString()); + Common.setDeployMode(DeployMode.CLIENT); + } + + private void writeEngineConfig() throws IOException { + Path engineConfigFile = clusterHome.resolve("seatunnel.yaml"); + Files.write( + engineConfigFile, + renderTemplate( + ENGINE_CONFIG_TEMPLATE, + "slot_count", + WORKER_SLOT_COUNT, + "checkpoint_storage_directory", + checkpointStorageDirectory.toAbsolutePath()) + .getBytes(StandardCharsets.UTF_8)); + System.setProperty("seatunnel.config", engineConfigFile.toString()); + } + + private SeaTunnelConfig createMasterConfig(String clusterName) { + String hazelcastConfig = + renderTemplate( + HAZELCAST_MASTER_CONFIG_TEMPLATE, + "cluster_name", + clusterName, + "imap_directory", + mapStoreDirectory.toAbsolutePath()); + return createSeaTunnelConfig(hazelcastConfig); + } + + private SeaTunnelConfig createWorkerConfig(String clusterName, String masterAddress) { + String hazelcastConfig = + renderTemplate( + HAZELCAST_WORKER_CONFIG_TEMPLATE, + "cluster_name", + clusterName, + "master_address", + masterAddress); + return createSeaTunnelConfig(hazelcastConfig); + } + + private SeaTunnelConfig createSeaTunnelConfig(String hazelcastConfig) { + String engineConfig = + renderTemplate( + ENGINE_CONFIG_TEMPLATE, + "slot_count", + WORKER_SLOT_COUNT, + "checkpoint_storage_directory", + checkpointStorageDirectory.toAbsolutePath()); + SeaTunnelConfig config = ConfigProvider.locateAndGetSeaTunnelConfigFromString(engineConfig); + Config memberConfig = + new YamlConfigBuilder( + new ByteArrayInputStream( + hazelcastConfig.getBytes(StandardCharsets.UTF_8))) + .build(); + config.setHazelcastConfig(memberConfig); + return config; + } + + private void waitUntilClusterFormed() throws InterruptedException { + long deadline = System.nanoTime() + START_TIMEOUT.toNanos(); + while (System.nanoTime() < deadline) { + if (masterInstance.getCluster().getMembers().size() == 2 + && workerInstance.getCluster().getMembers().size() == 2) { + return; + } + Thread.sleep(100L); + } + throw new IllegalStateException("Timed out forming checkpoint master/worker cluster"); + } + + Path getClusterHome() { + return clusterHome; + } + + Path getCheckpointStorageDirectory() { + return checkpointStorageDirectory; + } + + Path getMapStoreDirectory() { + return mapStoreDirectory; + } + + SeaTunnelConfig getMasterConfig() { + return masterConfig; + } + + HazelcastInstanceImpl getMasterInstance() { + return masterInstance; + } + + SeaTunnelClient getClient() { + return client; + } + + /** Immutable binary memory size used only by this checkpoint environment. */ + public static final class MemorySize { + + private static final Pattern SIZE_PATTERN = Pattern.compile("([0-9]+)\\s*([a-zA-Z]+)"); + + private final long bytes; + + private MemorySize(long bytes) { + this.bytes = bytes; + } + + public static MemorySize parse(String value) { + Matcher matcher = SIZE_PATTERN.matcher(value.trim()); + if (!matcher.matches()) { + throw new IllegalArgumentException("Invalid memory size: " + value); + } + + long amount = Long.parseLong(matcher.group(1)); + String unit = matcher.group(2).toLowerCase(Locale.ROOT); + long multiplier; + switch (unit) { + case "b": + multiplier = 1L; + break; + case "kb": + case "kib": + multiplier = 1L << 10; + break; + case "mb": + case "mib": + multiplier = 1L << 20; + break; + case "gb": + case "gib": + multiplier = 1L << 30; + break; + default: + throw new IllegalArgumentException("Unsupported memory size unit: " + unit); + } + return new MemorySize(Math.multiplyExact(amount, multiplier)); + } + + public long getBytes() { + return bytes; + } + } + + private void restoreEnvironment() { + restoreSystemProperty("seatunnel.home", previousSeaTunnelHome); + restoreSystemProperty("SEATUNNEL_HOME", previousUppercaseSeaTunnelHome); + restoreSystemProperty("seatunnel.config", previousSeaTunnelConfig); + Common.setSeaTunnelHome(previousCommonSeaTunnelHome); + Common.setDeployMode(previousDeployMode); + } + + private void clearState() { + clusterHome = null; + checkpointStorageDirectory = null; + mapStoreDirectory = null; + masterConfig = null; + masterInstance = null; + workerInstance = null; + client = null; + } + + private static String loadTemplate(String resourceName) { + InputStream input = + SeaTunnelCheckpointEnvironmentContext.class.getResourceAsStream(resourceName); + if (input == null) { + throw new IllegalStateException("Benchmark template was not found: " + resourceName); + } + try (BufferedReader reader = + new BufferedReader(new InputStreamReader(input, StandardCharsets.UTF_8))) { + return reader.lines().collect(Collectors.joining("\n", "", "\n")); + } catch (IOException e) { + throw new IllegalStateException("Could not read benchmark template " + resourceName, e); + } + } + + private static String renderTemplate(String template, Object... replacements) { + String rendered = template; + for (int index = 0; index < replacements.length; index += 2) { + rendered = + rendered.replace( + "{{" + replacements[index] + "}}", + String.valueOf(replacements[index + 1])); + } + if (rendered.contains("{{")) { + throw new IllegalStateException( + "Checkpoint benchmark template contains an unresolved placeholder"); + } + return rendered; + } + + private static void restoreSystemProperty(String key, String value) { + if (value == null) { + System.clearProperty(key); + } else { + System.setProperty(key, value); + } + } + + private static void deleteRecursively(Path root) throws IOException { + if (root == null || !Files.exists(root)) { + return; + } + List<Path> paths; + try (Stream<Path> stream = Files.walk(root)) { + paths = stream.sorted(Comparator.reverseOrder()).collect(Collectors.toList()); + } + for (Path path : paths) { + Files.deleteIfExists(path); + } + } +} diff --git a/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/connector/source/BenchmarkSourceReader.java b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/connector/source/BenchmarkSourceReader.java index 7c0cdf7d61..7fa96b98b9 100644 --- a/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/connector/source/BenchmarkSourceReader.java +++ b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/connector/source/BenchmarkSourceReader.java @@ -115,7 +115,9 @@ public final class BenchmarkSourceReader } @Override - public void notifyCheckpointComplete(long checkpointId) {} + public void notifyCheckpointComplete(long checkpointId) { + // The synthetic source has no external offsets to commit after coordinator completion. + } static long scheduledMillis(BenchmarkSourceSplit split, long sequence) { if (split.getRatePerSecond() == 0) { diff --git a/seatunnel-benchmarks/src/main/resources/benchmark/engine-checkpoint.yaml.template b/seatunnel-benchmarks/src/main/resources/benchmark/engine-checkpoint.yaml.template new file mode 100644 index 0000000000..cd1e9d4eb8 --- /dev/null +++ b/seatunnel-benchmarks/src/main/resources/benchmark/engine-checkpoint.yaml.template @@ -0,0 +1,42 @@ +# +# 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. +# + +seatunnel: + engine: + backup-count: 0 + classloader-cache-mode: true + slot-service: + dynamic-slot: false + slot-num: {{slot_count}} + checkpoint: + interval: 86400000 + timeout: 120000 + storage: + type: hdfs + max-retained: 20 + plugin-config: + namespace: "{{checkpoint_storage_directory}}" + storage.type: local + fs.defaultFS: file:/// + telemetry: + metric: + enabled: false + logs: + scheduled-deletion-enable: false + http: + enable-http: false + enable-https: false diff --git a/seatunnel-benchmarks/src/main/resources/benchmark/hazelcast-checkpoint-master.yaml.template b/seatunnel-benchmarks/src/main/resources/benchmark/hazelcast-checkpoint-master.yaml.template new file mode 100644 index 0000000000..9c5223190a --- /dev/null +++ b/seatunnel-benchmarks/src/main/resources/benchmark/hazelcast-checkpoint-master.yaml.template @@ -0,0 +1,50 @@ +# +# 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. +# + +hazelcast: + cluster-name: "{{cluster_name}}" + network: + interfaces: + enabled: true + interfaces: + - 127.0.0.1 + join: + auto-detection: + enabled: false + multicast: + enabled: false + tcp-ip: + enabled: true + member-list: [] + port: + auto-increment: true + port: 5701 + map: + engine*: + backup-count: 0 + async-backup-count: 0 + map-store: + enabled: true + initial-mode: EAGER + factory-class-name: org.apache.seatunnel.engine.server.persistence.FileMapStoreFactory + properties: + type: hdfs + namespace: "{{imap_directory}}" + clusterName: "{{cluster_name}}" + fs.defaultFS: file:/// + properties: + hazelcast.phone.home.enabled: false diff --git a/seatunnel-benchmarks/src/main/resources/benchmark/hazelcast-checkpoint-worker.yaml.template b/seatunnel-benchmarks/src/main/resources/benchmark/hazelcast-checkpoint-worker.yaml.template new file mode 100644 index 0000000000..8f19d4bc9f --- /dev/null +++ b/seatunnel-benchmarks/src/main/resources/benchmark/hazelcast-checkpoint-worker.yaml.template @@ -0,0 +1,38 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +hazelcast: + cluster-name: "{{cluster_name}}" + network: + interfaces: + enabled: true + interfaces: + - 127.0.0.1 + join: + auto-detection: + enabled: false + multicast: + enabled: false + tcp-ip: + enabled: true + member-list: + - "{{master_address}}" + port: + auto-increment: true + port: 5701 + properties: + hazelcast.phone.home.enabled: false diff --git a/seatunnel-benchmarks/src/main/resources/benchmark/source-sink-checkpoint.conf.template b/seatunnel-benchmarks/src/main/resources/benchmark/source-sink-checkpoint.conf.template new file mode 100644 index 0000000000..b96e731f76 --- /dev/null +++ b/seatunnel-benchmarks/src/main/resources/benchmark/source-sink-checkpoint.conf.template @@ -0,0 +1,49 @@ +# +# 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. +# + +env { + job.name = "checkpoint-benchmark" + job.mode = "STREAMING" + checkpoint.interval = 86400000 + checkpoint.timeout = 120000 + parallelism = {{pipeline_parallelism}} +} + +source { + BenchmarkSource { + plugin_output = "checkpoint_rows" + parallelism = {{pipeline_parallelism}} + total_rows = 9223372036854775807 + rate_per_second = {{source_rate_per_second}} + payload_size = {{payload_size}} + start_delay_millis = 0 + emit_batch_size = 1 + } +} + +sink { + BenchmarkSink { + plugin_input = "checkpoint_rows" + parallelism = {{pipeline_parallelism}} + result_path = "{{result_path}}" + run_id = "checkpoint-benchmark" + expected_rows = 9223372036854775807 + rate_per_second = 0 + payload_size = {{payload_size}} + transform_operations = 0 + } +} diff --git a/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/CheckpointingTimeBenchmarkTest.java b/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/CheckpointingTimeBenchmarkTest.java new file mode 100644 index 0000000000..991b670a63 --- /dev/null +++ b/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/CheckpointingTimeBenchmarkTest.java @@ -0,0 +1,72 @@ +/* + * 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; + +import org.junit.jupiter.api.Test; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Threads; + +import java.nio.file.Path; +import java.nio.file.Paths; +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.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class CheckpointingTimeBenchmarkTest { + + @Test + void shouldUseCheckpointRecordSizesAndSeconds() { + assertEquals(1L, SeaTunnelCheckpointEnvironmentContext.DEBLOATING_RECORD_SIZE.getBytes()); + assertEquals( + 1_024L, SeaTunnelCheckpointEnvironmentContext.UNALIGNED_RECORD_SIZE.getBytes()); + assertEquals( + TimeUnit.SECONDS, + CheckpointingTimeBenchmark.class.getAnnotation(OutputTimeUnit.class).value()); + assertEquals( + Mode.AverageTime, + CheckpointingTimeBenchmark.class.getAnnotation(BenchmarkMode.class).value()[0]); + assertEquals(1, CheckpointingTimeBenchmark.class.getAnnotation(Threads.class).value()); + } + + @Test + void shouldRejectInvalidMemorySize() { + assertThrows( + IllegalArgumentException.class, + () -> SeaTunnelCheckpointEnvironmentContext.MemorySize.parse("1tb")); + } + + @Test + void shouldRenderCheckpointJobConfig() { + CheckpointingTimeBenchmarkPipeline pipeline = new CheckpointingTimeBenchmarkPipeline(); + pipeline.recordSize = "1kb"; + Path resultPath = Paths.get("target", "checkpoint-results").toAbsolutePath(); + + String jobConfig = pipeline.createCheckpointJobConfig(resultPath); + + assertTrue(jobConfig.contains("payload_size = 1024")); + assertTrue(jobConfig.contains("rate_per_second = 10000")); + assertTrue(jobConfig.contains("parallelism = 4")); + assertTrue(jobConfig.contains("result_path = \"" + resultPath + "\"")); + assertFalse(jobConfig.contains("{{")); + } +}
