This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 0bf03bdedf [#12495] feat(job): Expose job finishedAt across the public
surface (#12497)
0bf03bdedf is described below
commit 0bf03bdedfb6551238e9b13ef4c07033cef2f42a
Author: Jerry Shao <[email protected]>
AuthorDate: Wed Aug 19 13:51:07 2026 +0800
[#12495] feat(job): Expose job finishedAt across the public surface (#12497)
### What changes were proposed in this pull request?
Expose the job's finished-execution time (already tracked internally by
`JobEntity`/`JobPO`/storage) through the public surface:
- `JobHandle.finishedAt()` (Java `default` method) / `finished_at()`
(Python) — non-abstract, raising
`UnsupportedOperationException`/`NotImplementedError` by default so
existing external implementers don't break.
- `JobDTO.finishedAt` (Java `Instant`, Python `datetime`), matching the
existing `Audit.createTime` convention.
- `finishedAt` added to the `Job` schema in `docs/open-api/jobs.yaml`.
- `JobEntity.finishedAtAsInstant()` — single conversion point from the
storage layer's raw epoch-millis + "not finished" sentinel to `Instant`,
used by `JobOperations.toDTO` and `JobInfo.fromJobEntity`.
- `GenericJobHandle` (Java + Python) delegate to the DTO.
- `JobManager.pullAndUpdateJobStatus` now stamps `finishedAt` at the
moment a job transitions to a terminal state
(`SUCCEEDED`/`FAILED`/`CANCELLED`).
- `JobEntity.FINISHED_AT` is now a required field, enforced at build
time, so every construction site must decide the value explicitly;
`equals()`/`hashCode()` now include it.
### Why are the changes needed?
Callers could observe a job's status but had no way to know *when* it
actually finished, failed, or was cancelled — only the storage layer
tracked this internally.
Fix: #12495
### Does this PR introduce _any_ user-facing change?
Yes:
- New `finishedAt` field on the `Job` REST API response / OpenAPI
schema.
- New `JobHandle.finishedAt()` (Java) / `finished_at()` (Python) client
API, returning `null`/`None` if the job hasn't finished yet.
- `JobHandle` implementers outside this repo will get a default that
raises `UnsupportedOperationException`/`NotImplementedError` unless they
override it.
### How was this patch tested?
Unit tests across `common`, `core`, `server`, `clients/client-java`, and
`clients/client-python` (DTO ser/de, `JobEntity` builder/equality,
`JobManager` status-polling, `JobOperations`/`JobPO` conversion), plus
Java and Python integration tests (`JobIT`,
`tests/integration/test_supports_jobs.py`) running real jobs to
completion/failure/cancellation and asserting `finishedAt`.
---------
Co-authored-by: Claude Sonnet 5 <[email protected]>
---
.../java/org/apache/gravitino/job/JobHandle.java | 14 ++
.../apache/gravitino/client/GenericJobHandle.java | 6 +
.../apache/gravitino/client/TestSupportsJobs.java | 10 +-
.../gravitino/client/integration/test/JobIT.java | 10 ++
.../client-python/gravitino/api/job/job_handle.py | 8 +
.../gravitino/client/generic_job_handle.py | 3 +
clients/client-python/gravitino/dto/job/job_dto.py | 26 +++-
.../tests/integration/test_supports_jobs.py | 11 ++
.../tests/unittests/dto/job/test_job_dto_serde.py | 84 +++++++++++
.../tests/unittests/test_supports_jobs.py | 12 +-
.../java/org/apache/gravitino/dto/job/JobDTO.java | 16 +-
.../org/apache/gravitino/dto/job/TestJobDTO.java | 124 +++++++++++++++
.../java/org/apache/gravitino/job/JobManager.java | 10 ++
.../gravitino/listener/api/info/JobInfo.java | 28 +++-
.../java/org/apache/gravitino/meta/JobEntity.java | 27 +++-
.../gravitino/storage/relational/po/JobPO.java | 16 +-
.../org/apache/gravitino/job/TestJobManager.java | 11 +-
.../listener/api/event/TestJobEventDispatcher.java | 5 +
.../org/apache/gravitino/meta/TestJobEntity.java | 166 +++++++++++++++++++++
.../relational/TestJDBCBackendBatchGet.java | 2 +
.../gravitino/storage/relational/po/TestJobPO.java | 28 ++++
.../service/TestEntityChangeLogService.java | 1 +
.../relational/service/TestJobMetaService.java | 2 +
.../service/TestJobTemplateMetaService.java | 6 +
docs/open-api/jobs.yaml | 14 +-
.../gravitino/server/web/rest/JobOperations.java | 3 +-
.../server/web/rest/TestJobOperations.java | 36 ++++-
27 files changed, 649 insertions(+), 30 deletions(-)
diff --git a/api/src/main/java/org/apache/gravitino/job/JobHandle.java
b/api/src/main/java/org/apache/gravitino/job/JobHandle.java
index 9aded575cc..f5fa209199 100644
--- a/api/src/main/java/org/apache/gravitino/job/JobHandle.java
+++ b/api/src/main/java/org/apache/gravitino/job/JobHandle.java
@@ -18,6 +18,9 @@
*/
package org.apache.gravitino.job;
+import java.time.Instant;
+import javax.annotation.Nullable;
+
/**
* JobHandle is an interface that is returned by the job submission, which
provides methods to get
* the job name, job ID, job status, and to add listeners for jobs.
@@ -69,4 +72,15 @@ public interface JobHandle {
* @return the status of the job
*/
Status jobStatus();
+
+ /**
+ * Get the time when the job finished execution.
+ *
+ * @return the finished time of the job, or null if the job has not finished
execution yet
+ */
+ @Nullable
+ default Instant finishedAt() {
+ throw new UnsupportedOperationException(
+ "finishedAt() is not implemented by " + getClass().getName() + ";
override this method");
+ }
}
diff --git
a/clients/client-java/src/main/java/org/apache/gravitino/client/GenericJobHandle.java
b/clients/client-java/src/main/java/org/apache/gravitino/client/GenericJobHandle.java
index 0c270cc429..a0ba2bed45 100644
---
a/clients/client-java/src/main/java/org/apache/gravitino/client/GenericJobHandle.java
+++
b/clients/client-java/src/main/java/org/apache/gravitino/client/GenericJobHandle.java
@@ -18,6 +18,7 @@
*/
package org.apache.gravitino.client;
+import java.time.Instant;
import org.apache.gravitino.dto.job.JobDTO;
import org.apache.gravitino.job.JobHandle;
@@ -44,4 +45,9 @@ public class GenericJobHandle implements JobHandle {
public Status jobStatus() {
return jobDTO.status();
}
+
+ @Override
+ public Instant finishedAt() {
+ return jobDTO.finishedAt();
+ }
}
diff --git
a/clients/client-java/src/test/java/org/apache/gravitino/client/TestSupportsJobs.java
b/clients/client-java/src/test/java/org/apache/gravitino/client/TestSupportsJobs.java
index b73590b8a9..f7c9c994aa 100644
---
a/clients/client-java/src/test/java/org/apache/gravitino/client/TestSupportsJobs.java
+++
b/clients/client-java/src/test/java/org/apache/gravitino/client/TestSupportsJobs.java
@@ -278,7 +278,7 @@ public class TestSupportsJobs extends TestBase {
public void testGetJob() throws JsonProcessingException {
String jobId = "job-1";
String jobTemplateName = "shell-job-template";
- JobDTO expectedJob = newJobDTO(jobId, jobTemplateName);
+ JobDTO expectedJob = newJobDTO(jobId, jobTemplateName, Instant.now());
JobResponse resp = new JobResponse(expectedJob);
buildMockResource(Method.GET, jobRunsPath() + "/" + jobId, null, resp,
HttpStatus.SC_OK);
@@ -340,6 +340,7 @@ public class TestSupportsJobs extends TestBase {
Assertions.assertEquals(expected.jobId(), actual.jobId());
Assertions.assertEquals(expected.jobTemplateName(),
actual.jobTemplateName());
Assertions.assertEquals(expected.status(), actual.jobStatus());
+ Assertions.assertEquals(expected.finishedAt(), actual.finishedAt());
}
private String jobTemplatesPath() {
@@ -381,10 +382,15 @@ public class TestSupportsJobs extends TestBase {
}
private JobDTO newJobDTO(String jobId, String templateName) {
+ return newJobDTO(jobId, templateName, null);
+ }
+
+ private JobDTO newJobDTO(String jobId, String templateName, Instant
finishedAt) {
return new JobDTO(
jobId,
templateName,
JobHandle.Status.QUEUED,
-
AuditDTO.builder().withCreator("test").withCreateTime(Instant.now()).build());
+
AuditDTO.builder().withCreator("test").withCreateTime(Instant.now()).build(),
+ finishedAt);
}
}
diff --git
a/clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/JobIT.java
b/clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/JobIT.java
index 9dfa11ba8a..d523efd90c 100644
---
a/clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/JobIT.java
+++
b/clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/JobIT.java
@@ -288,6 +288,7 @@ public class JobIT extends BaseIT {
ImmutableMap.of("arg1", "value1", "arg2", "success", "env_var",
"value2"));
Assertions.assertEquals(JobHandle.Status.QUEUED, jobHandle1.jobStatus());
Assertions.assertEquals(template.name(), jobHandle1.jobTemplateName());
+ Assertions.assertNull(jobHandle1.finishedAt());
JobHandle jobHandle2 =
metalake.runJob(
@@ -295,6 +296,7 @@ public class JobIT extends BaseIT {
ImmutableMap.of("arg1", "value3", "arg2", "success", "env_var",
"value4"));
Assertions.assertEquals(JobHandle.Status.QUEUED, jobHandle2.jobStatus());
Assertions.assertEquals(template.name(), jobHandle2.jobTemplateName());
+ Assertions.assertNull(jobHandle2.finishedAt());
List<JobHandle> jobs = metalake.listJobs(template.name());
Assertions.assertEquals(2, jobs.size());
@@ -324,6 +326,8 @@ public class JobIT extends BaseIT {
updatedJobs.stream().map(JobHandle::jobStatus).collect(Collectors.toSet());
Assertions.assertEquals(1, jobStatuses.size());
Assertions.assertTrue(jobStatuses.contains(JobHandle.Status.SUCCEEDED));
+ // Finished jobs should carry a non-null finishedAt.
+ updatedJobs.forEach(job -> Assertions.assertNotNull(job.finishedAt()));
}
@Test
@@ -338,6 +342,7 @@ public class JobIT extends BaseIT {
ImmutableMap.of("arg1", "value1", "arg2", "success", "env_var",
"value2"));
Assertions.assertEquals(JobHandle.Status.QUEUED, jobHandle.jobStatus());
Assertions.assertEquals(template.name(), jobHandle.jobTemplateName());
+ Assertions.assertNull(jobHandle.finishedAt());
Awaitility.await()
.atMost(3, TimeUnit.MINUTES)
@@ -350,6 +355,7 @@ public class JobIT extends BaseIT {
JobHandle retrievedJob = metalake.getJob(jobHandle.jobId());
Assertions.assertEquals(jobHandle.jobId(), retrievedJob.jobId());
Assertions.assertEquals(JobHandle.Status.SUCCEEDED,
retrievedJob.jobStatus());
+ Assertions.assertNotNull(retrievedJob.finishedAt());
// Test run a failed job
JobHandle failedJobHandle =
@@ -357,6 +363,7 @@ public class JobIT extends BaseIT {
template.name(),
ImmutableMap.of("arg1", "value1", "arg2", "fail", "env_var",
"value2"));
Assertions.assertEquals(JobHandle.Status.QUEUED,
failedJobHandle.jobStatus());
+ Assertions.assertNull(failedJobHandle.finishedAt());
Awaitility.await()
.atMost(3, TimeUnit.MINUTES)
@@ -369,6 +376,7 @@ public class JobIT extends BaseIT {
JobHandle retrievedFailedJob = metalake.getJob(failedJobHandle.jobId());
Assertions.assertEquals(failedJobHandle.jobId(),
retrievedFailedJob.jobId());
Assertions.assertEquals(JobHandle.Status.FAILED,
retrievedFailedJob.jobStatus());
+ Assertions.assertNotNull(retrievedFailedJob.finishedAt());
// Test get a non-existent job
Assertions.assertThrows(NoSuchJobException.class, () ->
metalake.getJob("non_existent_job_id"));
@@ -386,6 +394,7 @@ public class JobIT extends BaseIT {
ImmutableMap.of("arg1", "value1", "arg2", "success", "env_var",
"value2"));
Assertions.assertEquals(JobHandle.Status.QUEUED, jobHandle.jobStatus());
Assertions.assertEquals(template.name(), jobHandle.jobTemplateName());
+ Assertions.assertNull(jobHandle.finishedAt());
// Cancel the job
metalake.cancelJob(jobHandle.jobId());
@@ -401,6 +410,7 @@ public class JobIT extends BaseIT {
JobHandle retrievedJob = metalake.getJob(jobHandle.jobId());
Assertions.assertEquals(jobHandle.jobId(), retrievedJob.jobId());
Assertions.assertEquals(JobHandle.Status.CANCELLED,
retrievedJob.jobStatus());
+ Assertions.assertNotNull(retrievedJob.finishedAt());
// Test cancel a non-existent job
Assertions.assertThrows(
diff --git a/clients/client-python/gravitino/api/job/job_handle.py
b/clients/client-python/gravitino/api/job/job_handle.py
index b8da91fcd3..e57a0bcaf8 100644
--- a/clients/client-python/gravitino/api/job/job_handle.py
+++ b/clients/client-python/gravitino/api/job/job_handle.py
@@ -16,7 +16,9 @@
# under the License.
from abc import ABC, abstractmethod
+from datetime import datetime
from enum import Enum
+from typing import Optional
class JobHandle(ABC):
@@ -52,3 +54,9 @@ class JobHandle(ABC):
@abstractmethod
def job_status(self) -> Status:
pass
+
+ def finished_at(self) -> Optional[datetime]:
+ """Returns the time the job finished execution, or ``None`` if the job
has not finished
+ execution yet.
+ """
+ raise NotImplementedError("finished_at is not implemented")
diff --git a/clients/client-python/gravitino/client/generic_job_handle.py
b/clients/client-python/gravitino/client/generic_job_handle.py
index 95b568a613..b100867ae8 100644
--- a/clients/client-python/gravitino/client/generic_job_handle.py
+++ b/clients/client-python/gravitino/client/generic_job_handle.py
@@ -32,3 +32,6 @@ class GenericJobHandle(JobHandle):
def job_status(self):
return self._job_dto.status()
+
+ def finished_at(self):
+ return self._job_dto.finished_at()
diff --git a/clients/client-python/gravitino/dto/job/job_dto.py
b/clients/client-python/gravitino/dto/job/job_dto.py
index eab42da57c..8be68d4681 100644
--- a/clients/client-python/gravitino/dto/job/job_dto.py
+++ b/clients/client-python/gravitino/dto/job/job_dto.py
@@ -16,10 +16,17 @@
# under the License.
from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Optional
+
from dataclasses_json import config, DataClassJsonMixin
from gravitino.api.job.job_handle import JobHandle
-from gravitino.dto.audit_dto import AuditDTO
+from gravitino.dto.audit_dto import (
+ AuditDTO,
+ _deserialize_datetime,
+ _serialize_datetime,
+)
@dataclass
@@ -36,6 +43,17 @@ class JobDTO(DataClassJsonMixin):
)
)
_audit: AuditDTO = field(metadata=config(field_name="audit"))
+ _finished_at: Optional[datetime] = field(
+ default=None,
+ metadata=config(
+ field_name="finishedAt",
+ encoder=_serialize_datetime,
+ decoder=_deserialize_datetime,
+ ),
+ )
+
+ def __post_init__(self) -> None:
+ self._finished_at = _deserialize_datetime(self._finished_at)
def job_id(self) -> str:
"""Returns the job ID."""
@@ -53,6 +71,12 @@ class JobDTO(DataClassJsonMixin):
"""Returns the audit information of the job."""
return self._audit
+ def finished_at(self) -> Optional[datetime]:
+ """Returns the time the job finished execution, or ``None`` if the job
has not finished
+ execution yet.
+ """
+ return self._finished_at
+
def validate(self) -> None:
"""Validates the JobDTO, ensuring required fields are present and
non-empty."""
if self._job_id is None or not self._job_id.strip():
diff --git a/clients/client-python/tests/integration/test_supports_jobs.py
b/clients/client-python/tests/integration/test_supports_jobs.py
index e613085cea..a8e3fc56e2 100644
--- a/clients/client-python/tests/integration/test_supports_jobs.py
+++ b/clients/client-python/tests/integration/test_supports_jobs.py
@@ -273,12 +273,14 @@ class TestSupportsJobs(IntegrationTestEnv):
)
self.assertEqual(job_handle1.job_status(), JobHandle.Status.QUEUED)
self.assertEqual(job_handle1.job_template_name(), template.name)
+ self.assertIsNone(job_handle1.finished_at())
job_handle2 = self._metalake.run_job(
template.name, {"arg1": "value3", "arg2": "success", "env_var":
"value4"}
)
self.assertEqual(job_handle2.job_status(), JobHandle.Status.QUEUED)
self.assertEqual(job_handle2.job_template_name(), template.name)
+ self.assertIsNone(job_handle2.finished_at())
# List jobs
jobs = self._metalake.list_jobs(template.name)
@@ -305,6 +307,9 @@ class TestSupportsJobs(IntegrationTestEnv):
job_statuses = {job.job_status() for job in updated_jobs}
self.assertEqual(len(job_statuses), 1)
self.assertIn(JobHandle.Status.SUCCEEDED, job_statuses)
+ # Finished jobs should carry a non-None finished_at.
+ for job in updated_jobs:
+ self.assertIsNotNone(job.finished_at())
def test_run_and_get_job(self):
template = self.builder.with_name("test_run_get").build()
@@ -316,6 +321,7 @@ class TestSupportsJobs(IntegrationTestEnv):
)
self.assertEqual(job_handle.job_status(), JobHandle.Status.QUEUED)
self.assertEqual(job_handle.job_template_name(), template.name)
+ self.assertIsNone(job_handle.finished_at())
# Wait for job to complete
self._wait_until(
@@ -326,12 +332,14 @@ class TestSupportsJobs(IntegrationTestEnv):
retrieved_job = self._metalake.get_job(job_handle.job_id())
self.assertEqual(job_handle.job_id(), retrieved_job.job_id())
self.assertEqual(JobHandle.Status.SUCCEEDED,
retrieved_job.job_status())
+ self.assertIsNotNone(retrieved_job.finished_at())
# Test failed job
failed_job_handle = self._metalake.run_job(
template.name, {"arg1": "value1", "arg2": "fail", "env_var":
"value2"}
)
self.assertEqual(failed_job_handle.job_status(),
JobHandle.Status.QUEUED)
+ self.assertIsNone(failed_job_handle.finished_at())
self._wait_until(
lambda:
self._metalake.get_job(failed_job_handle.job_id()).job_status()
@@ -341,6 +349,7 @@ class TestSupportsJobs(IntegrationTestEnv):
retrieved_failed_job =
self._metalake.get_job(failed_job_handle.job_id())
self.assertEqual(failed_job_handle.job_id(),
retrieved_failed_job.job_id())
self.assertEqual(JobHandle.Status.FAILED,
retrieved_failed_job.job_status())
+ self.assertIsNotNone(retrieved_failed_job.finished_at())
# Test non-existent job
with self.assertRaises(NoSuchJobException):
@@ -356,6 +365,7 @@ class TestSupportsJobs(IntegrationTestEnv):
)
self.assertEqual(job_handle.job_status(), JobHandle.Status.QUEUED)
self.assertEqual(job_handle.job_template_name(), template.name)
+ self.assertIsNone(job_handle.finished_at())
sleep(1)
@@ -371,6 +381,7 @@ class TestSupportsJobs(IntegrationTestEnv):
retrieved_job = self._metalake.get_job(job_handle.job_id())
self.assertEqual(job_handle.job_id(), retrieved_job.job_id())
self.assertEqual(JobHandle.Status.CANCELLED,
retrieved_job.job_status())
+ self.assertIsNotNone(retrieved_job.finished_at())
# Test cancel non-existent job
with self.assertRaises(NoSuchJobException):
diff --git
a/clients/client-python/tests/unittests/dto/job/test_job_dto_serde.py
b/clients/client-python/tests/unittests/dto/job/test_job_dto_serde.py
new file mode 100644
index 0000000000..8db836189d
--- /dev/null
+++ b/clients/client-python/tests/unittests/dto/job/test_job_dto_serde.py
@@ -0,0 +1,84 @@
+# 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.
+import unittest
+from datetime import datetime, timezone
+
+from gravitino.api.job.job_handle import JobHandle
+from gravitino.dto.audit_dto import AuditDTO
+from gravitino.dto.job.job_dto import JobDTO
+
+
+class TestJobDTOSerDe(unittest.TestCase):
+
+ def test_ser_de_with_finished_at(self):
+ finished_at = datetime.now(timezone.utc)
+ job_dto = JobDTO(
+ _job_id="job-123",
+ _job_template_name="test_template",
+ _status=JobHandle.Status.SUCCEEDED,
+ _audit=AuditDTO(_creator="test",
_create_time=datetime.now(timezone.utc)),
+ _finished_at=finished_at,
+ )
+
+ json_str = job_dto.to_json()
+ self.assertIn("finishedAt", json_str)
+
+ deser_job_dto = JobDTO.from_json(json_str)
+ self.assertEqual(job_dto, deser_job_dto)
+ self.assertEqual(finished_at, deser_job_dto.finished_at())
+
+ def test_ser_de_with_none_finished_at(self):
+ job_dto = JobDTO(
+ _job_id="job-456",
+ _job_template_name="test_template",
+ _status=JobHandle.Status.QUEUED,
+ _audit=AuditDTO(_creator="test",
_create_time=datetime.now(timezone.utc)),
+ )
+
+ json_str = job_dto.to_json()
+ deser_job_dto = JobDTO.from_json(json_str)
+ self.assertEqual(job_dto, deser_job_dto)
+ self.assertIsNone(deser_job_dto.finished_at())
+
+ def test_deserialize_from_string(self):
+ json_str = (
+ '{"jobId": "job-789", "jobTemplateName": "test_template", '
+ '"status": "failed", '
+ '"audit": {"creator": "test", "createTime":
"2024-01-01T00:00:00Z"}, '
+ '"finishedAt": "2024-01-01T01:00:00Z"}'
+ )
+
+ job_dto = JobDTO.from_json(json_str)
+
+ self.assertEqual("job-789", job_dto.job_id())
+ self.assertEqual("test_template", job_dto.job_template_name())
+ self.assertEqual(JobHandle.Status.FAILED, job_dto.status())
+ self.assertEqual(
+ datetime(2024, 1, 1, 1, 0, 0, tzinfo=timezone.utc),
job_dto.finished_at()
+ )
+
+ def test_deserialize_from_string_without_finished_at(self):
+ json_str = (
+ '{"jobId": "job-1000", "jobTemplateName": "test_template", '
+ '"status": "queued", '
+ '"audit": {"creator": "test", "createTime":
"2024-01-01T00:00:00Z"}}'
+ )
+
+ job_dto = JobDTO.from_json(json_str, infer_missing=True)
+
+ self.assertEqual("job-1000", job_dto.job_id())
+ self.assertIsNone(job_dto.finished_at())
diff --git a/clients/client-python/tests/unittests/test_supports_jobs.py
b/clients/client-python/tests/unittests/test_supports_jobs.py
index a24766d714..26458c32c1 100644
--- a/clients/client-python/tests/unittests/test_supports_jobs.py
+++ b/clients/client-python/tests/unittests/test_supports_jobs.py
@@ -15,7 +15,9 @@
# specific language governing permissions and limitations
# under the License.
import unittest
+from datetime import datetime, timezone
from http.client import HTTPResponse
+from typing import Optional
from unittest.mock import Mock, patch
from gravitino import GravitinoClient
@@ -205,7 +207,9 @@ class TestSupportsJobs(unittest.TestCase):
)
job_template_name = "test_shell_job"
- job_dto = self._new_job_dto(job_template_name)
+ job_dto = self._new_job_dto(
+ job_template_name, finished_at=datetime.now(timezone.utc)
+ )
resp = JobResponse(_job=job_dto, _code=0)
mock_resp = self._mock_http_response(resp.to_json())
@@ -310,15 +314,19 @@ class TestSupportsJobs(unittest.TestCase):
mock_resp = Response(mock_http_resp)
return mock_resp
- def _new_job_dto(self, job_template_name: str) -> JobDTO:
+ def _new_job_dto(
+ self, job_template_name: str, finished_at: Optional[datetime] = None
+ ) -> JobDTO:
return JobDTO(
_job_id="job-123",
_job_template_name=job_template_name,
_status=JobHandle.Status.QUEUED,
_audit=AuditDTO(_creator="test",
_create_time="2023-10-01T00:00:00Z"),
+ _finished_at=finished_at,
)
def _compare_job_handle(self, job_handle: JobHandle, job_dto: JobDTO):
self.assertEqual(job_handle.job_id(), job_dto.job_id())
self.assertEqual(job_handle.job_template_name(),
job_dto.job_template_name())
self.assertEqual(job_handle.job_status(), job_dto.status())
+ self.assertEqual(job_handle.finished_at(), job_dto.finished_at())
diff --git a/common/src/main/java/org/apache/gravitino/dto/job/JobDTO.java
b/common/src/main/java/org/apache/gravitino/dto/job/JobDTO.java
index 9d57b2c259..d8af081a47 100644
--- a/common/src/main/java/org/apache/gravitino/dto/job/JobDTO.java
+++ b/common/src/main/java/org/apache/gravitino/dto/job/JobDTO.java
@@ -29,6 +29,7 @@ import
com.fasterxml.jackson.databind.annotation.JsonDeserialize;
import com.fasterxml.jackson.databind.annotation.JsonSerialize;
import com.google.common.base.Preconditions;
import java.io.IOException;
+import java.time.Instant;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.ToString;
@@ -58,9 +59,12 @@ public class JobDTO {
@JsonProperty("audit")
private final AuditDTO audit;
+ @JsonProperty("finishedAt")
+ private final Instant finishedAt;
+
/** Default constructor for Jackson deserialization. */
private JobDTO() {
- this(null, null, null, null);
+ this(null, null, null, null, null);
}
/**
@@ -70,12 +74,20 @@ public class JobDTO {
* @param jobTemplateName The name of the job template used for this job.
* @param status The current status of the job.
* @param audit The audit information associated with the job.
+ * @param finishedAt The time when the job finished execution, or null if
the job has not finished
+ * execution yet.
*/
- public JobDTO(String jobId, String jobTemplateName, JobHandle.Status status,
AuditDTO audit) {
+ public JobDTO(
+ String jobId,
+ String jobTemplateName,
+ JobHandle.Status status,
+ AuditDTO audit,
+ Instant finishedAt) {
this.jobId = jobId;
this.jobTemplateName = jobTemplateName;
this.status = status;
this.audit = audit;
+ this.finishedAt = finishedAt;
}
/**
diff --git a/common/src/test/java/org/apache/gravitino/dto/job/TestJobDTO.java
b/common/src/test/java/org/apache/gravitino/dto/job/TestJobDTO.java
new file mode 100644
index 0000000000..b422f96828
--- /dev/null
+++ b/common/src/test/java/org/apache/gravitino/dto/job/TestJobDTO.java
@@ -0,0 +1,124 @@
+/*
+ * 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.gravitino.dto.job;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import java.time.Instant;
+import org.apache.gravitino.dto.AuditDTO;
+import org.apache.gravitino.job.JobHandle;
+import org.apache.gravitino.json.JsonUtils;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestJobDTO {
+
+ @Test
+ public void testSerDeWithFinishedAt() throws JsonProcessingException {
+ Instant finishedAt = Instant.now();
+ JobDTO jobDTO =
+ new JobDTO(
+ "job-123",
+ "testTemplate",
+ JobHandle.Status.SUCCEEDED,
+
AuditDTO.builder().withCreator("test").withCreateTime(Instant.now()).build(),
+ finishedAt);
+
+ Assertions.assertDoesNotThrow(jobDTO::validate);
+
+ String serJson = JsonUtils.objectMapper().writeValueAsString(jobDTO);
+ Assertions.assertTrue(serJson.contains("\"finishedAt\""));
+
+ JobDTO deserJobDTO = JsonUtils.objectMapper().readValue(serJson,
JobDTO.class);
+ Assertions.assertEquals(jobDTO, deserJobDTO);
+ Assertions.assertEquals(finishedAt, deserJobDTO.finishedAt());
+ }
+
+ @Test
+ public void testSerDeWithNullFinishedAt() throws JsonProcessingException {
+ JobDTO jobDTO =
+ new JobDTO(
+ "job-456",
+ "testTemplate",
+ JobHandle.Status.QUEUED,
+
AuditDTO.builder().withCreator("test").withCreateTime(Instant.now()).build(),
+ null);
+
+ Assertions.assertDoesNotThrow(jobDTO::validate);
+
+ String serJson = JsonUtils.objectMapper().writeValueAsString(jobDTO);
+ JobDTO deserJobDTO = JsonUtils.objectMapper().readValue(serJson,
JobDTO.class);
+ Assertions.assertEquals(jobDTO, deserJobDTO);
+ Assertions.assertNull(deserJobDTO.finishedAt());
+ }
+
+ @Test
+ public void testSerializeToString() throws JsonProcessingException {
+ Instant createTime = Instant.parse("2024-01-01T00:00:00Z");
+ Instant finishedAt = Instant.parse("2024-01-01T01:00:00Z");
+ JobDTO jobDTO =
+ new JobDTO(
+ "job-789",
+ "testTemplate",
+ JobHandle.Status.FAILED,
+
AuditDTO.builder().withCreator("test").withCreateTime(createTime).build(),
+ finishedAt);
+
+ String serJson = JsonUtils.objectMapper().writeValueAsString(jobDTO);
+
+ Assertions.assertTrue(serJson.contains("\"jobId\":\"job-789\""));
+
Assertions.assertTrue(serJson.contains("\"jobTemplateName\":\"testTemplate\""));
+ Assertions.assertTrue(serJson.contains("\"status\":\"failed\""));
+
Assertions.assertTrue(serJson.contains("\"finishedAt\":\"2024-01-01T01:00:00Z\""));
+ }
+
+ @Test
+ public void testDeserializeFromString() throws JsonProcessingException {
+ String json =
+ "{"
+ + "\"jobId\":\"job-999\","
+ + "\"jobTemplateName\":\"testTemplate\","
+ + "\"status\":\"succeeded\","
+ +
"\"audit\":{\"creator\":\"test\",\"createTime\":\"2024-01-01T00:00:00Z\"},"
+ + "\"finishedAt\":\"2024-01-01T01:00:00Z\""
+ + "}";
+
+ JobDTO jobDTO = JsonUtils.objectMapper().readValue(json, JobDTO.class);
+
+ Assertions.assertEquals("job-999", jobDTO.jobId());
+ Assertions.assertEquals("testTemplate", jobDTO.jobTemplateName());
+ Assertions.assertEquals(JobHandle.Status.SUCCEEDED, jobDTO.status());
+ Assertions.assertEquals(Instant.parse("2024-01-01T01:00:00Z"),
jobDTO.finishedAt());
+ }
+
+ @Test
+ public void testDeserializeFromStringWithoutFinishedAt() throws
JsonProcessingException {
+ String json =
+ "{"
+ + "\"jobId\":\"job-1000\","
+ + "\"jobTemplateName\":\"testTemplate\","
+ + "\"status\":\"queued\","
+ +
"\"audit\":{\"creator\":\"test\",\"createTime\":\"2024-01-01T00:00:00Z\"}"
+ + "}";
+
+ JobDTO jobDTO = JsonUtils.objectMapper().readValue(json, JobDTO.class);
+
+ Assertions.assertEquals("job-1000", jobDTO.jobId());
+ Assertions.assertNull(jobDTO.finishedAt());
+ }
+}
diff --git a/core/src/main/java/org/apache/gravitino/job/JobManager.java
b/core/src/main/java/org/apache/gravitino/job/JobManager.java
index efa24d70ea..299551dee2 100644
--- a/core/src/main/java/org/apache/gravitino/job/JobManager.java
+++ b/core/src/main/java/org/apache/gravitino/job/JobManager.java
@@ -465,6 +465,8 @@ public class JobManager implements JobOperationDispatcher {
.withCreator(PrincipalUtils.getCurrentPrincipal().getName())
.withCreateTime(Instant.now())
.build())
+ // A newly submitted job is queued, not finished yet.
+ .withFinishedAt(0L)
.build();
try {
@@ -515,6 +517,8 @@ public class JobManager implements JobOperationDispatcher {
.withLastModifier(PrincipalUtils.getCurrentPrincipal().getName())
.withLastModifiedTime(Instant.now())
.build())
+ // CANCELLING is not a terminal state, so the job is not finished
yet.
+ .withFinishedAt(jobEntity.finishedAt())
.build();
return TreeLockUtils.doWithTreeLock(
NameIdentifierUtil.ofJob(metalake, jobId),
@@ -588,6 +592,11 @@ public class JobManager implements JobOperationDispatcher {
}
if (newStatus != job.status()) {
+ boolean isFinished =
+ newStatus == JobHandle.Status.SUCCEEDED
+ || newStatus == JobHandle.Status.FAILED
+ || newStatus == JobHandle.Status.CANCELLED;
+
JobEntity newJobEntity =
JobEntity.builder()
.withId(job.id())
@@ -602,6 +611,7 @@ public class JobManager implements JobOperationDispatcher {
.withLastModifier(PrincipalUtils.getCurrentPrincipal().getName())
.withLastModifiedTime(Instant.now())
.build())
+ .withFinishedAt(isFinished ?
Instant.now().toEpochMilli() : job.finishedAt())
.build();
// Update the job entity with new status.
diff --git
a/core/src/main/java/org/apache/gravitino/listener/api/info/JobInfo.java
b/core/src/main/java/org/apache/gravitino/listener/api/info/JobInfo.java
index 156789b13b..5abe692982 100644
--- a/core/src/main/java/org/apache/gravitino/listener/api/info/JobInfo.java
+++ b/core/src/main/java/org/apache/gravitino/listener/api/info/JobInfo.java
@@ -19,6 +19,8 @@
package org.apache.gravitino.listener.api.info;
+import java.time.Instant;
+import javax.annotation.Nullable;
import org.apache.gravitino.Audit;
import org.apache.gravitino.annotation.DeveloperApi;
import org.apache.gravitino.job.JobHandle;
@@ -38,11 +40,19 @@ public final class JobInfo {
private final Audit audit;
- private JobInfo(String jobId, String jobTemplateName, JobHandle.Status
jobStatus, Audit audit) {
+ private final Instant finishedAt;
+
+ private JobInfo(
+ String jobId,
+ String jobTemplateName,
+ JobHandle.Status jobStatus,
+ Audit audit,
+ Instant finishedAt) {
this.jobId = jobId;
this.jobTemplateName = jobTemplateName;
this.jobStatus = jobStatus;
this.audit = audit;
+ this.finishedAt = finishedAt;
}
/**
@@ -53,7 +63,11 @@ public final class JobInfo {
*/
public static JobInfo fromJobEntity(JobEntity jobEntity) {
return new JobInfo(
- jobEntity.name(), jobEntity.jobTemplateName(), jobEntity.status(),
jobEntity.auditInfo());
+ jobEntity.name(),
+ jobEntity.jobTemplateName(),
+ jobEntity.status(),
+ jobEntity.auditInfo(),
+ jobEntity.finishedAtAsInstant());
}
/**
@@ -91,4 +105,14 @@ public final class JobInfo {
public Audit auditInfo() {
return audit;
}
+
+ /**
+ * Returns the time when the job finished execution.
+ *
+ * @return the finished time of the job, or null if the job has not finished
execution yet
+ */
+ @Nullable
+ public Instant finishedAt() {
+ return finishedAt;
+ }
}
diff --git a/core/src/main/java/org/apache/gravitino/meta/JobEntity.java
b/core/src/main/java/org/apache/gravitino/meta/JobEntity.java
index 6dff1cc84b..604de3c72e 100644
--- a/core/src/main/java/org/apache/gravitino/meta/JobEntity.java
+++ b/core/src/main/java/org/apache/gravitino/meta/JobEntity.java
@@ -20,9 +20,11 @@
package org.apache.gravitino.meta;
import com.google.common.collect.Maps;
+import java.time.Instant;
import java.util.Collections;
import java.util.Map;
import java.util.Objects;
+import javax.annotation.Nullable;
import lombok.ToString;
import org.apache.gravitino.Auditable;
import org.apache.gravitino.Entity;
@@ -49,7 +51,11 @@ public class JobEntity implements Entity, Auditable,
HasIdentifier {
Field.required(
"audit_info", AuditInfo.class, "The audit details of the job
template entity.");
public static final Field FINISHED_AT =
- Field.optional("job_finished_at", Long.class, "The time when the job
finished execution.");
+ Field.required(
+ "job_finished_at",
+ Long.class,
+ "The time when the job finished execution, using the storage layer's
"
+ + "\"not finished\" sentinel (<= 0) when the job has not
finished execution yet.");
private Long id;
private String jobExecutionId;
@@ -104,6 +110,19 @@ public class JobEntity implements Entity, Auditable,
HasIdentifier {
return finishedAt;
}
+ /**
+ * Converts the raw {@code finishedAt} epoch-millis value to an {@link
Instant}, treating {@code
+ * null} or a non-positive value (the "not finished" sentinel used by the
storage layer) as {@code
+ * null}.
+ *
+ * @return the {@link Instant} the job finished execution, or {@code null}
if the job has not
+ * finished execution yet
+ */
+ @Nullable
+ public Instant finishedAtAsInstant() {
+ return (finishedAt == null || finishedAt <= 0) ? null :
Instant.ofEpochMilli(finishedAt);
+ }
+
@Override
public AuditInfo auditInfo() {
return auditInfo;
@@ -129,12 +148,14 @@ public class JobEntity implements Entity, Auditable,
HasIdentifier {
&& Objects.equals(status, that.status)
&& Objects.equals(jobTemplateName, that.jobTemplateName)
&& Objects.equals(namespace, that.namespace)
- && Objects.equals(auditInfo, that.auditInfo);
+ && Objects.equals(auditInfo, that.auditInfo)
+ && Objects.equals(finishedAt, that.finishedAt);
}
@Override
public int hashCode() {
- return Objects.hash(id, jobExecutionId, namespace, status,
jobTemplateName, auditInfo);
+ return Objects.hash(
+ id, jobExecutionId, namespace, status, jobTemplateName, auditInfo,
finishedAt);
}
public static Builder builder() {
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/po/JobPO.java
b/core/src/main/java/org/apache/gravitino/storage/relational/po/JobPO.java
index 8cb9d2e379..a3c0db8a9b 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/po/JobPO.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/po/JobPO.java
@@ -99,23 +99,17 @@ public class JobPO {
}
public static JobPO initializeJobPO(JobEntity jobEntity, JobPOBuilder
builder) {
- // We should not keep the terminated job entities in the database forever,
so we set the
- // current time as the finished timestamp if the job is in a terminal
state,
- // So the entity GC cleaner will clean it up later.
- long finished = DEFAULT_DELETED_AT;
- if (jobEntity.status() == JobHandle.Status.CANCELLED
- || jobEntity.status() == JobHandle.Status.FAILED
- || jobEntity.status() == JobHandle.Status.SUCCEEDED) {
- finished = System.currentTimeMillis();
- }
-
+ // finishedAt is a required field on JobEntity - the caller (e.g.
JobManager, when the job
+ // transitions to a terminal state) is guaranteed to have already set it,
using the storage
+ // layer's "not finished" sentinel (<= 0) otherwise. The entity GC cleaner
relies on this
+ // timestamp being set to clean up terminated jobs later.
try {
return builder
.withJobRunId(jobEntity.id())
.withJobTemplateName(jobEntity.jobTemplateName())
.withJobExecutionId(jobEntity.jobExecutionId())
.withJobRunStatus(jobEntity.status().name())
- .withJobFinishedAt(finished)
+ .withJobFinishedAt(jobEntity.finishedAt())
.withAuditInfo(JsonUtils.anyFieldMapper().writeValueAsString(jobEntity.auditInfo()))
.withCurrentVersion(INIT_VERSION)
.withLastVersion(INIT_VERSION)
diff --git a/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
b/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
index 5edbbc18ef..4665a3320e 100644
--- a/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
+++ b/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
@@ -83,6 +83,7 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
@@ -632,7 +633,15 @@ public class TestJobManager {
when(jobExecutor.getJobStatus(job.jobExecutionId())).thenReturn(JobHandle.Status.SUCCEEDED);
Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
- verify(entityStore, times(1)).put(any(JobEntity.class), anyBoolean());
+
+ ArgumentCaptor<JobEntity> captor =
ArgumentCaptor.forClass(JobEntity.class);
+ verify(entityStore, times(1)).put(captor.capture(), anyBoolean());
+
+ // Once a job transitions to a terminal status, finishedAt must be set.
+ JobEntity updatedJob = captor.getValue();
+ Assertions.assertEquals(JobHandle.Status.SUCCEEDED, updatedJob.status());
+ Assertions.assertNotNull(updatedJob.finishedAt());
+ Assertions.assertTrue(updatedJob.finishedAt() > 0);
}
@Test
diff --git
a/core/src/test/java/org/apache/gravitino/listener/api/event/TestJobEventDispatcher.java
b/core/src/test/java/org/apache/gravitino/listener/api/event/TestJobEventDispatcher.java
index fd87bc0a0c..6f57ced26d 100644
---
a/core/src/test/java/org/apache/gravitino/listener/api/event/TestJobEventDispatcher.java
+++
b/core/src/test/java/org/apache/gravitino/listener/api/event/TestJobEventDispatcher.java
@@ -23,6 +23,7 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
+import java.time.Instant;
import java.util.Collections;
import java.util.Map;
import java.util.Objects;
@@ -486,6 +487,7 @@ public class TestJobEventDispatcher {
Assertions.assertEquals(expected.jobId(), actual.jobId());
Assertions.assertEquals(expected.jobTemplateName(),
actual.jobTemplateName());
Assertions.assertEquals(expected.jobStatus(), actual.jobStatus());
+ Assertions.assertEquals(expected.finishedAt(), actual.finishedAt());
}
private JobOperationDispatcher mockJobDispatcher() {
@@ -547,6 +549,8 @@ public class TestJobEventDispatcher {
when(entity.name()).thenReturn("job-12345");
when(entity.auditInfo()).thenReturn(mock(AuditInfo.class));
when(entity.status()).thenReturn(JobHandle.Status.SUCCEEDED);
+ when(entity.finishedAt()).thenReturn(1700000000000L);
+
when(entity.finishedAtAsInstant()).thenReturn(Instant.ofEpochMilli(1700000000000L));
return entity;
}
@@ -556,6 +560,7 @@ public class TestJobEventDispatcher {
when(info.jobId()).thenReturn("job-12345");
when(info.jobTemplateName()).thenReturn("testJob");
when(info.jobStatus()).thenReturn(JobHandle.Status.SUCCEEDED);
+ when(info.finishedAt()).thenReturn(Instant.ofEpochMilli(1700000000000L));
return info;
}
}
diff --git a/core/src/test/java/org/apache/gravitino/meta/TestJobEntity.java
b/core/src/test/java/org/apache/gravitino/meta/TestJobEntity.java
new file mode 100644
index 0000000000..909798c2e7
--- /dev/null
+++ b/core/src/test/java/org/apache/gravitino/meta/TestJobEntity.java
@@ -0,0 +1,166 @@
+/*
+ * 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.gravitino.meta;
+
+import java.time.Instant;
+import org.apache.gravitino.job.JobHandle;
+import org.apache.gravitino.utils.NamespaceUtil;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestJobEntity {
+
+ private static final AuditInfo AUDIT_INFO =
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build();
+
+ @Test
+ public void testBuildRequiresFinishedAt() {
+ // finishedAt is a required field - it must be explicitly set (using the
storage layer's
+ // "not finished" sentinel, <= 0, when the job hasn't finished),
regardless of status. This
+ // prevents JobPO from having to silently fabricate or default the value.
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ JobEntity.builder()
+ .withId(1L)
+ .withJobExecutionId("job-execution-1")
+ .withJobTemplateName("test-job-template")
+ .withStatus(JobHandle.Status.QUEUED)
+ .withNamespace(NamespaceUtil.ofJob("test"))
+ .withAuditInfo(AUDIT_INFO)
+ .build());
+
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ JobEntity.builder()
+ .withId(2L)
+ .withJobExecutionId("job-execution-2")
+ .withJobTemplateName("test-job-template")
+ .withStatus(JobHandle.Status.SUCCEEDED)
+ .withNamespace(NamespaceUtil.ofJob("test"))
+ .withAuditInfo(AUDIT_INFO)
+ .build());
+ }
+
+ @Test
+ public void testFinishedAt() {
+ JobEntity jobEntity =
+ JobEntity.builder()
+ .withId(1L)
+ .withJobExecutionId("job-execution-1")
+ .withJobTemplateName("test-job-template")
+ .withStatus(JobHandle.Status.SUCCEEDED)
+ .withNamespace(NamespaceUtil.ofJob("test"))
+ .withAuditInfo(AUDIT_INFO)
+ .withFinishedAt(1700000000000L)
+ .build();
+
+ Assertions.assertEquals(1700000000000L, jobEntity.finishedAt());
+ }
+
+ @Test
+ public void testFinishedAtAsInstantWhenNotFinished() {
+ // The storage layer's sentinel (<= 0) means "not finished".
+ JobEntity zeroFinishedAt =
+ JobEntity.builder()
+ .withId(1L)
+ .withJobExecutionId("job-execution-1")
+ .withJobTemplateName("test-job-template")
+ .withStatus(JobHandle.Status.QUEUED)
+ .withNamespace(NamespaceUtil.ofJob("test"))
+ .withAuditInfo(AUDIT_INFO)
+ .withFinishedAt(0L)
+ .build();
+ Assertions.assertNull(zeroFinishedAt.finishedAtAsInstant());
+
+ JobEntity negativeFinishedAt =
+ JobEntity.builder()
+ .withId(2L)
+ .withJobExecutionId("job-execution-2")
+ .withJobTemplateName("test-job-template")
+ .withStatus(JobHandle.Status.STARTED)
+ .withNamespace(NamespaceUtil.ofJob("test"))
+ .withAuditInfo(AUDIT_INFO)
+ .withFinishedAt(-1L)
+ .build();
+ Assertions.assertNull(negativeFinishedAt.finishedAtAsInstant());
+ }
+
+ @Test
+ public void testFinishedAtAsInstantWhenFinished() {
+ long epochMilli = Instant.now().toEpochMilli();
+ JobEntity jobEntity =
+ JobEntity.builder()
+ .withId(1L)
+ .withJobExecutionId("job-execution-1")
+ .withJobTemplateName("test-job-template")
+ .withStatus(JobHandle.Status.SUCCEEDED)
+ .withNamespace(NamespaceUtil.ofJob("test"))
+ .withAuditInfo(AUDIT_INFO)
+ .withFinishedAt(epochMilli)
+ .build();
+
+ Assertions.assertEquals(Instant.ofEpochMilli(epochMilli),
jobEntity.finishedAtAsInstant());
+ }
+
+ @Test
+ public void testEqualsAndHashCodeIncludeFinishedAt() {
+ JobEntity notFinished =
+ JobEntity.builder()
+ .withId(1L)
+ .withJobExecutionId("job-execution-1")
+ .withJobTemplateName("test-job-template")
+ .withStatus(JobHandle.Status.STARTED)
+ .withNamespace(NamespaceUtil.ofJob("test"))
+ .withAuditInfo(AUDIT_INFO)
+ .withFinishedAt(0L)
+ .build();
+
+ JobEntity sameAsNotFinished =
+ JobEntity.builder()
+ .withId(1L)
+ .withJobExecutionId("job-execution-1")
+ .withJobTemplateName("test-job-template")
+ .withStatus(JobHandle.Status.STARTED)
+ .withNamespace(NamespaceUtil.ofJob("test"))
+ .withAuditInfo(AUDIT_INFO)
+ .withFinishedAt(0L)
+ .build();
+
+ Assertions.assertEquals(notFinished, sameAsNotFinished);
+ Assertions.assertEquals(notFinished.hashCode(),
sameAsNotFinished.hashCode());
+
+ // Same identity/status/audit but a different finishedAt (e.g. the same
job just after it
+ // transitioned to a terminal state) must not compare equal.
+ JobEntity finished =
+ JobEntity.builder()
+ .withId(1L)
+ .withJobExecutionId("job-execution-1")
+ .withJobTemplateName("test-job-template")
+ .withStatus(JobHandle.Status.STARTED)
+ .withNamespace(NamespaceUtil.ofJob("test"))
+ .withAuditInfo(AUDIT_INFO)
+ .withFinishedAt(1700000000000L)
+ .build();
+
+ Assertions.assertNotEquals(notFinished, finished);
+ Assertions.assertNotEquals(notFinished.hashCode(), finished.hashCode());
+ }
+}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/TestJDBCBackendBatchGet.java
b/core/src/test/java/org/apache/gravitino/storage/relational/TestJDBCBackendBatchGet.java
index a7410990b3..a67f40d950 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/TestJDBCBackendBatchGet.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/TestJDBCBackendBatchGet.java
@@ -611,6 +611,7 @@ public class TestJDBCBackendBatchGet extends
TestJDBCBackend {
.withStatus(JobHandle.Status.STARTED)
.withJobTemplateName("template1")
.withAuditInfo(AUDIT_INFO)
+ .withFinishedAt(0L)
.build();
JobEntity job2 =
JobEntity.builder()
@@ -620,6 +621,7 @@ public class TestJDBCBackendBatchGet extends
TestJDBCBackend {
.withStatus(JobHandle.Status.QUEUED)
.withJobTemplateName("template2")
.withAuditInfo(AUDIT_INFO)
+ .withFinishedAt(0L)
.build();
backend.insert(job1, false);
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/po/TestJobPO.java
b/core/src/test/java/org/apache/gravitino/storage/relational/po/TestJobPO.java
index 3fd8f9988b..ea7c2c2fc7 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/po/TestJobPO.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/po/TestJobPO.java
@@ -128,6 +128,7 @@ public class TestJobPO {
.withNamespace(NamespaceUtil.ofJob("test"))
.withAuditInfo(
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .withFinishedAt(0L)
.build();
JobPO.JobPOBuilder builder = JobPO.builder().withMetalakeId(1L);
@@ -140,5 +141,32 @@ public class TestJobPO {
Assertions.assertEquals(jobEntity.status(), resultEntity.status());
Assertions.assertEquals(jobEntity.namespace(), resultEntity.namespace());
Assertions.assertEquals(jobEntity.auditInfo().creator(),
resultEntity.auditInfo().creator());
+ // A queued job has no finishedAt, so it should round-trip as the "not
finished" sentinel.
+ Assertions.assertEquals(0L, resultEntity.finishedAt());
+ }
+
+ @Test
+ public void testJobPOFinishedAt() {
+ // initializeJobPO must trust the finishedAt already set on the entity by
the caller (e.g.
+ // JobManager), rather than deriving it independently from the job's
status.
+ long finishedAt = Instant.now().toEpochMilli();
+ JobEntity jobEntity =
+ JobEntity.builder()
+ .withId(1L)
+ .withJobExecutionId("job-execution-1")
+ .withJobTemplateName("test-job-template")
+ .withStatus(JobHandle.Status.SUCCEEDED)
+ .withNamespace(NamespaceUtil.ofJob("test"))
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .withFinishedAt(finishedAt)
+ .build();
+
+ JobPO.JobPOBuilder builder = JobPO.builder().withMetalakeId(1L);
+ JobPO jobPO = JobPO.initializeJobPO(jobEntity, builder);
+ JobEntity resultEntity = JobPO.fromJobPO(jobPO,
NamespaceUtil.ofJob("test"));
+
+ Assertions.assertEquals(finishedAt, jobPO.jobFinishedAt());
+ Assertions.assertEquals(finishedAt, resultEntity.finishedAt());
}
}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestEntityChangeLogService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestEntityChangeLogService.java
index 8611b77db5..70337d5fff 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestEntityChangeLogService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestEntityChangeLogService.java
@@ -652,6 +652,7 @@ public class TestEntityChangeLogService extends
TestJDBCBackend {
.withJobTemplateName(job.jobTemplateName())
.withStatus(JobHandle.Status.STARTED)
.withAuditInfo(AUDIT_INFO)
+ .withFinishedAt(0L)
.build();
backend.insert(runningJob, true);
assertEntityChange(
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobMetaService.java
index 9136d5561f..82d30c8123 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobMetaService.java
@@ -130,6 +130,7 @@ public class TestJobMetaService extends TestJDBCBackend {
.withNamespace(job.namespace())
.withAuditInfo(job.auditInfo())
.withJobTemplateName(job.jobTemplateName())
+ .withFinishedAt(System.currentTimeMillis())
.build();
Assertions.assertDoesNotThrow(() ->
JobMetaService.getInstance().insertJob(jobOverwrite, true));
JobEntity updatedJob =
@@ -179,6 +180,7 @@ public class TestJobMetaService extends TestJDBCBackend {
.withNamespace(job.namespace())
.withAuditInfo(job.auditInfo())
.withJobTemplateName(job.jobTemplateName())
+ .withFinishedAt(timestamp)
.build();
Assertions.assertDoesNotThrow(() ->
JobMetaService.getInstance().insertJob(updatedJob, true));
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobTemplateMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobTemplateMetaService.java
index 3dfe80f408..e83c846f04 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobTemplateMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobTemplateMetaService.java
@@ -277,6 +277,11 @@ public class TestJobTemplateMetaService extends
TestJDBCBackend {
}
static JobEntity newJobEntity(String templateName, JobHandle.Status status,
String metalake) {
+ boolean isFinished =
+ status == JobHandle.Status.SUCCEEDED
+ || status == JobHandle.Status.FAILED
+ || status == JobHandle.Status.CANCELLED;
+
return JobEntity.builder()
.withId(RandomIdGenerator.INSTANCE.nextId())
.withJobExecutionId(RandomIdGenerator.INSTANCE.nextId() + "")
@@ -284,6 +289,7 @@ public class TestJobTemplateMetaService extends
TestJDBCBackend {
.withJobTemplateName(templateName)
.withStatus(status)
.withAuditInfo(AUDIT_INFO)
+ .withFinishedAt(isFinished ? System.currentTimeMillis() : 0L)
.build();
}
}
diff --git a/docs/open-api/jobs.yaml b/docs/open-api/jobs.yaml
index 70d3a80602..34a3ab2a01 100644
--- a/docs/open-api/jobs.yaml
+++ b/docs/open-api/jobs.yaml
@@ -514,6 +514,11 @@ components:
- "canceled"
audit:
$ref: "./openapi.yaml#/components/schemas/Audit"
+ finishedAt:
+ type: string
+ format: date-time
+ nullable: true
+ description: The time when the job finished execution, or null if
the job has not finished execution yet
TemplateUpdate:
@@ -920,7 +925,8 @@ components:
"audit": {
"createTime": "2025-08-12T02:14:28.205023Z",
"creator": "anonymous"
- }
+ },
+ "finishedAt": "2025-08-12T02:15:03.512847Z"
},
{
"jobId": "job-67890",
@@ -929,7 +935,8 @@ components:
"audit": {
"createTime": "2025-08-12T02:14:28.205023Z",
"creator": "anonymous"
- }
+ },
+ "finishedAt": "2025-08-12T02:15:47.891023Z"
}
]
}
@@ -944,7 +951,8 @@ components:
"audit": {
"createTime": "2025-08-12T02:14:28.205023Z",
"creator": "anonymous"
- }
+ },
+ "finishedAt": "2025-08-12T02:15:03.512847Z"
}
}
diff --git
a/server/src/main/java/org/apache/gravitino/server/web/rest/JobOperations.java
b/server/src/main/java/org/apache/gravitino/server/web/rest/JobOperations.java
index 77fe02e9d7..1a033abda6 100644
---
a/server/src/main/java/org/apache/gravitino/server/web/rest/JobOperations.java
+++
b/server/src/main/java/org/apache/gravitino/server/web/rest/JobOperations.java
@@ -476,7 +476,8 @@ public class JobOperations {
jobEntity.name(),
jobEntity.jobTemplateName(),
jobEntity.status(),
- DTOConverters.toDTO(jobEntity.auditInfo()));
+ DTOConverters.toDTO(jobEntity.auditInfo()),
+ jobEntity.finishedAtAsInstant());
}
private static List<JobDTO> toJobDTOs(List<JobEntity> jobEntities) {
diff --git
a/server/src/test/java/org/apache/gravitino/server/web/rest/TestJobOperations.java
b/server/src/test/java/org/apache/gravitino/server/web/rest/TestJobOperations.java
index 651ab955da..1e2fd543d8 100644
---
a/server/src/test/java/org/apache/gravitino/server/web/rest/TestJobOperations.java
+++
b/server/src/test/java/org/apache/gravitino/server/web/rest/TestJobOperations.java
@@ -42,6 +42,7 @@ import javax.ws.rs.core.Response;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.gravitino.Config;
import org.apache.gravitino.GravitinoEnv;
+import org.apache.gravitino.dto.job.JobDTO;
import org.apache.gravitino.dto.job.JobTemplateDTO;
import org.apache.gravitino.dto.job.ShellTemplateUpdateDTO;
import org.apache.gravitino.dto.requests.JobRunRequest;
@@ -641,7 +642,8 @@ public class TestJobOperations extends JerseyTest {
String templateName = "shell_template_1";
JobEntity job1 = newJobEntity(templateName, JobHandle.Status.QUEUED);
JobEntity job2 = newJobEntity(templateName, JobHandle.Status.STARTED);
- JobEntity job3 = newJobEntity("spark_template_1",
JobHandle.Status.SUCCEEDED);
+ JobEntity job3 =
+ newJobEntity("spark_template_1", JobHandle.Status.SUCCEEDED,
Instant.now().toEpochMilli());
when(jobOperationDispatcher.listJobs(metalake, Optional.empty()))
.thenReturn(Lists.newArrayList(job1, job2, job3));
@@ -663,6 +665,13 @@ public class TestJobOperations extends JerseyTest {
Assertions.assertEquals(JobOperations.toDTO(job2),
jobListResponse.getJobs().get(1));
Assertions.assertEquals(JobOperations.toDTO(job3),
jobListResponse.getJobs().get(2));
+ // Not-yet-finished jobs round-trip finishedAt as null over the wire.
+ Assertions.assertNull(jobListResponse.getJobs().get(0).finishedAt());
+ Assertions.assertNull(jobListResponse.getJobs().get(1).finishedAt());
+ // A finished job round-trips its finishedAt as an Instant over the wire.
+ Assertions.assertEquals(
+ Instant.ofEpochMilli(job3.finishedAt()),
jobListResponse.getJobs().get(2).finishedAt());
+
// Test list jobs by template name
when(jobOperationDispatcher.listJobs(metalake, Optional.of(templateName)))
.thenReturn(Lists.newArrayList(job1, job2));
@@ -793,7 +802,8 @@ public class TestJobOperations extends JerseyTest {
@Test
public void testCancelJob() {
- JobEntity job = newJobEntity("shell_template_1", JobHandle.Status.STARTED);
+ JobEntity job =
+ newJobEntity("shell_template_1", JobHandle.Status.CANCELLED,
Instant.now().toEpochMilli());
when(jobOperationDispatcher.cancelJob(metalake,
job.name())).thenReturn(job);
@@ -810,6 +820,8 @@ public class TestJobOperations extends JerseyTest {
JobResponse jobResp = resp.readEntity(JobResponse.class);
Assertions.assertEquals(0, jobResp.getCode());
Assertions.assertEquals(JobOperations.toDTO(job), jobResp.getJob());
+ // A finished (cancelled) job round-trips its finishedAt as an Instant
over the wire.
+ Assertions.assertEquals(Instant.ofEpochMilli(job.finishedAt()),
jobResp.getJob().finishedAt());
// Test throw NoSuchJobException
doThrow(new NoSuchJobException("mock error"))
@@ -830,6 +842,21 @@ public class TestJobOperations extends JerseyTest {
Assertions.assertEquals(NoSuchJobException.class.getSimpleName(),
errorResp.getType());
}
+ @Test
+ public void testToDTOFinishedAt() {
+ // Sentinel value (<= 0) used by the storage layer means "not finished".
+ JobEntity sentinelJob = newJobEntity("shell_template_1",
JobHandle.Status.STARTED, 0L);
+ JobDTO sentinelJobDTO = JobOperations.toDTO(sentinelJob);
+ Assertions.assertNull(sentinelJobDTO.finishedAt());
+
+ // Finished, finishedAt is converted from epoch millis to an Instant.
+ long epochMilli = Instant.now().toEpochMilli();
+ JobEntity finishedJob =
+ newJobEntity("shell_template_1", JobHandle.Status.SUCCEEDED,
epochMilli);
+ JobDTO finishedJobDTO = JobOperations.toDTO(finishedJob);
+ Assertions.assertEquals(Instant.ofEpochMilli(epochMilli),
finishedJobDTO.finishedAt());
+ }
+
private String jobTemplatePath() {
return "/metalakes/" + metalake + "/jobs/templates";
}
@@ -876,6 +903,10 @@ public class TestJobOperations extends JerseyTest {
}
private JobEntity newJobEntity(String templateName, JobHandle.Status status)
{
+ return newJobEntity(templateName, status, 0L);
+ }
+
+ private JobEntity newJobEntity(String templateName, JobHandle.Status status,
Long finishedAt) {
Random rand = new Random();
return JobEntity.builder()
.withId(rand.nextLong())
@@ -885,6 +916,7 @@ public class TestJobOperations extends JerseyTest {
.withStatus(status)
.withAuditInfo(
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .withFinishedAt(finishedAt)
.build();
}
}
