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-12794-d743994181b0ad7c0c6d0f1f97c26583a039ec4e in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 85de8925d0a5d7dfbe3fc34cf1db0d6d8c7c4972 Author: SEZ <[email protected]> AuthorDate: Thu Oct 22 14:18:54 2026 +0000 [Improve][Zeta] Add generatedAt to the checkpoint overview response (#12794) --- docs/en/engines/zeta/rest-api-v2.md | 4 +- docs/zh/engines/zeta/rest-api-v2.md | 4 +- .../rest/service/CheckpointMonitorRestService.java | 16 ++- .../service/CheckpointMonitorRestServiceTest.java | 111 +++++++++++++++++++++ 4 files changed, 132 insertions(+), 3 deletions(-) diff --git a/docs/en/engines/zeta/rest-api-v2.md b/docs/en/engines/zeta/rest-api-v2.md index a56142ff51..97fc473174 100644 --- a/docs/en/engines/zeta/rest-api-v2.md +++ b/docs/en/engines/zeta/rest-api-v2.md @@ -1705,6 +1705,7 @@ More information about `Telemetry` can be found in the [Telemetry](telemetry.md) ```json { "jobId": "1234567890", + "generatedAt": 1720000006500, "updatedAt": 1720000000123, "pipelines": [ { @@ -1767,7 +1768,8 @@ More information about `Telemetry` can be found in the [Telemetry](telemetry.md) | Field | Description | | --- | --- | | `jobId` | Job ID. | -| `updatedAt` | Latest snapshot timestamp (millisecond). | +| `generatedAt` | When this response was produced, in epoch milliseconds, taken from the clock of the member that served it. Always present, including when there is no overview to report. Use this, not `updatedAt`, to work out how long an `inProgress` checkpoint has been running: subtract its `triggerTimestamp` from this value. That subtraction assumes member clocks are NTP-synchronised: when the master cannot be reached the request is served by another member, so the two timestamps can [...] +| `updatedAt` | When the overview was last written, in epoch milliseconds — that is, the time of the most recent checkpoint event (trigger, acknowledgement, completion, failure or restore). It does **not** advance while a checkpoint is simply waiting, so it is not a response timestamp and cannot be used to age an in-progress checkpoint. Absent when the monitor has nothing recorded for the job. | | `pipelines` | List of pipeline statistics. | | `pipelines[].pipelineId` | Pipeline ID. | | `pipelines[].counts.triggered/completed/failed/inProgress/restored` | Checkpoint statistics:<br/>- `triggered`: total triggered checkpoints.<br/>- `completed`: total successful checkpoints.<br/>- `failed`: total failed checkpoints.<br/>- `inProgress`: checkpoints currently running.<br/>- `restored`: number of restore (including savepoint) attempts. | diff --git a/docs/zh/engines/zeta/rest-api-v2.md b/docs/zh/engines/zeta/rest-api-v2.md index 033d1714fa..20c22f5fe0 100644 --- a/docs/zh/engines/zeta/rest-api-v2.md +++ b/docs/zh/engines/zeta/rest-api-v2.md @@ -1669,6 +1669,7 @@ logger 会恢复到首次被覆盖之前的状态：配置文件中配置的级 ```json { "jobId": "1234567890", + "generatedAt": 1720000006500, "updatedAt": 1720000000123, "pipelines": [ { @@ -1731,7 +1732,8 @@ logger 会恢复到首次被覆盖之前的状态：配置文件中配置的级 | 字段 | 描述 | | --- | --- | | `jobId` | 作业 ID。 | -| `updatedAt` | 概览最近刷新时间（毫秒时间戳）。 | +| `generatedAt` | 本次响应生成的时间（毫秒时间戳），取自提供该响应的成员节点时钟。始终存在，即使没有可报告的概览。计算 `inProgress` checkpoint 已运行多久时请用它而不是 `updatedAt`：用该值减去 checkpoint 的 `triggerTimestamp`。该相减假定各成员节点时钟已通过 NTP 同步：当 master 不可达时请求会由其他成员提供服务，因此两个时间戳可能来自不同节点。 | +| `updatedAt` | 概览最近一次被写入的时间（毫秒时间戳），即最近一次 checkpoint 事件（触发、确认、完成、失败或恢复）的时间。checkpoint 仅在等待时它**不会**前进，因此它不是响应时间戳，不能用来计算进行中 checkpoint 的年龄。当监控服务没有该作业的记录时该字段不存在。 | | `pipelines` | pipeline 统计列表。 | | `pipelines[].pipelineId` | pipeline ID。 | | `pipelines[].counts.triggered/completed/failed/inProgress/restored` | Checkpoint 统计：<br/>- `triggered`：自作业启动以来触发次数。<br/>- `completed`：成功完成次数。<br/>- `failed`：失败次数。<br/>- `inProgress`：当前正在执行的 checkpoint 数量。<br/>- `restored`：触发恢复（包括 savepoint 恢复）的次数。 | diff --git a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/service/CheckpointMonitorRestService.java b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/service/CheckpointMonitorRestService.java index 40d30ca918..4fcfdc90e1 100644 --- a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/service/CheckpointMonitorRestService.java +++ b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/service/CheckpointMonitorRestService.java @@ -42,7 +42,21 @@ public class CheckpointMonitorRestService extends BaseService { public JsonObject getOverview(long jobId) { CheckpointMonitorService monitorService = getMonitorService(); - JsonObject result = new JsonObject().add("jobId", String.valueOf(jobId)); + // `generatedAt` is stamped while the response is being built, so a + // client can work out how old an in-progress checkpoint is from this + // response alone. It is deliberately not the same thing as + // `updatedAt` below: that one is the time of the last recorded + // checkpoint event, so while a checkpoint hangs waiting for its last + // subtask it stops advancing, and subtracting a trigger timestamp + // from it would make a stalled checkpoint look seconds old forever. + // + // Added outside the block below so it is present even when there is + // no overview to report, which is also the case where a client most + // needs to know the answer came from a live server. + JsonObject result = + new JsonObject() + .add("jobId", String.valueOf(jobId)) + .add("generatedAt", System.currentTimeMillis()); if (monitorService == null) { return result; } diff --git a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/service/CheckpointMonitorRestServiceTest.java b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/service/CheckpointMonitorRestServiceTest.java new file mode 100644 index 0000000000..8738dcd2f3 --- /dev/null +++ b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/service/CheckpointMonitorRestServiceTest.java @@ -0,0 +1,111 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.engine.server.rest.service; + +import org.apache.seatunnel.engine.core.checkpoint.CheckpointOverview; +import org.apache.seatunnel.engine.server.SeaTunnelServer; +import org.apache.seatunnel.engine.server.checkpoint.monitor.CheckpointMonitorService; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import com.hazelcast.internal.json.JsonObject; +import com.hazelcast.spi.impl.NodeEngineImpl; + +import java.util.Optional; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * Tests for the {@code /jobs/checkpoints/{jobId}} response envelope. + * + * <p>The field under test is {@code generatedAt}. It exists because a client cannot compute the age + * of an in-progress checkpoint from {@code updatedAt}: that is the time of the last recorded + * checkpoint event, so while a checkpoint hangs waiting for its last subtask it stops advancing, + * and the stall becomes invisible to anything subtracting a trigger timestamp from it. + */ +public class CheckpointMonitorRestServiceTest { + + private static final long JOB_ID = 852L; + + /** + * Overrides the server lookup so the response can be built without a running cluster. {@code + * getSeaTunnelServer} is protected for this reason; {@code getMonitorService} below it is + * private and reaches the mock through it. + */ + private static CheckpointMonitorRestService serviceReturning( + CheckpointMonitorService monitorService) { + SeaTunnelServer server = mock(SeaTunnelServer.class); + when(server.getCheckpointMonitorService()).thenReturn(monitorService); + return new CheckpointMonitorRestService(mock(NodeEngineImpl.class)) { + @Override + protected SeaTunnelServer getSeaTunnelServer(boolean shouldBeMaster) { + return server; + } + }; + } + + @Test + public void testGeneratedAtIsStampedAtResponseTime() { + CheckpointOverview overview = new CheckpointOverview(JOB_ID); + overview.setUpdatedAt(1_000L); + CheckpointMonitorService monitorService = mock(CheckpointMonitorService.class); + when(monitorService.getOverview(JOB_ID)).thenReturn(Optional.of(overview)); + + long before = System.currentTimeMillis(); + JsonObject result = serviceReturning(monitorService).getOverview(JOB_ID); + long after = System.currentTimeMillis(); + + long generatedAt = result.getLong("generatedAt", -1L); + Assertions.assertTrue( + generatedAt >= before && generatedAt <= after, + "generatedAt should be the response time, but was " + generatedAt); + + // And it must not be confused with updatedAt, which the engine only + // advances when a checkpoint event is recorded. + Assertions.assertEquals(1_000L, result.getLong("updatedAt", -1L)); + Assertions.assertNotEquals(result.getLong("updatedAt", -1L), generatedAt); + } + + @Test + public void testGeneratedAtIsPresentWhenThereIsNoOverview() { + // A job the monitor has nothing recorded for: the response carries no + // pipelines, and a client still needs to know it came from a live + // server rather than a stale cache. + CheckpointMonitorService monitorService = mock(CheckpointMonitorService.class); + when(monitorService.getOverview(JOB_ID)).thenReturn(Optional.empty()); + + JsonObject result = serviceReturning(monitorService).getOverview(JOB_ID); + + Assertions.assertTrue(result.getLong("generatedAt", -1L) > 0L); + Assertions.assertNull(result.get("pipelines")); + } + + @Test + public void testGeneratedAtIsPresentWhenTheMonitorServiceIsOff() { + // enable-http without the monitor service running: getMonitorService() + // returns null and the method returns early, before the branch that + // adds updatedAt and pipelines. + JsonObject result = serviceReturning(null).getOverview(JOB_ID); + + Assertions.assertTrue(result.getLong("generatedAt", -1L) > 0L); + Assertions.assertEquals(String.valueOf(JOB_ID), result.getString("jobId", null)); + Assertions.assertNull(result.get("updatedAt")); + } +}
