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 62c5a2f2b8 [#12508] feat(job): Expose job queuedAt and startedAt
across the public surface (#12509)
62c5a2f2b8 is described below
commit 62c5a2f2b80b3090e1c115205f579e322ac9a57f
Author: Jerry Shao <[email protected]>
AuthorDate: Thu Aug 20 11:22:37 2026 +0800
[#12508] feat(job): Expose job queuedAt and startedAt across the public
surface (#12509)
### What changes were proposed in this pull request?
Expose two additional job execution timestamps — `queuedAt` (when the
job was submitted) and `startedAt` (when execution began) — through the
public surface, alongside the existing `finishedAt`:
- `queuedAt` — derived from the job's existing audit `createTime`; no
storage schema change.
- `startedAt` — a genuinely new state requiring a `job_started_at`
column in `job_run_meta` (MySQL/PostgreSQL/H2 fresh-install schemas +
the 1.3.0→2.0.0 upgrade scripts), following the same sentinel convention
as `job_finished_at`.
- `JobHandle.queuedAt()`/`startedAt()` (Java `default` methods / Python)
— non-abstract, raising by default so existing external implementers
don't break.
- `JobDTO.queuedAt`/`startedAt`, `JobEntity.startedAtAsInstant()`, wired
through `JobOperations.toDTO`, `JobInfo`, and `GenericJobHandle` (Java +
Python).
- `JobManager.pullAndUpdateJobStatus` stamps `startedAt` at the
`QUEUED`/`CANCELLING`→`STARTED` transition, and falls back to `queuedAt`
when a job reaches `SUCCEEDED`/`FAILED` without ever being observed as
`STARTED` (reaching those states proves it ran). `CANCELLED` is
deliberately excluded from that fallback, since a cancelled job may have
been killed while still `QUEUED` and genuinely never started.
- OpenAPI schema and examples updated for both fields.
### Why are the changes needed?
Callers could already see when a job finished but had no way to see when
it was queued or when execution actually started — useful for measuring
queue wait time and execution duration.
Fix: #12508
### Does this PR introduce _any_ user-facing change?
Yes:
- New `queuedAt`/`startedAt` fields on the `Job` REST API response /
OpenAPI schema.
- New `JobHandle.queuedAt()`/`startedAt()` (Java) /
`queued_at()`/`started_at()` (Python) client API.
- New `job_started_at` column added to the `job_run_meta` table
(fresh-install schema and upgrade script for MySQL/PostgreSQL/H2).
- `JobHandle` implementers outside this repo will get defaults that
raise `UnsupportedOperationException`/`NotImplementedError` unless they
override the new methods.
### 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 including the `startedAt` fallback and its
`CANCELLED` exclusion, `JobOperations`/`JobPO` conversion, real SQL
round-trip through H2), plus Java and Python integration tests (`JobIT`,
`tests/integration/test_supports_jobs.py`) running real jobs to
completion/failure/cancellation and asserting
`queuedAt`/`startedAt`/`finishedAt`.
---------
Co-authored-by: Claude Sonnet 5 <[email protected]>
---
.../java/org/apache/gravitino/job/JobHandle.java | 21 +++
.../apache/gravitino/client/GenericJobHandle.java | 10 ++
.../apache/gravitino/client/TestSupportsJobs.java | 14 +-
.../gravitino/client/integration/test/JobIT.java | 29 +++-
.../client-python/gravitino/api/job/job_handle.py | 10 ++
.../gravitino/client/generic_job_handle.py | 6 +
clients/client-python/gravitino/dto/job/job_dto.py | 28 ++++
.../tests/integration/test_supports_jobs.py | 24 ++-
.../tests/unittests/dto/job/test_job_dto_serde.py | 31 +++-
.../tests/unittests/test_supports_jobs.py | 13 +-
.../java/org/apache/gravitino/dto/job/JobDTO.java | 15 +-
.../org/apache/gravitino/dto/job/TestJobDTO.java | 28 +++-
.../java/org/apache/gravitino/job/JobManager.java | 67 ++++++---
.../gravitino/listener/api/info/JobInfo.java | 25 ++++
.../java/org/apache/gravitino/meta/JobEntity.java | 33 ++++-
.../provider/base/JobMetaBaseSQLProvider.java | 29 ++--
.../postgresql/JobMetaPostgreSQLProvider.java | 9 +-
.../gravitino/storage/relational/po/JobPO.java | 15 +-
.../org/apache/gravitino/job/TestJobManager.java | 152 +++++++++++++++++++
.../listener/api/event/TestJobEventDispatcher.java | 11 +-
.../gravitino/listener/api/info/TestJobInfo.java | 79 ++++++++++
.../org/apache/gravitino/meta/TestJobEntity.java | 161 ++++++++++++++++++++-
.../relational/TestJDBCBackendBatchGet.java | 2 +
.../gravitino/storage/relational/po/TestJobPO.java | 33 ++++-
.../service/TestEntityChangeLogService.java | 1 +
.../relational/service/TestJobMetaService.java | 5 +-
.../service/TestJobTemplateMetaService.java | 3 +
docs/open-api/jobs.yaml | 16 ++
scripts/h2/schema-2.0.0-h2.sql | 1 +
scripts/h2/upgrade-1.3.0-to-2.0.0-h2.sql | 2 +
scripts/mysql/schema-2.0.0-mysql.sql | 1 +
scripts/mysql/upgrade-1.3.0-to-2.0.0-mysql.sql | 3 +
scripts/postgresql/schema-2.0.0-postgresql.sql | 2 +
.../upgrade-1.3.0-to-2.0.0-postgresql.sql | 3 +
.../gravitino/server/web/rest/JobOperations.java | 2 +
.../server/web/rest/TestJobOperations.java | 54 ++++++-
36 files changed, 877 insertions(+), 61 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 f5fa209199..b365ada40a 100644
--- a/api/src/main/java/org/apache/gravitino/job/JobHandle.java
+++ b/api/src/main/java/org/apache/gravitino/job/JobHandle.java
@@ -73,6 +73,27 @@ public interface JobHandle {
*/
Status jobStatus();
+ /**
+ * Get the time when the job was queued for execution.
+ *
+ * @return the queued time of the job
+ */
+ default Instant queuedAt() {
+ throw new UnsupportedOperationException(
+ "queuedAt() is not implemented by " + getClass().getName() + ";
override this method");
+ }
+
+ /**
+ * Get the time when the job started execution.
+ *
+ * @return the started time of the job, or null if the job has not started
execution yet
+ */
+ @Nullable
+ default Instant startedAt() {
+ throw new UnsupportedOperationException(
+ "startedAt() is not implemented by " + getClass().getName() + ";
override this method");
+ }
+
/**
* Get the time when the job finished execution.
*
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 a0ba2bed45..2c2cc829c7 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
@@ -46,6 +46,16 @@ public class GenericJobHandle implements JobHandle {
return jobDTO.status();
}
+ @Override
+ public Instant queuedAt() {
+ return jobDTO.queuedAt();
+ }
+
+ @Override
+ public Instant startedAt() {
+ return jobDTO.startedAt();
+ }
+
@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 f7c9c994aa..4fa6469fa1 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, Instant.now());
+ JobDTO expectedJob = newJobDTO(jobId, jobTemplateName, Instant.now(),
Instant.now());
JobResponse resp = new JobResponse(expectedJob);
buildMockResource(Method.GET, jobRunsPath() + "/" + jobId, null, resp,
HttpStatus.SC_OK);
@@ -340,6 +340,8 @@ 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.queuedAt(), actual.queuedAt());
+ Assertions.assertEquals(expected.startedAt(), actual.startedAt());
Assertions.assertEquals(expected.finishedAt(), actual.finishedAt());
}
@@ -386,11 +388,19 @@ public class TestSupportsJobs extends TestBase {
}
private JobDTO newJobDTO(String jobId, String templateName, Instant
finishedAt) {
+ return newJobDTO(jobId, templateName, null, finishedAt);
+ }
+
+ private JobDTO newJobDTO(
+ String jobId, String templateName, Instant startedAt, Instant
finishedAt) {
+ Instant now = Instant.now();
return new JobDTO(
jobId,
templateName,
JobHandle.Status.QUEUED,
-
AuditDTO.builder().withCreator("test").withCreateTime(Instant.now()).build(),
+ AuditDTO.builder().withCreator("test").withCreateTime(now).build(),
+ now,
+ startedAt,
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 d523efd90c..5813751113 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,8 @@ 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.assertNotNull(jobHandle1.queuedAt());
+ Assertions.assertNull(jobHandle1.startedAt());
Assertions.assertNull(jobHandle1.finishedAt());
JobHandle jobHandle2 =
@@ -296,6 +298,8 @@ 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.assertNotNull(jobHandle2.queuedAt());
+ Assertions.assertNull(jobHandle2.startedAt());
Assertions.assertNull(jobHandle2.finishedAt());
List<JobHandle> jobs = metalake.listJobs(template.name());
@@ -326,8 +330,14 @@ 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()));
+ // Finished jobs should carry a non-null queuedAt/finishedAt. startedAt is
not asserted here:
+ // a fast job can transition QUEUED -> SUCCEEDED between two polls without
ever being
+ // observed as STARTED, in which case startedAt legitimately stays null.
+ updatedJobs.forEach(
+ job -> {
+ Assertions.assertNotNull(job.queuedAt());
+ Assertions.assertNotNull(job.finishedAt());
+ });
}
@Test
@@ -342,6 +352,8 @@ 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.assertNotNull(jobHandle.queuedAt());
+ Assertions.assertNull(jobHandle.startedAt());
Assertions.assertNull(jobHandle.finishedAt());
Awaitility.await()
@@ -355,6 +367,9 @@ 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.queuedAt());
+ // startedAt is not asserted here: the job may transition QUEUED ->
SUCCEEDED between two
+ // polls without ever being observed as STARTED, in which case it
legitimately stays null.
Assertions.assertNotNull(retrievedJob.finishedAt());
// Test run a failed job
@@ -363,6 +378,8 @@ public class JobIT extends BaseIT {
template.name(),
ImmutableMap.of("arg1", "value1", "arg2", "fail", "env_var",
"value2"));
Assertions.assertEquals(JobHandle.Status.QUEUED,
failedJobHandle.jobStatus());
+ Assertions.assertNotNull(failedJobHandle.queuedAt());
+ Assertions.assertNull(failedJobHandle.startedAt());
Assertions.assertNull(failedJobHandle.finishedAt());
Awaitility.await()
@@ -376,6 +393,9 @@ 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.queuedAt());
+ // startedAt is not asserted here: FAILED does not prove the job ever
started (it can be
+ // reached directly from QUEUED, e.g. if the executor fails to launch the
job at all).
Assertions.assertNotNull(retrievedFailedJob.finishedAt());
// Test get a non-existent job
@@ -394,6 +414,8 @@ 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.assertNotNull(jobHandle.queuedAt());
+ Assertions.assertNull(jobHandle.startedAt());
Assertions.assertNull(jobHandle.finishedAt());
// Cancel the job
@@ -410,6 +432,9 @@ 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.queuedAt());
+ // startedAt is not asserted here: the job may be cancelled before it is
ever observed as
+ // STARTED, in which case startedAt legitimately stays null.
Assertions.assertNotNull(retrievedJob.finishedAt());
// Test cancel a non-existent job
diff --git a/clients/client-python/gravitino/api/job/job_handle.py
b/clients/client-python/gravitino/api/job/job_handle.py
index e57a0bcaf8..1caebc28c5 100644
--- a/clients/client-python/gravitino/api/job/job_handle.py
+++ b/clients/client-python/gravitino/api/job/job_handle.py
@@ -55,6 +55,16 @@ class JobHandle(ABC):
def job_status(self) -> Status:
pass
+ def queued_at(self) -> Optional[datetime]:
+ """Returns the time the job was queued for execution."""
+ raise NotImplementedError("queued_at is not implemented")
+
+ def started_at(self) -> Optional[datetime]:
+ """Returns the time the job started execution, or ``None`` if the job
has not started
+ execution yet.
+ """
+ raise NotImplementedError("started_at is not implemented")
+
def finished_at(self) -> Optional[datetime]:
"""Returns the time the job finished execution, or ``None`` if the job
has not finished
execution yet.
diff --git a/clients/client-python/gravitino/client/generic_job_handle.py
b/clients/client-python/gravitino/client/generic_job_handle.py
index b100867ae8..e5c2a10a82 100644
--- a/clients/client-python/gravitino/client/generic_job_handle.py
+++ b/clients/client-python/gravitino/client/generic_job_handle.py
@@ -33,5 +33,11 @@ class GenericJobHandle(JobHandle):
def job_status(self):
return self._job_dto.status()
+ def queued_at(self):
+ return self._job_dto.queued_at()
+
+ def started_at(self):
+ return self._job_dto.started_at()
+
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 8be68d4681..5b2e6bd187 100644
--- a/clients/client-python/gravitino/dto/job/job_dto.py
+++ b/clients/client-python/gravitino/dto/job/job_dto.py
@@ -43,6 +43,22 @@ class JobDTO(DataClassJsonMixin):
)
)
_audit: AuditDTO = field(metadata=config(field_name="audit"))
+ _queued_at: Optional[datetime] = field(
+ default=None,
+ metadata=config(
+ field_name="queuedAt",
+ encoder=_serialize_datetime,
+ decoder=_deserialize_datetime,
+ ),
+ )
+ _started_at: Optional[datetime] = field(
+ default=None,
+ metadata=config(
+ field_name="startedAt",
+ encoder=_serialize_datetime,
+ decoder=_deserialize_datetime,
+ ),
+ )
_finished_at: Optional[datetime] = field(
default=None,
metadata=config(
@@ -53,6 +69,8 @@ class JobDTO(DataClassJsonMixin):
)
def __post_init__(self) -> None:
+ self._queued_at = _deserialize_datetime(self._queued_at)
+ self._started_at = _deserialize_datetime(self._started_at)
self._finished_at = _deserialize_datetime(self._finished_at)
def job_id(self) -> str:
@@ -71,6 +89,16 @@ class JobDTO(DataClassJsonMixin):
"""Returns the audit information of the job."""
return self._audit
+ def queued_at(self) -> Optional[datetime]:
+ """Returns the time the job was queued for execution."""
+ return self._queued_at
+
+ def started_at(self) -> Optional[datetime]:
+ """Returns the time the job started execution, or ``None`` if the job
has not started
+ execution yet.
+ """
+ return self._started_at
+
def finished_at(self) -> Optional[datetime]:
"""Returns the time the job finished execution, or ``None`` if the job
has not finished
execution yet.
diff --git a/clients/client-python/tests/integration/test_supports_jobs.py
b/clients/client-python/tests/integration/test_supports_jobs.py
index a8e3fc56e2..f558bebdb2 100644
--- a/clients/client-python/tests/integration/test_supports_jobs.py
+++ b/clients/client-python/tests/integration/test_supports_jobs.py
@@ -273,6 +273,8 @@ class TestSupportsJobs(IntegrationTestEnv):
)
self.assertEqual(job_handle1.job_status(), JobHandle.Status.QUEUED)
self.assertEqual(job_handle1.job_template_name(), template.name)
+ self.assertIsNotNone(job_handle1.queued_at())
+ self.assertIsNone(job_handle1.started_at())
self.assertIsNone(job_handle1.finished_at())
job_handle2 = self._metalake.run_job(
@@ -280,6 +282,8 @@ class TestSupportsJobs(IntegrationTestEnv):
)
self.assertEqual(job_handle2.job_status(), JobHandle.Status.QUEUED)
self.assertEqual(job_handle2.job_template_name(), template.name)
+ self.assertIsNotNone(job_handle2.queued_at())
+ self.assertIsNone(job_handle2.started_at())
self.assertIsNone(job_handle2.finished_at())
# List jobs
@@ -307,8 +311,11 @@ 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.
+ # Finished jobs should carry a non-None queued_at/finished_at.
started_at is not
+ # asserted here: a fast job can transition QUEUED -> SUCCEEDED between
two polls
+ # without ever being observed as STARTED, in which case it
legitimately stays None.
for job in updated_jobs:
+ self.assertIsNotNone(job.queued_at())
self.assertIsNotNone(job.finished_at())
def test_run_and_get_job(self):
@@ -321,6 +328,8 @@ class TestSupportsJobs(IntegrationTestEnv):
)
self.assertEqual(job_handle.job_status(), JobHandle.Status.QUEUED)
self.assertEqual(job_handle.job_template_name(), template.name)
+ self.assertIsNotNone(job_handle.queued_at())
+ self.assertIsNone(job_handle.started_at())
self.assertIsNone(job_handle.finished_at())
# Wait for job to complete
@@ -332,6 +341,9 @@ 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.queued_at())
+ # started_at is not asserted here: the job may transition QUEUED ->
SUCCEEDED between
+ # two polls without ever being observed as STARTED, in which case it
stays None.
self.assertIsNotNone(retrieved_job.finished_at())
# Test failed job
@@ -339,6 +351,8 @@ class TestSupportsJobs(IntegrationTestEnv):
template.name, {"arg1": "value1", "arg2": "fail", "env_var":
"value2"}
)
self.assertEqual(failed_job_handle.job_status(),
JobHandle.Status.QUEUED)
+ self.assertIsNotNone(failed_job_handle.queued_at())
+ self.assertIsNone(failed_job_handle.started_at())
self.assertIsNone(failed_job_handle.finished_at())
self._wait_until(
@@ -349,6 +363,9 @@ 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.queued_at())
+ # started_at is not asserted here: FAILED does not prove the job ever
started (it can
+ # be reached directly from QUEUED, e.g. if the executor fails to
launch the job).
self.assertIsNotNone(retrieved_failed_job.finished_at())
# Test non-existent job
@@ -365,6 +382,8 @@ class TestSupportsJobs(IntegrationTestEnv):
)
self.assertEqual(job_handle.job_status(), JobHandle.Status.QUEUED)
self.assertEqual(job_handle.job_template_name(), template.name)
+ self.assertIsNotNone(job_handle.queued_at())
+ self.assertIsNone(job_handle.started_at())
self.assertIsNone(job_handle.finished_at())
sleep(1)
@@ -381,6 +400,9 @@ 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.queued_at())
+ # started_at is not asserted here: the job may be cancelled before it
is ever
+ # observed as STARTED, in which case started_at legitimately stays
None.
self.assertIsNotNone(retrieved_job.finished_at())
# Test cancel non-existent job
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
index 8db836189d..ab39812b9e 100644
--- 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
@@ -25,33 +25,45 @@ from gravitino.dto.job.job_dto import JobDTO
class TestJobDTOSerDe(unittest.TestCase):
def test_ser_de_with_finished_at(self):
+ queued_at = datetime.now(timezone.utc)
+ started_at = datetime.now(timezone.utc)
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)),
+ _queued_at=queued_at,
+ _started_at=started_at,
_finished_at=finished_at,
)
json_str = job_dto.to_json()
+ self.assertIn("queuedAt", json_str)
+ self.assertIn("startedAt", json_str)
self.assertIn("finishedAt", json_str)
deser_job_dto = JobDTO.from_json(json_str)
self.assertEqual(job_dto, deser_job_dto)
+ self.assertEqual(queued_at, deser_job_dto.queued_at())
+ self.assertEqual(started_at, deser_job_dto.started_at())
self.assertEqual(finished_at, deser_job_dto.finished_at())
- def test_ser_de_with_none_finished_at(self):
+ def test_ser_de_with_none_started_and_finished_at(self):
+ queued_at = datetime.now(timezone.utc)
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)),
+ _queued_at=queued_at,
)
json_str = job_dto.to_json()
deser_job_dto = JobDTO.from_json(json_str)
self.assertEqual(job_dto, deser_job_dto)
+ self.assertEqual(queued_at, deser_job_dto.queued_at())
+ self.assertIsNone(deser_job_dto.started_at())
self.assertIsNone(deser_job_dto.finished_at())
def test_deserialize_from_string(self):
@@ -59,6 +71,8 @@ class TestJobDTOSerDe(unittest.TestCase):
'{"jobId": "job-789", "jobTemplateName": "test_template", '
'"status": "failed", '
'"audit": {"creator": "test", "createTime":
"2024-01-01T00:00:00Z"}, '
+ '"queuedAt": "2024-01-01T00:00:00Z", '
+ '"startedAt": "2024-01-01T00:30:00Z", '
'"finishedAt": "2024-01-01T01:00:00Z"}'
)
@@ -67,18 +81,29 @@ class TestJobDTOSerDe(unittest.TestCase):
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, 0, 0, 0, tzinfo=timezone.utc),
job_dto.queued_at()
+ )
+ self.assertEqual(
+ datetime(2024, 1, 1, 0, 30, 0, tzinfo=timezone.utc),
job_dto.started_at()
+ )
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):
+ def test_deserialize_from_string_without_started_or_finished_at(self):
json_str = (
'{"jobId": "job-1000", "jobTemplateName": "test_template", '
'"status": "queued", '
- '"audit": {"creator": "test", "createTime":
"2024-01-01T00:00:00Z"}}'
+ '"audit": {"creator": "test", "createTime":
"2024-01-01T00:00:00Z"}, '
+ '"queuedAt": "2024-01-01T00:00:00Z"}'
)
job_dto = JobDTO.from_json(json_str, infer_missing=True)
self.assertEqual("job-1000", job_dto.job_id())
+ self.assertEqual(
+ datetime(2024, 1, 1, 0, 0, 0, tzinfo=timezone.utc),
job_dto.queued_at()
+ )
+ self.assertIsNone(job_dto.started_at())
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 26458c32c1..0ca6367490 100644
--- a/clients/client-python/tests/unittests/test_supports_jobs.py
+++ b/clients/client-python/tests/unittests/test_supports_jobs.py
@@ -208,7 +208,9 @@ class TestSupportsJobs(unittest.TestCase):
job_template_name = "test_shell_job"
job_dto = self._new_job_dto(
- job_template_name, finished_at=datetime.now(timezone.utc)
+ job_template_name,
+ finished_at=datetime.now(timezone.utc),
+ started_at=datetime.now(timezone.utc),
)
resp = JobResponse(_job=job_dto, _code=0)
mock_resp = self._mock_http_response(resp.to_json())
@@ -315,13 +317,18 @@ class TestSupportsJobs(unittest.TestCase):
return mock_resp
def _new_job_dto(
- self, job_template_name: str, finished_at: Optional[datetime] = None
+ self,
+ job_template_name: str,
+ finished_at: Optional[datetime] = None,
+ started_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"),
+ _queued_at=datetime(2023, 10, 1, tzinfo=timezone.utc),
+ _started_at=started_at,
_finished_at=finished_at,
)
@@ -329,4 +336,6 @@ class TestSupportsJobs(unittest.TestCase):
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.queued_at(), job_dto.queued_at())
+ self.assertEqual(job_handle.started_at(), job_dto.started_at())
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 d8af081a47..3c810a0866 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
@@ -59,12 +59,18 @@ public class JobDTO {
@JsonProperty("audit")
private final AuditDTO audit;
+ @JsonProperty("queuedAt")
+ private final Instant queuedAt;
+
+ @JsonProperty("startedAt")
+ private final Instant startedAt;
+
@JsonProperty("finishedAt")
private final Instant finishedAt;
/** Default constructor for Jackson deserialization. */
private JobDTO() {
- this(null, null, null, null, null);
+ this(null, null, null, null, null, null, null);
}
/**
@@ -74,6 +80,9 @@ 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 queuedAt The time when the job was queued for execution.
+ * @param startedAt The time when the job started execution, or null if the
job has not started
+ * execution yet.
* @param finishedAt The time when the job finished execution, or null if
the job has not finished
* execution yet.
*/
@@ -82,11 +91,15 @@ public class JobDTO {
String jobTemplateName,
JobHandle.Status status,
AuditDTO audit,
+ Instant queuedAt,
+ Instant startedAt,
Instant finishedAt) {
this.jobId = jobId;
this.jobTemplateName = jobTemplateName;
this.status = status;
this.audit = audit;
+ this.queuedAt = queuedAt;
+ this.startedAt = startedAt;
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
index b422f96828..bf5e78fa45 100644
--- a/common/src/test/java/org/apache/gravitino/dto/job/TestJobDTO.java
+++ b/common/src/test/java/org/apache/gravitino/dto/job/TestJobDTO.java
@@ -30,6 +30,8 @@ public class TestJobDTO {
@Test
public void testSerDeWithFinishedAt() throws JsonProcessingException {
+ Instant queuedAt = Instant.now();
+ Instant startedAt = Instant.now();
Instant finishedAt = Instant.now();
JobDTO jobDTO =
new JobDTO(
@@ -37,26 +39,35 @@ public class TestJobDTO {
"testTemplate",
JobHandle.Status.SUCCEEDED,
AuditDTO.builder().withCreator("test").withCreateTime(Instant.now()).build(),
+ queuedAt,
+ startedAt,
finishedAt);
Assertions.assertDoesNotThrow(jobDTO::validate);
String serJson = JsonUtils.objectMapper().writeValueAsString(jobDTO);
+ Assertions.assertTrue(serJson.contains("\"queuedAt\""));
+ Assertions.assertTrue(serJson.contains("\"startedAt\""));
Assertions.assertTrue(serJson.contains("\"finishedAt\""));
JobDTO deserJobDTO = JsonUtils.objectMapper().readValue(serJson,
JobDTO.class);
Assertions.assertEquals(jobDTO, deserJobDTO);
+ Assertions.assertEquals(queuedAt, deserJobDTO.queuedAt());
+ Assertions.assertEquals(startedAt, deserJobDTO.startedAt());
Assertions.assertEquals(finishedAt, deserJobDTO.finishedAt());
}
@Test
public void testSerDeWithNullFinishedAt() throws JsonProcessingException {
+ Instant queuedAt = Instant.now();
JobDTO jobDTO =
new JobDTO(
"job-456",
"testTemplate",
JobHandle.Status.QUEUED,
AuditDTO.builder().withCreator("test").withCreateTime(Instant.now()).build(),
+ queuedAt,
+ null,
null);
Assertions.assertDoesNotThrow(jobDTO::validate);
@@ -64,12 +75,16 @@ public class TestJobDTO {
String serJson = JsonUtils.objectMapper().writeValueAsString(jobDTO);
JobDTO deserJobDTO = JsonUtils.objectMapper().readValue(serJson,
JobDTO.class);
Assertions.assertEquals(jobDTO, deserJobDTO);
+ Assertions.assertEquals(queuedAt, deserJobDTO.queuedAt());
+ Assertions.assertNull(deserJobDTO.startedAt());
Assertions.assertNull(deserJobDTO.finishedAt());
}
@Test
public void testSerializeToString() throws JsonProcessingException {
Instant createTime = Instant.parse("2024-01-01T00:00:00Z");
+ Instant queuedAt = Instant.parse("2024-01-01T00:00:00Z");
+ Instant startedAt = Instant.parse("2024-01-01T00:30:00Z");
Instant finishedAt = Instant.parse("2024-01-01T01:00:00Z");
JobDTO jobDTO =
new JobDTO(
@@ -77,6 +92,8 @@ public class TestJobDTO {
"testTemplate",
JobHandle.Status.FAILED,
AuditDTO.builder().withCreator("test").withCreateTime(createTime).build(),
+ queuedAt,
+ startedAt,
finishedAt);
String serJson = JsonUtils.objectMapper().writeValueAsString(jobDTO);
@@ -84,6 +101,8 @@ public class TestJobDTO {
Assertions.assertTrue(serJson.contains("\"jobId\":\"job-789\""));
Assertions.assertTrue(serJson.contains("\"jobTemplateName\":\"testTemplate\""));
Assertions.assertTrue(serJson.contains("\"status\":\"failed\""));
+
Assertions.assertTrue(serJson.contains("\"queuedAt\":\"2024-01-01T00:00:00Z\""));
+
Assertions.assertTrue(serJson.contains("\"startedAt\":\"2024-01-01T00:30:00Z\""));
Assertions.assertTrue(serJson.contains("\"finishedAt\":\"2024-01-01T01:00:00Z\""));
}
@@ -95,6 +114,8 @@ public class TestJobDTO {
+ "\"jobTemplateName\":\"testTemplate\","
+ "\"status\":\"succeeded\","
+
"\"audit\":{\"creator\":\"test\",\"createTime\":\"2024-01-01T00:00:00Z\"},"
+ + "\"queuedAt\":\"2024-01-01T00:00:00Z\","
+ + "\"startedAt\":\"2024-01-01T00:30:00Z\","
+ "\"finishedAt\":\"2024-01-01T01:00:00Z\""
+ "}";
@@ -103,6 +124,8 @@ public class TestJobDTO {
Assertions.assertEquals("job-999", jobDTO.jobId());
Assertions.assertEquals("testTemplate", jobDTO.jobTemplateName());
Assertions.assertEquals(JobHandle.Status.SUCCEEDED, jobDTO.status());
+ Assertions.assertEquals(Instant.parse("2024-01-01T00:00:00Z"),
jobDTO.queuedAt());
+ Assertions.assertEquals(Instant.parse("2024-01-01T00:30:00Z"),
jobDTO.startedAt());
Assertions.assertEquals(Instant.parse("2024-01-01T01:00:00Z"),
jobDTO.finishedAt());
}
@@ -113,12 +136,15 @@ public class TestJobDTO {
+ "\"jobId\":\"job-1000\","
+ "\"jobTemplateName\":\"testTemplate\","
+ "\"status\":\"queued\","
- +
"\"audit\":{\"creator\":\"test\",\"createTime\":\"2024-01-01T00:00:00Z\"}"
+ +
"\"audit\":{\"creator\":\"test\",\"createTime\":\"2024-01-01T00:00:00Z\"},"
+ + "\"queuedAt\":\"2024-01-01T00:00:00Z\""
+ "}";
JobDTO jobDTO = JsonUtils.objectMapper().readValue(json, JobDTO.class);
Assertions.assertEquals("job-1000", jobDTO.jobId());
+ Assertions.assertEquals(Instant.parse("2024-01-01T00:00:00Z"),
jobDTO.queuedAt());
+ Assertions.assertNull(jobDTO.startedAt());
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 299551dee2..aacb39a370 100644
--- a/core/src/main/java/org/apache/gravitino/job/JobManager.java
+++ b/core/src/main/java/org/apache/gravitino/job/JobManager.java
@@ -465,7 +465,8 @@ public class JobManager implements JobOperationDispatcher {
.withCreator(PrincipalUtils.getCurrentPrincipal().getName())
.withCreateTime(Instant.now())
.build())
- // A newly submitted job is queued, not finished yet.
+ // A newly submitted job is queued, not started or finished yet.
+ .withStartedAt(0L)
.withFinishedAt(0L)
.build();
@@ -503,34 +504,42 @@ public class JobManager implements JobOperationDispatcher
{
}
// Update the job status to CANCELING
- JobEntity newJobEntity =
- JobEntity.builder()
- .withId(jobEntity.id())
- .withJobExecutionId(jobEntity.jobExecutionId())
- .withJobTemplateName(jobEntity.jobTemplateName())
- .withStatus(JobHandle.Status.CANCELLING)
- .withNamespace(jobEntity.namespace())
- .withAuditInfo(
- AuditInfo.builder()
- .withCreator(jobEntity.auditInfo().creator())
- .withCreateTime(jobEntity.auditInfo().createTime())
-
.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),
LockType.WRITE,
() -> {
try {
+ // Re-fetch under the lock rather than reusing the snapshot taken
before the
+ // (potentially slow) external cancel call above - a concurrent
status poll could
+ // have persisted a real startedAt/finishedAt in that gap, and
carrying forward the
+ // stale snapshot would clobber it back to the sentinel.
+ JobEntity latestJobEntity = getJob(metalake, jobId);
+ JobEntity newJobEntity =
+ JobEntity.builder()
+ .withId(latestJobEntity.id())
+ .withJobExecutionId(latestJobEntity.jobExecutionId())
+ .withJobTemplateName(latestJobEntity.jobTemplateName())
+ .withStatus(JobHandle.Status.CANCELLING)
+ .withNamespace(latestJobEntity.namespace())
+ .withAuditInfo(
+ AuditInfo.builder()
+ .withCreator(latestJobEntity.auditInfo().creator())
+
.withCreateTime(latestJobEntity.auditInfo().createTime())
+
.withLastModifier(PrincipalUtils.getCurrentPrincipal().getName())
+ .withLastModifiedTime(Instant.now())
+ .build())
+ // CANCELLING is not a terminal state; carry forward
whatever
+ // startedAt/finishedAt the job already had.
+ .withStartedAt(latestJobEntity.startedAt())
+ .withFinishedAt(latestJobEntity.finishedAt())
+ .build();
+
// Update the job entity in the entity store
entityStore.put(newJobEntity, true /* overwrite */);
return newJobEntity;
} catch (IOException e) {
throw new RuntimeException(
- String.format("Failed to update job entity %s to CANCELING
status", newJobEntity),
+ String.format("Failed to update job entity for job %s to
CANCELING status", jobId),
e);
}
});
@@ -592,11 +601,30 @@ public class JobManager implements JobOperationDispatcher
{
}
if (newStatus != job.status()) {
+ boolean isStarted = newStatus == JobHandle.Status.STARTED;
boolean isFinished =
newStatus == JobHandle.Status.SUCCEEDED
|| newStatus == JobHandle.Status.FAILED
|| newStatus == JobHandle.Status.CANCELLED;
+ // Only a directly-observed STARTED transition is trustworthy
evidence of when a
+ // job started. SUCCEEDED/FAILED do not prove the job ever
reached STARTED: FAILED
+ // in particular can be reached directly from QUEUED (e.g.
NoSuchJobException from
+ // the executor, or LocalJobExecutor failing before it records
STARTED), and even
+ // for SUCCEEDED, backfilling startedAt from the queued time
would understate queue
+ // latency and overstate execution duration in any derived
metric. So startedAt is
+ // left unset unless a STARTED transition was actually observed.
+ //
+ // Only stamp startedAt on the first STARTED observation
(job.startedAt() <= 0).
+ // A CANCELLING job already carries forward a real startedAt
from cancelJob, and
+ // since cancellation is asynchronous, a poll can still observe
STARTED while
+ // cancellation is in flight - overwriting the recorded start
time with this later
+ // poll timestamp would lose the accurate value.
+ long startedAt =
+ isStarted && job.startedAt() <= 0
+ ? Instant.now().toEpochMilli()
+ : job.startedAt();
+
JobEntity newJobEntity =
JobEntity.builder()
.withId(job.id())
@@ -611,6 +639,7 @@ public class JobManager implements JobOperationDispatcher {
.withLastModifier(PrincipalUtils.getCurrentPrincipal().getName())
.withLastModifiedTime(Instant.now())
.build())
+ .withStartedAt(startedAt)
.withFinishedAt(isFinished ?
Instant.now().toEpochMilli() : job.finishedAt())
.build();
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 5abe692982..121c1b0112 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
@@ -40,6 +40,8 @@ public final class JobInfo {
private final Audit audit;
+ private final Instant startedAt;
+
private final Instant finishedAt;
private JobInfo(
@@ -47,11 +49,13 @@ public final class JobInfo {
String jobTemplateName,
JobHandle.Status jobStatus,
Audit audit,
+ Instant startedAt,
Instant finishedAt) {
this.jobId = jobId;
this.jobTemplateName = jobTemplateName;
this.jobStatus = jobStatus;
this.audit = audit;
+ this.startedAt = startedAt;
this.finishedAt = finishedAt;
}
@@ -67,6 +71,7 @@ public final class JobInfo {
jobEntity.jobTemplateName(),
jobEntity.status(),
jobEntity.auditInfo(),
+ jobEntity.startedAtAsInstant(),
jobEntity.finishedAtAsInstant());
}
@@ -106,6 +111,26 @@ public final class JobInfo {
return audit;
}
+ /**
+ * Returns the time when the job was queued for execution. This is the same
as the job's creation
+ * time.
+ *
+ * @return the queued time of the job
+ */
+ public Instant queuedAt() {
+ return audit.createTime();
+ }
+
+ /**
+ * Returns the time when the job started execution.
+ *
+ * @return the started time of the job, or null if the job has not started
execution yet
+ */
+ @Nullable
+ public Instant startedAt() {
+ return startedAt;
+ }
+
/**
* Returns the time when the job finished execution.
*
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 604de3c72e..c7c2a0bec5 100644
--- a/core/src/main/java/org/apache/gravitino/meta/JobEntity.java
+++ b/core/src/main/java/org/apache/gravitino/meta/JobEntity.java
@@ -50,6 +50,12 @@ public class JobEntity implements Entity, Auditable,
HasIdentifier {
public static final Field AUDIT_INFO =
Field.required(
"audit_info", AuditInfo.class, "The audit details of the job
template entity.");
+ public static final Field STARTED_AT =
+ Field.required(
+ "job_started_at",
+ Long.class,
+ "The time when the job started execution, using the storage layer's "
+ + "\"not started\" sentinel (<= 0) when the job has not started
execution yet.");
public static final Field FINISHED_AT =
Field.required(
"job_finished_at",
@@ -63,6 +69,7 @@ public class JobEntity implements Entity, Auditable,
HasIdentifier {
private String jobTemplateName;
private Namespace namespace;
private AuditInfo auditInfo;
+ private Long startedAt;
private Long finishedAt;
private JobEntity() {}
@@ -75,6 +82,7 @@ public class JobEntity implements Entity, Auditable,
HasIdentifier {
fields.put(TEMPLATE_NAME, jobTemplateName);
fields.put(STATUS, status);
fields.put(AUDIT_INFO, auditInfo);
+ fields.put(STARTED_AT, startedAt);
fields.put(FINISHED_AT, finishedAt);
return Collections.unmodifiableMap(fields);
}
@@ -106,6 +114,23 @@ public class JobEntity implements Entity, Auditable,
HasIdentifier {
return jobTemplateName;
}
+ public Long startedAt() {
+ return startedAt;
+ }
+
+ /**
+ * Converts the raw {@code startedAt} epoch-millis value to an {@link
Instant}, treating {@code
+ * null} or a non-positive value (the "not started" sentinel used by the
storage layer) as {@code
+ * null}.
+ *
+ * @return the {@link Instant} the job started execution, or {@code null} if
the job has not
+ * started execution yet
+ */
+ @Nullable
+ public Instant startedAtAsInstant() {
+ return (startedAt == null || startedAt <= 0) ? null :
Instant.ofEpochMilli(startedAt);
+ }
+
public Long finishedAt() {
return finishedAt;
}
@@ -149,13 +174,14 @@ public class JobEntity implements Entity, Auditable,
HasIdentifier {
&& Objects.equals(jobTemplateName, that.jobTemplateName)
&& Objects.equals(namespace, that.namespace)
&& Objects.equals(auditInfo, that.auditInfo)
+ && Objects.equals(startedAt, that.startedAt)
&& Objects.equals(finishedAt, that.finishedAt);
}
@Override
public int hashCode() {
return Objects.hash(
- id, jobExecutionId, namespace, status, jobTemplateName, auditInfo,
finishedAt);
+ id, jobExecutionId, namespace, status, jobTemplateName, auditInfo,
startedAt, finishedAt);
}
public static Builder builder() {
@@ -199,6 +225,11 @@ public class JobEntity implements Entity, Auditable,
HasIdentifier {
return this;
}
+ public Builder withStartedAt(Long startedAt) {
+ jobEntity.startedAt = startedAt;
+ return this;
+ }
+
public Builder withFinishedAt(Long finishedAt) {
jobEntity.finishedAt = finishedAt;
return this;
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/JobMetaBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/JobMetaBaseSQLProvider.java
index 054c093311..5babf351fd 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/JobMetaBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/JobMetaBaseSQLProvider.java
@@ -31,16 +31,16 @@ public class JobMetaBaseSQLProvider {
return "INSERT INTO "
+ JobMetaMapper.TABLE_NAME
+ " (job_run_id, job_template_id, metalake_id,"
- + " job_execution_id, job_run_status, job_finished_at, audit_info,
current_version,"
- + " last_version, deleted_at)"
+ + " job_execution_id, job_run_status, job_started_at, job_finished_at,
audit_info,"
+ + " current_version, last_version, deleted_at)"
+ " VALUES (#{jobMeta.jobRunId},"
+ " (SELECT job_template_id FROM "
+ JobTemplateMetaMapper.TABLE_NAME
+ " WHERE job_template_name = #{jobMeta.jobTemplateName}"
+ " AND metalake_id = #{jobMeta.metalakeId} AND deleted_at = 0),"
+ " #{jobMeta.metalakeId}, #{jobMeta.jobExecutionId},"
- + " #{jobMeta.jobRunStatus}, #{jobMeta.jobFinishedAt},
#{jobMeta.auditInfo},"
- + " #{jobMeta.currentVersion}, #{jobMeta.lastVersion},"
+ + " #{jobMeta.jobRunStatus}, #{jobMeta.jobStartedAt},
#{jobMeta.jobFinishedAt},"
+ + " #{jobMeta.auditInfo}, #{jobMeta.currentVersion},
#{jobMeta.lastVersion},"
+ " #{jobMeta.deletedAt})";
}
@@ -48,16 +48,16 @@ public class JobMetaBaseSQLProvider {
return "INSERT INTO "
+ JobMetaMapper.TABLE_NAME
+ " (job_run_id, job_template_id, metalake_id,"
- + " job_execution_id, job_run_status, job_finished_at, audit_info,
current_version,"
- + " last_version, deleted_at)"
+ + " job_execution_id, job_run_status, job_started_at, job_finished_at,
audit_info,"
+ + " current_version, last_version, deleted_at)"
+ " VALUES (#{jobMeta.jobRunId},"
+ " (SELECT job_template_id FROM "
+ JobTemplateMetaMapper.TABLE_NAME
+ " WHERE job_template_name = #{jobMeta.jobTemplateName}"
+ " AND metalake_id = #{jobMeta.metalakeId} AND deleted_at = 0),"
+ " #{jobMeta.metalakeId}, #{jobMeta.jobExecutionId},"
- + " #{jobMeta.jobRunStatus}, #{jobMeta.jobFinishedAt},
#{jobMeta.auditInfo},"
- + " #{jobMeta.currentVersion}, #{jobMeta.lastVersion},"
+ + " #{jobMeta.jobRunStatus}, #{jobMeta.jobStartedAt},
#{jobMeta.jobFinishedAt},"
+ + " #{jobMeta.auditInfo}, #{jobMeta.currentVersion},
#{jobMeta.lastVersion},"
+ " #{jobMeta.deletedAt})"
+ " ON DUPLICATE KEY UPDATE"
+ " job_template_id = (SELECT job_template_id FROM "
@@ -67,6 +67,7 @@ public class JobMetaBaseSQLProvider {
+ " metalake_id = #{jobMeta.metalakeId},"
+ " job_execution_id = #{jobMeta.jobExecutionId},"
+ " job_run_status = #{jobMeta.jobRunStatus},"
+ + " job_started_at = #{jobMeta.jobStartedAt},"
+ " job_finished_at = #{jobMeta.jobFinishedAt},"
+ " audit_info = #{jobMeta.auditInfo},"
+ " current_version = #{jobMeta.currentVersion},"
@@ -77,7 +78,8 @@ public class JobMetaBaseSQLProvider {
public String listJobPOsByMetalake(@Param("metalakeName") String
metalakeName) {
return "SELECT jrm.job_run_id AS jobRunId, jtm.job_template_name AS
jobTemplateName,"
+ " jrm.metalake_id AS metalakeId, jrm.job_execution_id AS
jobExecutionId,"
- + " jrm.job_run_status AS jobRunStatus, jrm.job_finished_at AS
jobFinishedAt,"
+ + " jrm.job_run_status AS jobRunStatus, jrm.job_started_at AS
jobStartedAt,"
+ + " jrm.job_finished_at AS jobFinishedAt,"
+ " jrm.audit_info AS auditInfo,"
+ " jrm.current_version AS currentVersion, jrm.last_version AS
lastVersion,"
+ " jrm.deleted_at AS deletedAt"
@@ -98,7 +100,8 @@ public class JobMetaBaseSQLProvider {
@Param("jobTemplateName") String jobTemplateName) {
return "SELECT jrm.job_run_id AS jobRunId, jtm.job_template_name AS
jobTemplateName,"
+ " jrm.metalake_id AS metalakeId, jrm.job_execution_id AS
jobExecutionId,"
- + " jrm.job_run_status AS jobRunStatus, jrm.job_finished_at AS
jobFinishedAt,"
+ + " jrm.job_run_status AS jobRunStatus, jrm.job_started_at AS
jobStartedAt,"
+ + " jrm.job_finished_at AS jobFinishedAt,"
+ " jrm.audit_info AS auditInfo,"
+ " jrm.current_version AS currentVersion, jrm.last_version AS
lastVersion,"
+ " jrm.deleted_at AS deletedAt"
@@ -118,7 +121,8 @@ public class JobMetaBaseSQLProvider {
@Param("metalakeName") String metalakeName, @Param("jobRunId") Long
jobRunId) {
return "SELECT jrm.job_run_id AS jobRunId, jtm.job_template_name AS
jobTemplateName,"
+ " jrm.metalake_id AS metalakeId, jrm.job_execution_id AS
jobExecutionId,"
- + " jrm.job_run_status AS jobRunStatus, jrm.job_finished_at AS
jobFinishedAt,"
+ + " jrm.job_run_status AS jobRunStatus, jrm.job_started_at AS
jobStartedAt,"
+ + " jrm.job_finished_at AS jobFinishedAt,"
+ " jrm.audit_info AS auditInfo,"
+ " jrm.current_version AS currentVersion, jrm.last_version AS
lastVersion,"
+ " jrm.deleted_at AS deletedAt"
@@ -188,7 +192,8 @@ public class JobMetaBaseSQLProvider {
return "<script>"
+ "SELECT jrm.job_run_id AS jobRunId, jtm.job_template_name AS
jobTemplateName,"
+ " jrm.metalake_id AS metalakeId, jrm.job_execution_id AS
jobExecutionId,"
- + " jrm.job_run_status AS jobRunStatus, jrm.job_finished_at AS
jobFinishedAt,"
+ + " jrm.job_run_status AS jobRunStatus, jrm.job_started_at AS
jobStartedAt,"
+ + " jrm.job_finished_at AS jobFinishedAt,"
+ " jrm.audit_info AS auditInfo,"
+ " jrm.current_version AS currentVersion, jrm.last_version AS
lastVersion,"
+ " jrm.deleted_at AS deletedAt"
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/JobMetaPostgreSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/JobMetaPostgreSQLProvider.java
index 1addd89e54..ffe4ca8ea8 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/JobMetaPostgreSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/JobMetaPostgreSQLProvider.java
@@ -32,16 +32,16 @@ public class JobMetaPostgreSQLProvider extends
JobMetaBaseSQLProvider {
return "INSERT INTO "
+ JobMetaMapper.TABLE_NAME
+ " (job_run_id, job_template_id, metalake_id,"
- + " job_execution_id, job_run_status, job_finished_at, audit_info,
current_version,"
- + " last_version, deleted_at)"
+ + " job_execution_id, job_run_status, job_started_at, job_finished_at,
audit_info,"
+ + " current_version, last_version, deleted_at)"
+ " VALUES (#{jobMeta.jobRunId},"
+ " (SELECT job_template_id FROM "
+ JobTemplateMetaMapper.TABLE_NAME
+ " WHERE job_template_name = #{jobMeta.jobTemplateName}"
+ " AND metalake_id = #{jobMeta.metalakeId} AND deleted_at = 0),"
+ " #{jobMeta.metalakeId}, #{jobMeta.jobExecutionId},"
- + " #{jobMeta.jobRunStatus}, #{jobMeta.jobFinishedAt},
#{jobMeta.auditInfo},"
- + " #{jobMeta.currentVersion}, #{jobMeta.lastVersion},"
+ + " #{jobMeta.jobRunStatus}, #{jobMeta.jobStartedAt},
#{jobMeta.jobFinishedAt},"
+ + " #{jobMeta.auditInfo}, #{jobMeta.currentVersion},
#{jobMeta.lastVersion},"
+ " #{jobMeta.deletedAt})"
+ " ON CONFLICT (job_run_id) DO UPDATE SET"
+ " job_template_id = (SELECT job_template_id FROM "
@@ -51,6 +51,7 @@ public class JobMetaPostgreSQLProvider extends
JobMetaBaseSQLProvider {
+ " metalake_id = #{jobMeta.metalakeId},"
+ " job_execution_id = #{jobMeta.jobExecutionId},"
+ " job_run_status = #{jobMeta.jobRunStatus},"
+ + " job_started_at = #{jobMeta.jobStartedAt},"
+ " job_finished_at = #{jobMeta.jobFinishedAt},"
+ " audit_info = #{jobMeta.auditInfo},"
+ " current_version = #{jobMeta.currentVersion},"
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 a3c0db8a9b..c5db26071e 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
@@ -45,6 +45,7 @@ public class JobPO {
private Long metalakeId;
private String jobExecutionId;
private String jobRunStatus;
+ private Long jobStartedAt;
private Long jobFinishedAt;
private String auditInfo;
private Long currentVersion;
@@ -62,6 +63,7 @@ public class JobPO {
Long metalakeId,
String jobExecutionId,
String jobRunStatus,
+ Long jobStartedAt,
Long jobFinishedAt,
String auditInfo,
Long currentVersion,
@@ -75,6 +77,7 @@ public class JobPO {
StringUtils.isNotBlank(jobExecutionId), "jobExecutionId cannot be
blank");
Preconditions.checkArgument(
StringUtils.isNotBlank(jobRunStatus), "jobRunStatus cannot be blank");
+ Preconditions.checkArgument(jobStartedAt != null, "jobStartedAt cannot be
null");
Preconditions.checkArgument(jobFinishedAt != null, "jobFinishedAt cannot
be null");
Preconditions.checkArgument(StringUtils.isNotBlank(auditInfo), "auditInfo
cannot be blank");
Preconditions.checkArgument(currentVersion != null, "currentVersion cannot
be null");
@@ -86,6 +89,7 @@ public class JobPO {
this.metalakeId = metalakeId;
this.jobExecutionId = jobExecutionId;
this.jobRunStatus = jobRunStatus;
+ this.jobStartedAt = jobStartedAt;
this.jobFinishedAt = jobFinishedAt;
this.auditInfo = auditInfo;
this.currentVersion = currentVersion;
@@ -99,16 +103,18 @@ public class JobPO {
}
public static JobPO initializeJobPO(JobEntity jobEntity, JobPOBuilder
builder) {
- // 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.
+ // startedAt/finishedAt are required fields on JobEntity - the caller
(e.g. JobManager, when
+ // the job transitions to STARTED/a terminal state) is guaranteed to have
already set them,
+ // using the storage layer's "not started"/"not finished" sentinel (<= 0)
otherwise. The
+ // entity GC cleaner relies on the finishedAt 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())
+ .withJobStartedAt(jobEntity.startedAt())
.withJobFinishedAt(jobEntity.finishedAt())
.withAuditInfo(JsonUtils.anyFieldMapper().writeValueAsString(jobEntity.auditInfo()))
.withCurrentVersion(INIT_VERSION)
@@ -129,6 +135,7 @@ public class JobPO {
.withStatus(JobHandle.Status.valueOf(jobPO.jobRunStatus))
.withJobTemplateName(jobPO.jobTemplateName)
.withAuditInfo(JsonUtils.anyFieldMapper().readValue(jobPO.auditInfo,
AuditInfo.class))
+ .withStartedAt(jobPO.jobStartedAt())
.withFinishedAt(jobPO.jobFinishedAt())
.build();
} catch (JsonProcessingException e) {
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 4665a3320e..9b5ec6e27b 100644
--- a/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
+++ b/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
@@ -644,6 +644,157 @@ public class TestJobManager {
Assertions.assertTrue(updatedJob.finishedAt() > 0);
}
+ @Test
+ public void testPullJobStatusStartedAt() throws IOException {
+ JobEntity job =
+ JobEntity.builder()
+ .withId(1L)
+ .withJobExecutionId("job-execution-1")
+ .withNamespace(NamespaceUtil.ofJob(metalake))
+ .withJobTemplateName("shell_job")
+ .withStatus(JobHandle.Status.QUEUED)
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .withStartedAt(0L)
+ .withFinishedAt(0L)
+ .build();
+
+ BaseMetalake mockMetalake =
+ BaseMetalake.builder()
+ .withName(metalake)
+ .withId(idGenerator.nextId())
+ .withVersion(SchemaVersion.V_0_1)
+ .withAuditInfo(AuditInfo.EMPTY)
+ .build();
+ when(entityStore.list(Namespace.empty(), BaseMetalake.class,
Entity.EntityType.METALAKE))
+ .thenReturn(ImmutableList.of(mockMetalake));
+ mockedMetalake
+ .when(() -> MetalakeManager.listInUseMetalakes(entityStore))
+ .thenReturn(ImmutableList.of(metalake));
+
+ when(jobManager.listJobs(metalake,
Optional.empty())).thenReturn(ImmutableList.of(job));
+
+ // QUEUED -> STARTED: startedAt must be set, finishedAt must remain unset.
+
when(jobExecutor.getJobStatus(job.jobExecutionId())).thenReturn(JobHandle.Status.STARTED);
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+
+ ArgumentCaptor<JobEntity> startedCaptor =
ArgumentCaptor.forClass(JobEntity.class);
+ verify(entityStore, times(1)).put(startedCaptor.capture(), anyBoolean());
+ JobEntity startedJob = startedCaptor.getValue();
+ Assertions.assertEquals(JobHandle.Status.STARTED, startedJob.status());
+ Assertions.assertNotNull(startedJob.startedAt());
+ Assertions.assertTrue(startedJob.startedAt() > 0);
+ Assertions.assertEquals(0L, startedJob.finishedAt());
+
+ // STARTED -> SUCCEEDED: finishedAt must be set, and the
previously-recorded startedAt must
+ // be carried forward unchanged, not overwritten.
+ Mockito.clearInvocations(entityStore);
+ when(jobManager.listJobs(metalake,
Optional.empty())).thenReturn(ImmutableList.of(startedJob));
+
when(jobExecutor.getJobStatus(job.jobExecutionId())).thenReturn(JobHandle.Status.SUCCEEDED);
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+
+ ArgumentCaptor<JobEntity> finishedCaptor =
ArgumentCaptor.forClass(JobEntity.class);
+ verify(entityStore, times(1)).put(finishedCaptor.capture(), anyBoolean());
+ JobEntity finishedJob = finishedCaptor.getValue();
+ Assertions.assertEquals(JobHandle.Status.SUCCEEDED, finishedJob.status());
+ Assertions.assertEquals(startedJob.startedAt(), finishedJob.startedAt());
+ Assertions.assertNotNull(finishedJob.finishedAt());
+ Assertions.assertTrue(finishedJob.finishedAt() > 0);
+ }
+
+ @Test
+ public void
testPullJobStatusStartedAtNotBackfilledOnDirectTerminalTransition()
+ throws IOException {
+ // A job that transitions QUEUED -> SUCCEEDED directly (skipping any poll
that observes it
+ // as STARTED) does not prove exactly when it started - e.g.
LocalJobExecutor can also reach
+ // FAILED directly from QUEUED without ever recording STARTED. Backfilling
startedAt from the
+ // queued time would understate queue latency and overstate execution
duration, so startedAt
+ // stays unset (0) unless a STARTED transition was actually observed.
+ Instant queuedAt = Instant.now();
+ JobEntity queuedJob =
+ JobEntity.builder()
+ .withId(1L)
+ .withJobExecutionId("job-execution-1")
+ .withNamespace(NamespaceUtil.ofJob(metalake))
+ .withJobTemplateName("shell_job")
+ .withStatus(JobHandle.Status.QUEUED)
+
.withAuditInfo(AuditInfo.builder().withCreator("test").withCreateTime(queuedAt).build())
+ .withStartedAt(0L)
+ .withFinishedAt(0L)
+ .build();
+
+ BaseMetalake mockMetalake =
+ BaseMetalake.builder()
+ .withName(metalake)
+ .withId(idGenerator.nextId())
+ .withVersion(SchemaVersion.V_0_1)
+ .withAuditInfo(AuditInfo.EMPTY)
+ .build();
+ when(entityStore.list(Namespace.empty(), BaseMetalake.class,
Entity.EntityType.METALAKE))
+ .thenReturn(ImmutableList.of(mockMetalake));
+ mockedMetalake
+ .when(() -> MetalakeManager.listInUseMetalakes(entityStore))
+ .thenReturn(ImmutableList.of(metalake));
+
+ when(jobManager.listJobs(metalake,
Optional.empty())).thenReturn(ImmutableList.of(queuedJob));
+ when(jobExecutor.getJobStatus(queuedJob.jobExecutionId()))
+ .thenReturn(JobHandle.Status.SUCCEEDED);
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+
+ ArgumentCaptor<JobEntity> captor =
ArgumentCaptor.forClass(JobEntity.class);
+ verify(entityStore, times(1)).put(captor.capture(), anyBoolean());
+ JobEntity succeededJob = captor.getValue();
+ Assertions.assertEquals(JobHandle.Status.SUCCEEDED, succeededJob.status());
+ Assertions.assertEquals(0L, succeededJob.startedAt());
+ Assertions.assertNotNull(succeededJob.finishedAt());
+ Assertions.assertTrue(succeededJob.finishedAt() > 0);
+ }
+
+ @Test
+ public void testPullJobStatusStartedAtNotBackfilledOnDirectCancellation()
throws IOException {
+ // CANCELLED does not prove the job ever started (it may have been
cancelled while still
+ // QUEUED), so startedAt must NOT fall back to the queued time here - it
stays unset.
+ JobEntity cancellingJob =
+ JobEntity.builder()
+ .withId(1L)
+ .withJobExecutionId("job-execution-1")
+ .withNamespace(NamespaceUtil.ofJob(metalake))
+ .withJobTemplateName("shell_job")
+ .withStatus(JobHandle.Status.CANCELLING)
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .withStartedAt(0L)
+ .withFinishedAt(0L)
+ .build();
+
+ BaseMetalake mockMetalake =
+ BaseMetalake.builder()
+ .withName(metalake)
+ .withId(idGenerator.nextId())
+ .withVersion(SchemaVersion.V_0_1)
+ .withAuditInfo(AuditInfo.EMPTY)
+ .build();
+ when(entityStore.list(Namespace.empty(), BaseMetalake.class,
Entity.EntityType.METALAKE))
+ .thenReturn(ImmutableList.of(mockMetalake));
+ mockedMetalake
+ .when(() -> MetalakeManager.listInUseMetalakes(entityStore))
+ .thenReturn(ImmutableList.of(metalake));
+
+ when(jobManager.listJobs(metalake, Optional.empty()))
+ .thenReturn(ImmutableList.of(cancellingJob));
+ when(jobExecutor.getJobStatus(cancellingJob.jobExecutionId()))
+ .thenReturn(JobHandle.Status.CANCELLED);
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+
+ ArgumentCaptor<JobEntity> captor =
ArgumentCaptor.forClass(JobEntity.class);
+ verify(entityStore, times(1)).put(captor.capture(), anyBoolean());
+ JobEntity cancelledJob = captor.getValue();
+ Assertions.assertEquals(JobHandle.Status.CANCELLED, cancelledJob.status());
+ Assertions.assertEquals(0L, cancelledJob.startedAt());
+ Assertions.assertNotNull(cancelledJob.finishedAt());
+ Assertions.assertTrue(cancelledJob.finishedAt() > 0);
+ }
+
@Test
public void testCleanUpStagingDirs() throws IOException,
InterruptedException {
JobEntity job = newJobEntity("shell_job", JobHandle.Status.STARTED);
@@ -944,6 +1095,7 @@ public class TestJobManager {
.withJobExecutionId(rand.nextLong() + "")
.withNamespace(NamespaceUtil.ofJob(metalake))
.withJobTemplateName(templateName)
+ .withStartedAt(System.currentTimeMillis())
.withFinishedAt(System.currentTimeMillis())
.withStatus(status)
.withAuditInfo(
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 6f57ced26d..cd9eaf3a7a 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
@@ -487,6 +487,8 @@ public class TestJobEventDispatcher {
Assertions.assertEquals(expected.jobId(), actual.jobId());
Assertions.assertEquals(expected.jobTemplateName(),
actual.jobTemplateName());
Assertions.assertEquals(expected.jobStatus(), actual.jobStatus());
+ Assertions.assertEquals(expected.queuedAt(), actual.queuedAt());
+ Assertions.assertEquals(expected.startedAt(), actual.startedAt());
Assertions.assertEquals(expected.finishedAt(), actual.finishedAt());
}
@@ -544,11 +546,16 @@ public class TestJobEventDispatcher {
}
private JobEntity mockJobEntity() {
+ AuditInfo auditInfo = mock(AuditInfo.class);
+
when(auditInfo.createTime()).thenReturn(Instant.ofEpochMilli(1699999000000L));
+
JobEntity entity = mock(JobEntity.class);
when(entity.jobTemplateName()).thenReturn("testJob");
when(entity.name()).thenReturn("job-12345");
- when(entity.auditInfo()).thenReturn(mock(AuditInfo.class));
+ when(entity.auditInfo()).thenReturn(auditInfo);
when(entity.status()).thenReturn(JobHandle.Status.SUCCEEDED);
+ when(entity.startedAt()).thenReturn(1699999500000L);
+
when(entity.startedAtAsInstant()).thenReturn(Instant.ofEpochMilli(1699999500000L));
when(entity.finishedAt()).thenReturn(1700000000000L);
when(entity.finishedAtAsInstant()).thenReturn(Instant.ofEpochMilli(1700000000000L));
@@ -560,6 +567,8 @@ public class TestJobEventDispatcher {
when(info.jobId()).thenReturn("job-12345");
when(info.jobTemplateName()).thenReturn("testJob");
when(info.jobStatus()).thenReturn(JobHandle.Status.SUCCEEDED);
+ when(info.queuedAt()).thenReturn(Instant.ofEpochMilli(1699999000000L));
+ when(info.startedAt()).thenReturn(Instant.ofEpochMilli(1699999500000L));
when(info.finishedAt()).thenReturn(Instant.ofEpochMilli(1700000000000L));
return info;
}
diff --git
a/core/src/test/java/org/apache/gravitino/listener/api/info/TestJobInfo.java
b/core/src/test/java/org/apache/gravitino/listener/api/info/TestJobInfo.java
new file mode 100644
index 0000000000..833a37fbdc
--- /dev/null
+++ b/core/src/test/java/org/apache/gravitino/listener/api/info/TestJobInfo.java
@@ -0,0 +1,79 @@
+/*
+ * 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.listener.api.info;
+
+import java.time.Instant;
+import org.apache.gravitino.job.JobHandle;
+import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.meta.JobEntity;
+import org.apache.gravitino.utils.NamespaceUtil;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestJobInfo {
+
+ @Test
+ public void testFromJobEntityWhenNotStartedOrFinished() {
+ Instant queuedAt = Instant.now();
+ JobEntity jobEntity =
+ JobEntity.builder()
+ .withId(1L)
+ .withJobExecutionId("job-execution-1")
+ .withJobTemplateName("test-job-template")
+ .withStatus(JobHandle.Status.QUEUED)
+ .withNamespace(NamespaceUtil.ofJob("test"))
+
.withAuditInfo(AuditInfo.builder().withCreator("test").withCreateTime(queuedAt).build())
+ .withStartedAt(0L)
+ .withFinishedAt(0L)
+ .build();
+
+ JobInfo jobInfo = JobInfo.fromJobEntity(jobEntity);
+
+ Assertions.assertEquals(jobEntity.name(), jobInfo.jobId());
+ Assertions.assertEquals(jobEntity.jobTemplateName(),
jobInfo.jobTemplateName());
+ Assertions.assertEquals(jobEntity.status(), jobInfo.jobStatus());
+ Assertions.assertEquals(queuedAt, jobInfo.queuedAt());
+ Assertions.assertNull(jobInfo.startedAt());
+ Assertions.assertNull(jobInfo.finishedAt());
+ }
+
+ @Test
+ public void testFromJobEntityWhenStartedAndFinished() {
+ Instant queuedAt = Instant.now();
+ long startedAt = queuedAt.toEpochMilli() + 1000;
+ long finishedAt = queuedAt.toEpochMilli() + 2000;
+ 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(queuedAt).build())
+ .withStartedAt(startedAt)
+ .withFinishedAt(finishedAt)
+ .build();
+
+ JobInfo jobInfo = JobInfo.fromJobEntity(jobEntity);
+
+ Assertions.assertEquals(queuedAt, jobInfo.queuedAt());
+ Assertions.assertEquals(Instant.ofEpochMilli(startedAt),
jobInfo.startedAt());
+ Assertions.assertEquals(Instant.ofEpochMilli(finishedAt),
jobInfo.finishedAt());
+ }
+}
diff --git a/core/src/test/java/org/apache/gravitino/meta/TestJobEntity.java
b/core/src/test/java/org/apache/gravitino/meta/TestJobEntity.java
index 909798c2e7..cc9854849f 100644
--- a/core/src/test/java/org/apache/gravitino/meta/TestJobEntity.java
+++ b/core/src/test/java/org/apache/gravitino/meta/TestJobEntity.java
@@ -29,11 +29,45 @@ public class TestJobEntity {
private static final AuditInfo AUDIT_INFO =
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build();
+ @Test
+ public void testBuildRequiresStartedAt() {
+ // startedAt is a required field - it must be explicitly set (using the
storage layer's
+ // "not started" sentinel, <= 0, when the job hasn't started), regardless
of status. finishedAt
+ // is set here so only the missing startedAt triggers the failure.
+ 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)
+ .withFinishedAt(0L)
+ .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)
+ .withFinishedAt(1700000000000L)
+ .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.
+ // "not finished" sentinel, <= 0, when the job hasn't finished),
regardless of status.
+ // startedAt is set here so only the missing finishedAt triggers the
failure. Together with
+ // testBuildRequiresStartedAt, this prevents JobPO from having to silently
fabricate or
+ // default either value.
Assertions.assertThrows(
IllegalArgumentException.class,
() ->
@@ -44,6 +78,7 @@ public class TestJobEntity {
.withStatus(JobHandle.Status.QUEUED)
.withNamespace(NamespaceUtil.ofJob("test"))
.withAuditInfo(AUDIT_INFO)
+ .withStartedAt(0L)
.build());
Assertions.assertThrows(
@@ -56,9 +91,27 @@ public class TestJobEntity {
.withStatus(JobHandle.Status.SUCCEEDED)
.withNamespace(NamespaceUtil.ofJob("test"))
.withAuditInfo(AUDIT_INFO)
+ .withStartedAt(1700000000000L)
.build());
}
+ @Test
+ public void testStartedAt() {
+ JobEntity jobEntity =
+ JobEntity.builder()
+ .withId(1L)
+ .withJobExecutionId("job-execution-1")
+ .withJobTemplateName("test-job-template")
+ .withStatus(JobHandle.Status.STARTED)
+ .withNamespace(NamespaceUtil.ofJob("test"))
+ .withAuditInfo(AUDIT_INFO)
+ .withStartedAt(1700000000000L)
+ .withFinishedAt(0L)
+ .build();
+
+ Assertions.assertEquals(1700000000000L, jobEntity.startedAt());
+ }
+
@Test
public void testFinishedAt() {
JobEntity jobEntity =
@@ -69,12 +122,43 @@ public class TestJobEntity {
.withStatus(JobHandle.Status.SUCCEEDED)
.withNamespace(NamespaceUtil.ofJob("test"))
.withAuditInfo(AUDIT_INFO)
+ .withStartedAt(1699999999000L)
.withFinishedAt(1700000000000L)
.build();
Assertions.assertEquals(1700000000000L, jobEntity.finishedAt());
}
+ @Test
+ public void testStartedAtAsInstantWhenNotStarted() {
+ // The storage layer's sentinel (<= 0) means "not started".
+ JobEntity zeroStartedAt =
+ JobEntity.builder()
+ .withId(1L)
+ .withJobExecutionId("job-execution-1")
+ .withJobTemplateName("test-job-template")
+ .withStatus(JobHandle.Status.QUEUED)
+ .withNamespace(NamespaceUtil.ofJob("test"))
+ .withAuditInfo(AUDIT_INFO)
+ .withStartedAt(0L)
+ .withFinishedAt(0L)
+ .build();
+ Assertions.assertNull(zeroStartedAt.startedAtAsInstant());
+
+ JobEntity negativeStartedAt =
+ JobEntity.builder()
+ .withId(2L)
+ .withJobExecutionId("job-execution-2")
+ .withJobTemplateName("test-job-template")
+ .withStatus(JobHandle.Status.QUEUED)
+ .withNamespace(NamespaceUtil.ofJob("test"))
+ .withAuditInfo(AUDIT_INFO)
+ .withStartedAt(-1L)
+ .withFinishedAt(0L)
+ .build();
+ Assertions.assertNull(negativeStartedAt.startedAtAsInstant());
+ }
+
@Test
public void testFinishedAtAsInstantWhenNotFinished() {
// The storage layer's sentinel (<= 0) means "not finished".
@@ -86,6 +170,7 @@ public class TestJobEntity {
.withStatus(JobHandle.Status.QUEUED)
.withNamespace(NamespaceUtil.ofJob("test"))
.withAuditInfo(AUDIT_INFO)
+ .withStartedAt(0L)
.withFinishedAt(0L)
.build();
Assertions.assertNull(zeroFinishedAt.finishedAtAsInstant());
@@ -98,11 +183,30 @@ public class TestJobEntity {
.withStatus(JobHandle.Status.STARTED)
.withNamespace(NamespaceUtil.ofJob("test"))
.withAuditInfo(AUDIT_INFO)
+ .withStartedAt(1700000000000L)
.withFinishedAt(-1L)
.build();
Assertions.assertNull(negativeFinishedAt.finishedAtAsInstant());
}
+ @Test
+ public void testStartedAtAsInstantWhenStarted() {
+ long epochMilli = Instant.now().toEpochMilli();
+ JobEntity jobEntity =
+ JobEntity.builder()
+ .withId(1L)
+ .withJobExecutionId("job-execution-1")
+ .withJobTemplateName("test-job-template")
+ .withStatus(JobHandle.Status.STARTED)
+ .withNamespace(NamespaceUtil.ofJob("test"))
+ .withAuditInfo(AUDIT_INFO)
+ .withStartedAt(epochMilli)
+ .withFinishedAt(0L)
+ .build();
+
+ Assertions.assertEquals(Instant.ofEpochMilli(epochMilli),
jobEntity.startedAtAsInstant());
+ }
+
@Test
public void testFinishedAtAsInstantWhenFinished() {
long epochMilli = Instant.now().toEpochMilli();
@@ -114,12 +218,60 @@ public class TestJobEntity {
.withStatus(JobHandle.Status.SUCCEEDED)
.withNamespace(NamespaceUtil.ofJob("test"))
.withAuditInfo(AUDIT_INFO)
+ .withStartedAt(epochMilli - 1000)
.withFinishedAt(epochMilli)
.build();
Assertions.assertEquals(Instant.ofEpochMilli(epochMilli),
jobEntity.finishedAtAsInstant());
}
+ @Test
+ public void testEqualsAndHashCodeIncludeStartedAt() {
+ JobEntity notStarted =
+ JobEntity.builder()
+ .withId(1L)
+ .withJobExecutionId("job-execution-1")
+ .withJobTemplateName("test-job-template")
+ .withStatus(JobHandle.Status.QUEUED)
+ .withNamespace(NamespaceUtil.ofJob("test"))
+ .withAuditInfo(AUDIT_INFO)
+ .withStartedAt(0L)
+ .withFinishedAt(0L)
+ .build();
+
+ JobEntity sameAsNotStarted =
+ JobEntity.builder()
+ .withId(1L)
+ .withJobExecutionId("job-execution-1")
+ .withJobTemplateName("test-job-template")
+ .withStatus(JobHandle.Status.QUEUED)
+ .withNamespace(NamespaceUtil.ofJob("test"))
+ .withAuditInfo(AUDIT_INFO)
+ .withStartedAt(0L)
+ .withFinishedAt(0L)
+ .build();
+
+ Assertions.assertEquals(notStarted, sameAsNotStarted);
+ Assertions.assertEquals(notStarted.hashCode(),
sameAsNotStarted.hashCode());
+
+ // Same identity/status/audit but a different startedAt (e.g. the same job
just after it
+ // transitioned to STARTED) must not compare equal.
+ JobEntity started =
+ JobEntity.builder()
+ .withId(1L)
+ .withJobExecutionId("job-execution-1")
+ .withJobTemplateName("test-job-template")
+ .withStatus(JobHandle.Status.QUEUED)
+ .withNamespace(NamespaceUtil.ofJob("test"))
+ .withAuditInfo(AUDIT_INFO)
+ .withStartedAt(1700000000000L)
+ .withFinishedAt(0L)
+ .build();
+
+ Assertions.assertNotEquals(notStarted, started);
+ Assertions.assertNotEquals(notStarted.hashCode(), started.hashCode());
+ }
+
@Test
public void testEqualsAndHashCodeIncludeFinishedAt() {
JobEntity notFinished =
@@ -130,6 +282,7 @@ public class TestJobEntity {
.withStatus(JobHandle.Status.STARTED)
.withNamespace(NamespaceUtil.ofJob("test"))
.withAuditInfo(AUDIT_INFO)
+ .withStartedAt(1700000000000L)
.withFinishedAt(0L)
.build();
@@ -141,6 +294,7 @@ public class TestJobEntity {
.withStatus(JobHandle.Status.STARTED)
.withNamespace(NamespaceUtil.ofJob("test"))
.withAuditInfo(AUDIT_INFO)
+ .withStartedAt(1700000000000L)
.withFinishedAt(0L)
.build();
@@ -157,7 +311,8 @@ public class TestJobEntity {
.withStatus(JobHandle.Status.STARTED)
.withNamespace(NamespaceUtil.ofJob("test"))
.withAuditInfo(AUDIT_INFO)
- .withFinishedAt(1700000000000L)
+ .withStartedAt(1700000000000L)
+ .withFinishedAt(1700000001000L)
.build();
Assertions.assertNotEquals(notFinished, finished);
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 a67f40d950..36ddc2e942 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)
+ .withStartedAt(System.currentTimeMillis())
.withFinishedAt(0L)
.build();
JobEntity job2 =
@@ -621,6 +622,7 @@ public class TestJDBCBackendBatchGet extends
TestJDBCBackend {
.withStatus(JobHandle.Status.QUEUED)
.withJobTemplateName("template2")
.withAuditInfo(AUDIT_INFO)
+ .withStartedAt(0L)
.withFinishedAt(0L)
.build();
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 ea7c2c2fc7..76a390b173 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())
+ .withStartedAt(0L)
.withFinishedAt(0L)
.build();
@@ -141,14 +142,43 @@ 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.
+ // A queued job has no startedAt/finishedAt, so both should round-trip as
the "not
+ // started"/"not finished" sentinel.
+ Assertions.assertEquals(0L, resultEntity.startedAt());
Assertions.assertEquals(0L, resultEntity.finishedAt());
}
+ @Test
+ public void testJobPOStartedAt() {
+ // initializeJobPO must trust the startedAt already set on the entity by
the caller (e.g.
+ // JobManager), rather than deriving it independently from the job's
status.
+ long startedAt = Instant.now().toEpochMilli();
+ JobEntity jobEntity =
+ JobEntity.builder()
+ .withId(1L)
+ .withJobExecutionId("job-execution-1")
+ .withJobTemplateName("test-job-template")
+ .withStatus(JobHandle.Status.STARTED)
+ .withNamespace(NamespaceUtil.ofJob("test"))
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .withStartedAt(startedAt)
+ .withFinishedAt(0L)
+ .build();
+
+ JobPO.JobPOBuilder builder = JobPO.builder().withMetalakeId(1L);
+ JobPO jobPO = JobPO.initializeJobPO(jobEntity, builder);
+ JobEntity resultEntity = JobPO.fromJobPO(jobPO,
NamespaceUtil.ofJob("test"));
+
+ Assertions.assertEquals(startedAt, jobPO.jobStartedAt());
+ Assertions.assertEquals(startedAt, resultEntity.startedAt());
+ }
+
@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 startedAt = Instant.now().toEpochMilli() - 1000;
long finishedAt = Instant.now().toEpochMilli();
JobEntity jobEntity =
JobEntity.builder()
@@ -159,6 +189,7 @@ public class TestJobPO {
.withNamespace(NamespaceUtil.ofJob("test"))
.withAuditInfo(
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .withStartedAt(startedAt)
.withFinishedAt(finishedAt)
.build();
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 70337d5fff..1f30e64612 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)
+ .withStartedAt(System.currentTimeMillis())
.withFinishedAt(0L)
.build();
backend.insert(runningJob, true);
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 82d30c8123..a45294e553 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())
+ .withStartedAt(System.currentTimeMillis())
.withFinishedAt(System.currentTimeMillis())
.build();
Assertions.assertDoesNotThrow(() ->
JobMetaService.getInstance().insertJob(jobOverwrite, true));
@@ -138,7 +139,7 @@ public class TestJobMetaService extends TestJDBCBackend {
.getJobByIdentifier(NameIdentifierUtil.ofJob(METALAKE_NAME,
jobOverwrite.name()));
Assertions.assertEquals(jobOverwrite, updatedJob);
- // Test insert and get job with finishedAt
+ // Test insert and get job with startedAt/finishedAt
JobEntity finishedJob =
TestJobTemplateMetaService.newJobEntity(
jobTemplate.name(), JobHandle.Status.SUCCEEDED, METALAKE_NAME);
@@ -147,6 +148,7 @@ public class TestJobMetaService extends TestJDBCBackend {
JobEntity retrievedFinishedJob =
JobMetaService.getInstance()
.getJobByIdentifier(NameIdentifierUtil.ofJob(METALAKE_NAME,
finishedJob.name()));
+ Assertions.assertTrue(retrievedFinishedJob.startedAt() > 0);
Assertions.assertTrue(retrievedFinishedJob.finishedAt() > 0);
}
@@ -180,6 +182,7 @@ public class TestJobMetaService extends TestJDBCBackend {
.withNamespace(job.namespace())
.withAuditInfo(job.auditInfo())
.withJobTemplateName(job.jobTemplateName())
+ .withStartedAt(timestamp)
.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 e83c846f04..1f40cadea2 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,8 @@ public class TestJobTemplateMetaService extends
TestJDBCBackend {
}
static JobEntity newJobEntity(String templateName, JobHandle.Status status,
String metalake) {
+ // Any status other than QUEUED implies the job has at least started.
+ boolean isStarted = status != JobHandle.Status.QUEUED;
boolean isFinished =
status == JobHandle.Status.SUCCEEDED
|| status == JobHandle.Status.FAILED
@@ -289,6 +291,7 @@ public class TestJobTemplateMetaService extends
TestJDBCBackend {
.withJobTemplateName(templateName)
.withStatus(status)
.withAuditInfo(AUDIT_INFO)
+ .withStartedAt(isStarted ? System.currentTimeMillis() : 0L)
.withFinishedAt(isFinished ? System.currentTimeMillis() : 0L)
.build();
}
diff --git a/docs/open-api/jobs.yaml b/docs/open-api/jobs.yaml
index 34a3ab2a01..5237d5e958 100644
--- a/docs/open-api/jobs.yaml
+++ b/docs/open-api/jobs.yaml
@@ -495,6 +495,7 @@ components:
- jobTemplateName
- status
- audit
+ - queuedAt
properties:
jobId:
type: string
@@ -514,6 +515,15 @@ components:
- "canceled"
audit:
$ref: "./openapi.yaml#/components/schemas/Audit"
+ queuedAt:
+ type: string
+ format: date-time
+ description: The time when the job was queued for execution, same as
the job's creation time
+ startedAt:
+ type: string
+ format: date-time
+ nullable: true
+ description: The time when the job started execution, or null if the
job has not started execution yet
finishedAt:
type: string
format: date-time
@@ -926,6 +936,8 @@ components:
"createTime": "2025-08-12T02:14:28.205023Z",
"creator": "anonymous"
},
+ "queuedAt": "2025-08-12T02:14:28.205023Z",
+ "startedAt": "2025-08-12T02:14:31.098231Z",
"finishedAt": "2025-08-12T02:15:03.512847Z"
},
{
@@ -936,6 +948,8 @@ components:
"createTime": "2025-08-12T02:14:28.205023Z",
"creator": "anonymous"
},
+ "queuedAt": "2025-08-12T02:14:28.205023Z",
+ "startedAt": "2025-08-12T02:14:35.442190Z",
"finishedAt": "2025-08-12T02:15:47.891023Z"
}
]
@@ -952,6 +966,8 @@ components:
"createTime": "2025-08-12T02:14:28.205023Z",
"creator": "anonymous"
},
+ "queuedAt": "2025-08-12T02:14:28.205023Z",
+ "startedAt": "2025-08-12T02:14:31.098231Z",
"finishedAt": "2025-08-12T02:15:03.512847Z"
}
}
diff --git a/scripts/h2/schema-2.0.0-h2.sql b/scripts/h2/schema-2.0.0-h2.sql
index 834d234ff1..214d6f3af8 100644
--- a/scripts/h2/schema-2.0.0-h2.sql
+++ b/scripts/h2/schema-2.0.0-h2.sql
@@ -469,6 +469,7 @@ CREATE TABLE IF NOT EXISTS `job_run_meta` (
`metalake_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'metalake id',
`job_execution_id` varchar(256) NOT NULL COMMENT 'job execution id',
`job_run_status` varchar(64) NOT NULL COMMENT 'job run status',
+ `job_started_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'job
started at',
`job_finished_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'job
finished at',
`audit_info` CLOB NOT NULL COMMENT 'job run audit info',
`current_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'job run current
version',
diff --git a/scripts/h2/upgrade-1.3.0-to-2.0.0-h2.sql
b/scripts/h2/upgrade-1.3.0-to-2.0.0-h2.sql
index 489a8ea057..e9b03c2996 100644
--- a/scripts/h2/upgrade-1.3.0-to-2.0.0-h2.sql
+++ b/scripts/h2/upgrade-1.3.0-to-2.0.0-h2.sql
@@ -38,3 +38,5 @@ ALTER TABLE `idp_group_meta` ADD COLUMN `group_comment`
VARCHAR(1024) DEFAULT ''
CREATE UNIQUE INDEX IF NOT EXISTS `uk_ti_mi_mo_tv_del` ON `tag_relation_meta`
(`tag_id`, `metadata_object_id`, `metadata_object_type`, `tag_value`,
`deleted_at`);
CREATE INDEX IF NOT EXISTS `idx_tid_value` ON `tag_relation_meta` (`tag_id`,
`tag_value`);
+
+ALTER TABLE `job_run_meta` ADD COLUMN `job_started_at` BIGINT(20) UNSIGNED NOT
NULL DEFAULT 0 COMMENT 'job started at' AFTER `job_run_status`;
diff --git a/scripts/mysql/schema-2.0.0-mysql.sql
b/scripts/mysql/schema-2.0.0-mysql.sql
index 174af1dab9..c591af5e34 100644
--- a/scripts/mysql/schema-2.0.0-mysql.sql
+++ b/scripts/mysql/schema-2.0.0-mysql.sql
@@ -460,6 +460,7 @@ CREATE TABLE IF NOT EXISTS `job_run_meta` (
`metalake_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'metalake id',
`job_execution_id` varchar(256) NOT NULL COMMENT 'job execution id',
`job_run_status` varchar(64) NOT NULL COMMENT 'job run status',
+ `job_started_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'job
started at',
`job_finished_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'job
finished at',
`audit_info` MEDIUMTEXT NOT NULL COMMENT 'job run audit info',
`current_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'job run current
version',
diff --git a/scripts/mysql/upgrade-1.3.0-to-2.0.0-mysql.sql
b/scripts/mysql/upgrade-1.3.0-to-2.0.0-mysql.sql
index 2d0b8e1417..71b17ed5de 100644
--- a/scripts/mysql/upgrade-1.3.0-to-2.0.0-mysql.sql
+++ b/scripts/mysql/upgrade-1.3.0-to-2.0.0-mysql.sql
@@ -96,3 +96,6 @@ ALTER TABLE `table_version_info`
MODIFY COLUMN `deleted_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT
'table deletion timestamp, 0 means not deleted',
DROP INDEX `uk_table_id_version_deleted_at`,
ADD PRIMARY KEY (`table_id`, `version`, `deleted_at`);
+
+ALTER TABLE `job_run_meta`
+ ADD COLUMN `job_started_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT
'job started at' AFTER `job_run_status`;
diff --git a/scripts/postgresql/schema-2.0.0-postgresql.sql
b/scripts/postgresql/schema-2.0.0-postgresql.sql
index 28b19dd2ec..0f3b39eeb8 100644
--- a/scripts/postgresql/schema-2.0.0-postgresql.sql
+++ b/scripts/postgresql/schema-2.0.0-postgresql.sql
@@ -813,6 +813,7 @@ CREATE TABLE IF NOT EXISTS job_run_meta (
metalake_id BIGINT NOT NULL,
job_execution_id VARCHAR(256) NOT NULL,
job_run_status VARCHAR(64) NOT NULL,
+ job_started_at BIGINT NOT NULL DEFAULT 0,
job_finished_at BIGINT NOT NULL DEFAULT 0,
audit_info TEXT NOT NULL,
current_version INT NOT NULL DEFAULT 1,
@@ -830,6 +831,7 @@ COMMENT ON COLUMN job_run_meta.job_template_id IS 'job
template id';
COMMENT ON COLUMN job_run_meta.metalake_id IS 'metalake id';
COMMENT ON COLUMN job_run_meta.job_execution_id IS 'job execution id';
COMMENT ON COLUMN job_run_meta.job_run_status IS 'job run status';
+COMMENT ON COLUMN job_run_meta.job_started_at IS 'job run started at';
COMMENT ON COLUMN job_run_meta.job_finished_at IS 'job run finished at';
COMMENT ON COLUMN job_run_meta.audit_info IS 'job run audit info';
COMMENT ON COLUMN job_run_meta.current_version IS 'job run current version';
diff --git a/scripts/postgresql/upgrade-1.3.0-to-2.0.0-postgresql.sql
b/scripts/postgresql/upgrade-1.3.0-to-2.0.0-postgresql.sql
index 0244409955..90ce5bc69a 100644
--- a/scripts/postgresql/upgrade-1.3.0-to-2.0.0-postgresql.sql
+++ b/scripts/postgresql/upgrade-1.3.0-to-2.0.0-postgresql.sql
@@ -45,3 +45,6 @@ ALTER TABLE tag_relation_meta DROP CONSTRAINT IF EXISTS
tag_relation_meta_tag_id
CREATE UNIQUE INDEX IF NOT EXISTS uk_ti_mi_mo_tv_del ON tag_relation_meta
(tag_id, metadata_object_id, metadata_object_type, tag_value, deleted_at);
CREATE INDEX IF NOT EXISTS tag_relation_meta_idx_tag_id_value ON
tag_relation_meta (tag_id, tag_value);
+
+ALTER TABLE job_run_meta ADD COLUMN IF NOT EXISTS job_started_at BIGINT NOT
NULL DEFAULT 0;
+COMMENT ON COLUMN job_run_meta.job_started_at IS 'job run started at';
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 1a033abda6..74d1e0e5fd 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
@@ -477,6 +477,8 @@ public class JobOperations {
jobEntity.jobTemplateName(),
jobEntity.status(),
DTOConverters.toDTO(jobEntity.auditInfo()),
+ jobEntity.auditInfo().createTime(),
+ jobEntity.startedAtAsInstant(),
jobEntity.finishedAtAsInstant());
}
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 1e2fd543d8..cdd53e6cd8 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
@@ -641,7 +641,8 @@ public class TestJobOperations extends JerseyTest {
public void testListJobs() {
String templateName = "shell_template_1";
JobEntity job1 = newJobEntity(templateName, JobHandle.Status.QUEUED);
- JobEntity job2 = newJobEntity(templateName, JobHandle.Status.STARTED);
+ JobEntity job2 =
+ newJobEntity(templateName, JobHandle.Status.STARTED,
Instant.now().toEpochMilli(), 0L);
JobEntity job3 =
newJobEntity("spark_template_1", JobHandle.Status.SUCCEEDED,
Instant.now().toEpochMilli());
@@ -672,6 +673,17 @@ public class TestJobOperations extends JerseyTest {
Assertions.assertEquals(
Instant.ofEpochMilli(job3.finishedAt()),
jobListResponse.getJobs().get(2).finishedAt());
+ // queuedAt is always present, regardless of status.
+ Assertions.assertNotNull(jobListResponse.getJobs().get(0).queuedAt());
+ Assertions.assertNotNull(jobListResponse.getJobs().get(1).queuedAt());
+ Assertions.assertNotNull(jobListResponse.getJobs().get(2).queuedAt());
+
+ // A not-yet-started job round-trips startedAt as null over the wire.
+ Assertions.assertNull(jobListResponse.getJobs().get(0).startedAt());
+ // A started job round-trips its startedAt as an Instant over the wire.
+ Assertions.assertEquals(
+ Instant.ofEpochMilli(job2.startedAt()),
jobListResponse.getJobs().get(1).startedAt());
+
// Test list jobs by template name
when(jobOperationDispatcher.listJobs(metalake, Optional.of(templateName)))
.thenReturn(Lists.newArrayList(job1, job2));
@@ -802,8 +814,10 @@ public class TestJobOperations extends JerseyTest {
@Test
public void testCancelJob() {
+ long startedAt = Instant.now().toEpochMilli() - 1000;
+ long finishedAt = Instant.now().toEpochMilli();
JobEntity job =
- newJobEntity("shell_template_1", JobHandle.Status.CANCELLED,
Instant.now().toEpochMilli());
+ newJobEntity("shell_template_1", JobHandle.Status.CANCELLED,
startedAt, finishedAt);
when(jobOperationDispatcher.cancelJob(metalake,
job.name())).thenReturn(job);
@@ -820,6 +834,10 @@ public class TestJobOperations extends JerseyTest {
JobResponse jobResp = resp.readEntity(JobResponse.class);
Assertions.assertEquals(0, jobResp.getCode());
Assertions.assertEquals(JobOperations.toDTO(job), jobResp.getJob());
+ // queuedAt is always present, regardless of status.
+ Assertions.assertNotNull(jobResp.getJob().queuedAt());
+ // A cancelled job that had started round-trips its startedAt as an
Instant over the wire.
+ Assertions.assertEquals(Instant.ofEpochMilli(job.startedAt()),
jobResp.getJob().startedAt());
// A finished (cancelled) job round-trips its finishedAt as an Instant
over the wire.
Assertions.assertEquals(Instant.ofEpochMilli(job.finishedAt()),
jobResp.getJob().finishedAt());
@@ -857,6 +875,30 @@ public class TestJobOperations extends JerseyTest {
Assertions.assertEquals(Instant.ofEpochMilli(epochMilli),
finishedJobDTO.finishedAt());
}
+ @Test
+ public void testToDTOStartedAt() {
+ // Sentinel value (<= 0) used by the storage layer means "not started".
+ JobEntity sentinelJob = newJobEntity("shell_template_1",
JobHandle.Status.QUEUED, 0L, 0L);
+ JobDTO sentinelJobDTO = JobOperations.toDTO(sentinelJob);
+ Assertions.assertNull(sentinelJobDTO.startedAt());
+
+ // Started, startedAt is converted from epoch millis to an Instant.
+ long epochMilli = Instant.now().toEpochMilli();
+ JobEntity startedJob =
+ newJobEntity("shell_template_1", JobHandle.Status.STARTED, epochMilli,
0L);
+ JobDTO startedJobDTO = JobOperations.toDTO(startedJob);
+ Assertions.assertEquals(Instant.ofEpochMilli(epochMilli),
startedJobDTO.startedAt());
+ }
+
+ @Test
+ public void testToDTOQueuedAt() {
+ // queuedAt is always present - it's the job's creation time, not a
sentinel-backed field.
+ JobEntity job = newJobEntity("shell_template_1", JobHandle.Status.QUEUED);
+ JobDTO jobDTO = JobOperations.toDTO(job);
+ Assertions.assertEquals(job.auditInfo().createTime(), jobDTO.queuedAt());
+ Assertions.assertNotNull(jobDTO.queuedAt());
+ }
+
private String jobTemplatePath() {
return "/metalakes/" + metalake + "/jobs/templates";
}
@@ -903,10 +945,15 @@ public class TestJobOperations extends JerseyTest {
}
private JobEntity newJobEntity(String templateName, JobHandle.Status status)
{
- return newJobEntity(templateName, status, 0L);
+ return newJobEntity(templateName, status, 0L, 0L);
}
private JobEntity newJobEntity(String templateName, JobHandle.Status status,
Long finishedAt) {
+ return newJobEntity(templateName, status, 0L, finishedAt);
+ }
+
+ private JobEntity newJobEntity(
+ String templateName, JobHandle.Status status, Long startedAt, Long
finishedAt) {
Random rand = new Random();
return JobEntity.builder()
.withId(rand.nextLong())
@@ -916,6 +963,7 @@ public class TestJobOperations extends JerseyTest {
.withStatus(status)
.withAuditInfo(
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .withStartedAt(startedAt)
.withFinishedAt(finishedAt)
.build();
}
