This is an automated email from the ASF dual-hosted git repository.
mchades 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 e713cadb2b [#12676] feat(job): Persist and expose the resolved runtime
job template (#12715)
e713cadb2b is described below
commit e713cadb2bcce36fd95998ff2fb229046f67690d
Author: Jerry Shao <[email protected]>
AuthorDate: Tue Sep 1 23:17:55 2026 +0800
[#12676] feat(job): Persist and expose the resolved runtime job template
(#12715)
### What changes were proposed in this pull request?
When a job is run, the job template's placeholders are resolved and any
referenced files are downloaded before submission to the job executor.
This PR captures that *resolved* template (not just the original one)
and persists it alongside the job entity as a new `runtimeJobTemplate`
field, exposed end-to-end:
- Storage: schema migration + `JobPO` column + MySQL/PostgreSQL/H2
mapper providers
- Core: `JobManager` serializes the resolved template before submission
(so a serialization failure never leaves an orphaned job running on the
executor without a tracked entity) and `JobInfo` deserializes it for
listener events
- REST API: new `runtimeJobTemplate` field on the job response
(`docs/open-api/jobs.yaml`)
- Common: `JobDTO`/`DTOConverters` gain a shared
`fromRuntimeJobTemplateJson` helper used by both core and server
- Clients: `JobHandle.runtimeJobTemplate()` on Java and Python clients
Also folds in an unrelated black-formatting fix for 3 Python files
(`fileset_change.py`, `test_model_catalog.py`,
`test_relational_catalog.py`) as a drive-by style fix.
### Why are the changes needed?
Job templates can contain placeholders and template files that get
resolved/downloaded at run time. Without this change, callers can only
see the original template definition, not what was actually submitted
for execution — making it hard to debug or audit a specific job run
after the fact.
Fix: #12676
### Does this PR introduce _any_ user-facing change?
Yes:
- New `runtimeJobTemplate` field in the Job REST API response (nullable;
omitted for jobs run before this field existed)
- New `JobHandle.runtimeJobTemplate()` accessor on the Java and Python
clients
### How was this patch tested?
Added/updated unit tests across storage (PO/mapper/service), core
(`JobManager`, `JobInfo`), common (`JobDTO`, `DTOConverters`), server
(`JobOperations`), and both Java and Python clients, including full
serialization round-trips. Also validated `docs/open-api/jobs.yaml` via
`./gradlew :docs:build`.
https://claude.ai/code/session_01NLuzEFs3CmzP5duYonCUue
---------
Co-authored-by: Claude Sonnet 5 <[email protected]>
---
.../java/org/apache/gravitino/job/JobHandle.java | 14 +++
.../apache/gravitino/client/GenericJobHandle.java | 9 ++
.../apache/gravitino/client/TestSupportsJobs.java | 39 +++++-
.../gravitino/client/integration/test/JobIT.java | 36 ++++++
.../gravitino/api/file/fileset_change.py | 3 +-
.../client-python/gravitino/api/job/job_handle.py | 9 ++
.../gravitino/client/generic_job_handle.py | 7 ++
clients/client-python/gravitino/dto/job/job_dto.py | 31 ++++-
.../gravitino/dto/job/job_template_dto.py | 21 +++-
.../tests/integration/test_supports_jobs.py | 30 +++++
.../tests/unittests/dto/job/test_job_dto_serde.py | 44 +++++++
.../tests/unittests/test_supports_jobs.py | 45 +++++++
.../java/org/apache/gravitino/dto/job/JobDTO.java | 12 +-
.../apache/gravitino/dto/util/DTOConverters.java | 83 +++++++++++++
.../org/apache/gravitino/dto/job/TestJobDTO.java | 64 +++++++++-
.../gravitino/dto/util/TestDTOConverters.java | 122 +++++++++++++++++++
.../java/org/apache/gravitino/job/JobManager.java | 22 ++++
.../gravitino/listener/api/info/JobInfo.java | 51 +++++++-
.../java/org/apache/gravitino/meta/JobEntity.java | 39 +++++-
.../provider/base/JobMetaBaseSQLProvider.java | 20 +++-
.../postgresql/JobMetaPostgreSQLProvider.java | 8 +-
.../gravitino/storage/relational/po/JobPO.java | 18 ++-
.../org/apache/gravitino/job/TestJobManager.java | 128 ++++++++++++++++++++
.../gravitino/listener/api/info/TestJobInfo.java | 90 ++++++++++++++
.../org/apache/gravitino/meta/TestJobEntity.java | 86 ++++++++++++++
.../provider/base/TestJobMetaBaseSQLProvider.java | 84 +++++++++++++
.../postgresql/TestJobMetaPostgreSQLProvider.java | 43 +++++++
.../gravitino/storage/relational/po/TestJobPO.java | 68 ++++++++++-
.../relational/service/TestJobMetaService.java | 131 +++++++++++++++++++++
docs/open-api/jobs.yaml | 42 ++++++-
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 | 24 +++-
.../server/web/rest/TestJobOperations.java | 126 ++++++++++++++++++++
38 files changed, 1528 insertions(+), 33 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 b365ada40a..7da6129df7 100644
--- a/api/src/main/java/org/apache/gravitino/job/JobHandle.java
+++ b/api/src/main/java/org/apache/gravitino/job/JobHandle.java
@@ -104,4 +104,18 @@ public interface JobHandle {
throw new UnsupportedOperationException(
"finishedAt() is not implemented by " + getClass().getName() + ";
override this method");
}
+
+ /**
+ * Get the resolved job template that was actually submitted for execution,
with placeholders
+ * replaced and referenced files downloaded.
+ *
+ * @return the runtime job template, or null for jobs run before this field
was introduced
+ */
+ @Nullable
+ default JobTemplate runtimeJobTemplate() {
+ throw new UnsupportedOperationException(
+ "runtimeJobTemplate() is not implemented by "
+ + getClass().getName()
+ + "; override this method");
+ }
}
diff --git
a/clients/client-java/src/main/java/org/apache/gravitino/client/GenericJobHandle.java
b/clients/client-java/src/main/java/org/apache/gravitino/client/GenericJobHandle.java
index 2c2cc829c7..beb370b589 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
@@ -20,7 +20,9 @@ package org.apache.gravitino.client;
import java.time.Instant;
import org.apache.gravitino.dto.job.JobDTO;
+import org.apache.gravitino.dto.util.DTOConverters;
import org.apache.gravitino.job.JobHandle;
+import org.apache.gravitino.job.JobTemplate;
/** Represents a generic job handle. */
public class GenericJobHandle implements JobHandle {
@@ -60,4 +62,11 @@ public class GenericJobHandle implements JobHandle {
public Instant finishedAt() {
return jobDTO.finishedAt();
}
+
+ @Override
+ public JobTemplate runtimeJobTemplate() {
+ return jobDTO.runtimeJobTemplate() == null
+ ? null
+ : DTOConverters.fromDTO(jobDTO.runtimeJobTemplate());
+ }
}
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 f811728cb0..ff6db13a09 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
@@ -43,6 +43,7 @@ import org.apache.gravitino.dto.responses.JobListResponse;
import org.apache.gravitino.dto.responses.JobResponse;
import org.apache.gravitino.dto.responses.JobTemplateListResponse;
import org.apache.gravitino.dto.responses.JobTemplateResponse;
+import org.apache.gravitino.dto.util.DTOConverters;
import org.apache.gravitino.exceptions.InUseException;
import org.apache.gravitino.exceptions.JobTemplateAlreadyExistsException;
import org.apache.gravitino.exceptions.MetalakeNotInUseException;
@@ -315,6 +316,27 @@ public class TestSupportsJobs extends TestBase {
Assertions.assertThrows(NoSuchJobException.class, () ->
metalake.getJob(jobId));
}
+ @Test
+ public void testGetJobWithRuntimeJobTemplate() throws
JsonProcessingException {
+ String jobId = "job-1";
+ String jobTemplateName = "shell-job-template";
+ JobTemplateDTO runtimeJobTemplateDTO =
newShellJobTemplateDTO(jobTemplateName);
+ JobDTO expectedJob =
+ newJobDTO(jobId, jobTemplateName, Instant.now(), Instant.now(),
runtimeJobTemplateDTO);
+ JobResponse resp = new JobResponse(expectedJob);
+
+ buildMockResource(Method.GET, jobRunsPath() + "/" + jobId, null, resp,
HttpStatus.SC_OK);
+
+ JobHandle actualHandle = metalake.getJob(jobId);
+ compare(expectedJob, actualHandle);
+
+ // The handle must expose the resolved runtime template as an api-level
JobTemplate (not the
+ // wire DTO), converted via DTOConverters.fromDTO.
+ JobTemplate runtimeJobTemplate = actualHandle.runtimeJobTemplate();
+ Assertions.assertNotNull(runtimeJobTemplate);
+ Assertions.assertEquals(DTOConverters.fromDTO(runtimeJobTemplateDTO),
runtimeJobTemplate);
+ }
+
@Test
public void testRunJob() throws JsonProcessingException {
String jobTemplateName = "shell-job-template";
@@ -364,6 +386,11 @@ public class TestSupportsJobs extends TestBase {
Assertions.assertEquals(expected.queuedAt(), actual.queuedAt());
Assertions.assertEquals(expected.startedAt(), actual.startedAt());
Assertions.assertEquals(expected.finishedAt(), actual.finishedAt());
+ JobTemplate expectedRuntimeJobTemplate =
+ expected.runtimeJobTemplate() == null
+ ? null
+ : DTOConverters.fromDTO(expected.runtimeJobTemplate());
+ Assertions.assertEquals(expectedRuntimeJobTemplate,
actual.runtimeJobTemplate());
}
private String jobTemplatesPath() {
@@ -414,6 +441,15 @@ public class TestSupportsJobs extends TestBase {
private JobDTO newJobDTO(
String jobId, String templateName, Instant startedAt, Instant
finishedAt) {
+ return newJobDTO(jobId, templateName, startedAt, finishedAt, null);
+ }
+
+ private JobDTO newJobDTO(
+ String jobId,
+ String templateName,
+ Instant startedAt,
+ Instant finishedAt,
+ JobTemplateDTO runtimeJobTemplate) {
Instant now = Instant.now();
return new JobDTO(
jobId,
@@ -422,6 +458,7 @@ public class TestSupportsJobs extends TestBase {
AuditDTO.builder().withCreator("test").withCreateTime(now).build(),
now,
startedAt,
- finishedAt);
+ finishedAt,
+ runtimeJobTemplate);
}
}
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 5813751113..61bb843233 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
@@ -402,6 +402,42 @@ public class JobIT extends BaseIT {
Assertions.assertThrows(NoSuchJobException.class, () ->
metalake.getJob("non_existent_job_id"));
}
+ @Test
+ public void testRunJobPopulatesRuntimeJobTemplate() {
+ JobTemplate template =
builder.withName("test_run_runtime_template").build();
+ Assertions.assertDoesNotThrow(() ->
metalake.registerJobTemplate(template));
+
+ JobHandle jobHandle =
+ metalake.runJob(
+ template.name(),
+ ImmutableMap.of("arg1", "value1", "arg2", "success", "env_var",
"value2"));
+
+ // The resolved runtime template is set at submission time, before the job
even starts
+ // executing, and must carry the actual substituted values rather than the
original
+ // template's raw {{placeholder}} strings.
+ JobTemplate runtimeJobTemplate = jobHandle.runtimeJobTemplate();
+ Assertions.assertNotNull(runtimeJobTemplate);
+ Assertions.assertEquals(template.name(), runtimeJobTemplate.name());
+ Assertions.assertEquals(template.comment(), runtimeJobTemplate.comment());
+ Assertions.assertEquals(
+ Lists.newArrayList("value1", "success"),
runtimeJobTemplate.arguments());
+ Assertions.assertEquals(
+ ImmutableMap.of("ENV_VAR", "value2"),
runtimeJobTemplate.environments());
+
+ Awaitility.await()
+ .atMost(3, TimeUnit.MINUTES)
+ .until(
+ () -> {
+ JobHandle updatedJob = metalake.getJob(jobHandle.jobId());
+ return updatedJob.jobStatus() == JobHandle.Status.SUCCEEDED;
+ });
+
+ // The runtime job template is fixed at submission time, so it must be
unchanged once the job
+ // reaches a terminal status and its entity has gone through the
status-poll update path.
+ JobHandle retrievedJob = metalake.getJob(jobHandle.jobId());
+ Assertions.assertEquals(runtimeJobTemplate,
retrievedJob.runtimeJobTemplate());
+ }
+
@Test
public void testRunAndCancelJob() {
JobTemplate template = builder.withName("test_run_cancel").build();
diff --git a/clients/client-python/gravitino/api/file/fileset_change.py
b/clients/client-python/gravitino/api/file/fileset_change.py
index 8ca6a6e4f1..7bc2f7bc6d 100644
--- a/clients/client-python/gravitino/api/file/fileset_change.py
+++ b/clients/client-python/gravitino/api/file/fileset_change.py
@@ -347,8 +347,7 @@ class FilesetChange(ABC):
if not isinstance(other, FilesetChange.SetSecretBinding):
return False
return (
- self._property == other.property()
- and self._binding == other.binding()
+ self._property == other.property() and self._binding ==
other.binding()
)
def __hash__(self):
diff --git a/clients/client-python/gravitino/api/job/job_handle.py
b/clients/client-python/gravitino/api/job/job_handle.py
index 1caebc28c5..4f9d749165 100644
--- a/clients/client-python/gravitino/api/job/job_handle.py
+++ b/clients/client-python/gravitino/api/job/job_handle.py
@@ -20,6 +20,8 @@ from datetime import datetime
from enum import Enum
from typing import Optional
+from gravitino.api.job.job_template import JobTemplate
+
class JobHandle(ABC):
class Status(Enum):
@@ -70,3 +72,10 @@ class JobHandle(ABC):
execution yet.
"""
raise NotImplementedError("finished_at is not implemented")
+
+ def runtime_job_template(self) -> Optional[JobTemplate]:
+ """Returns the resolved job template that was actually submitted for
execution, with
+ placeholders replaced and referenced files downloaded, or ``None`` for
jobs run before
+ this field was introduced.
+ """
+ raise NotImplementedError("runtime_job_template is not implemented")
diff --git a/clients/client-python/gravitino/client/generic_job_handle.py
b/clients/client-python/gravitino/client/generic_job_handle.py
index e5c2a10a82..ce0e2bbca8 100644
--- a/clients/client-python/gravitino/client/generic_job_handle.py
+++ b/clients/client-python/gravitino/client/generic_job_handle.py
@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
from gravitino.api.job.job_handle import JobHandle
+from gravitino.client.dto_converters import DTOConverters
from gravitino.dto.job.job_dto import JobDTO
@@ -41,3 +42,9 @@ class GenericJobHandle(JobHandle):
def finished_at(self):
return self._job_dto.finished_at()
+
+ def runtime_job_template(self):
+ runtime_job_template_dto = self._job_dto.runtime_job_template()
+ if runtime_job_template_dto is None:
+ return None
+ return DTOConverters.from_job_template_dto(runtime_job_template_dto)
diff --git a/clients/client-python/gravitino/dto/job/job_dto.py
b/clients/client-python/gravitino/dto/job/job_dto.py
index 5b2e6bd187..dd855e923e 100644
--- a/clients/client-python/gravitino/dto/job/job_dto.py
+++ b/clients/client-python/gravitino/dto/job/job_dto.py
@@ -17,7 +17,7 @@
from dataclasses import dataclass, field
from datetime import datetime
-from typing import Optional
+from typing import Dict, Optional
from dataclasses_json import config, DataClassJsonMixin
@@ -27,10 +27,23 @@ from gravitino.dto.audit_dto import (
_deserialize_datetime,
_serialize_datetime,
)
+from gravitino.dto.job.job_template_dto import JobTemplateDTO
+
+
+def _serialize_runtime_job_template(
+ value: Optional[JobTemplateDTO],
+) -> Optional[Dict]:
+ return None if value is None else value.to_dict()
+
+
+def _deserialize_runtime_job_template(
+ value: Optional[Dict],
+) -> Optional[JobTemplateDTO]:
+ return None if value is None else JobTemplateDTO.from_dict_by_type(value)
@dataclass
-class JobDTO(DataClassJsonMixin):
+class JobDTO(DataClassJsonMixin): # pylint:
disable=too-many-instance-attributes
"""Data transfer object representing a Job."""
_job_id: str = field(metadata=config(field_name="jobId"))
@@ -67,6 +80,14 @@ class JobDTO(DataClassJsonMixin):
decoder=_deserialize_datetime,
),
)
+ _runtime_job_template: Optional[JobTemplateDTO] = field(
+ default=None,
+ metadata=config(
+ field_name="runtimeJobTemplate",
+ encoder=_serialize_runtime_job_template,
+ decoder=_deserialize_runtime_job_template,
+ ),
+ )
def __post_init__(self) -> None:
self._queued_at = _deserialize_datetime(self._queued_at)
@@ -105,6 +126,12 @@ class JobDTO(DataClassJsonMixin):
"""
return self._finished_at
+ def runtime_job_template(self) -> Optional[JobTemplateDTO]:
+ """Returns the resolved job template that was actually submitted for
execution, or
+ ``None`` for jobs run before this field was introduced.
+ """
+ return self._runtime_job_template
+
def validate(self) -> None:
"""Validates the JobDTO, ensuring required fields are present and
non-empty."""
if self._job_id is None or not self._job_id.strip():
diff --git a/clients/client-python/gravitino/dto/job/job_template_dto.py
b/clients/client-python/gravitino/dto/job/job_template_dto.py
index 6ed6b1a6cc..79468589ae 100644
--- a/clients/client-python/gravitino/dto/job/job_template_dto.py
+++ b/clients/client-python/gravitino/dto/job/job_template_dto.py
@@ -97,17 +97,30 @@ class JobTemplateDTO(DataClassJsonMixin, ABC):
raise ValueError('"executable" is required and cannot be empty')
@classmethod
- def from_json(
- cls, s: str, infer_missing: bool = False, **kwargs
+ def from_dict_by_type(
+ cls, data: Dict, infer_missing: bool = False
) -> "JobTemplateDTO":
- """Creates a JobTemplateDTO from a JSON string."""
- data = json.loads(s)
+ """Creates a JobTemplateDTO from a dict, dispatching to the concrete
subclass based
+ on the "jobType" field.
+ """
job_type = JobType.job_type_deserialize(data.get("jobType"))
subclass = JOB_TYPE_TEMPLATE_MAPPING.get(job_type)
if not subclass:
raise ValueError(f"Unsupported job type: {job_type}")
return subclass.from_dict(data, infer_missing=infer_missing)
+ @classmethod
+ def from_json(
+ cls, s: str, infer_missing: bool = False, **kwargs
+ ) -> "JobTemplateDTO":
+ """Creates a JobTemplateDTO from a JSON string. Any extra keyword
arguments are passed
+ through to json.loads (e.g. parse_float, parse_int), matching the base
+ DataClassJsonMixin.from_json contract rather than silently discarding
them.
+ """
+ return cls.from_dict_by_type(
+ json.loads(s, **kwargs), infer_missing=infer_missing
+ )
+
JOB_TYPE_TEMPLATE_MAPPING: Dict[JobType, Type["JobTemplateDTO"]] = {}
diff --git a/clients/client-python/tests/integration/test_supports_jobs.py
b/clients/client-python/tests/integration/test_supports_jobs.py
index f558bebdb2..f8e3321861 100644
--- a/clients/client-python/tests/integration/test_supports_jobs.py
+++ b/clients/client-python/tests/integration/test_supports_jobs.py
@@ -372,6 +372,36 @@ class TestSupportsJobs(IntegrationTestEnv):
with self.assertRaises(NoSuchJobException):
self._metalake.get_job("non_existent_job_id")
+ def test_run_job_populates_runtime_job_template(self):
+ template = self.builder.with_name("test_run_runtime_template").build()
+ self._metalake.register_job_template(template)
+
+ job_handle = self._metalake.run_job(
+ template.name, {"arg1": "value1", "arg2": "success", "env_var":
"value2"}
+ )
+
+ # The resolved runtime template is set at submission time, before the
job even starts
+ # executing, and must carry the actual substituted values rather than
the original
+ # template's raw {{placeholder}} strings.
+ runtime_job_template = job_handle.runtime_job_template()
+ self.assertIsNotNone(runtime_job_template)
+ self.assertEqual(template.name, runtime_job_template.name)
+ self.assertEqual(template.comment, runtime_job_template.comment)
+ self.assertEqual(["value1", "success"], runtime_job_template.arguments)
+ self.assertEqual({"ENV_VAR": "value2"},
runtime_job_template.environments)
+
+ self._wait_until(
+ lambda: self._metalake.get_job(job_handle.job_id()).job_status()
+ == JobHandle.Status.SUCCEEDED,
+ timeout=180,
+ )
+
+ # The runtime job template is fixed at submission time, so it must be
unchanged once the
+ # job reaches a terminal status and its entity has gone through the
status-poll update
+ # path.
+ retrieved_job = self._metalake.get_job(job_handle.job_id())
+ self.assertEqual(runtime_job_template,
retrieved_job.runtime_job_template())
+
def test_run_and_cancel_job(self):
template = self.builder.with_name("test_run_cancel").build()
self._metalake.register_job_template(template)
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 ab39812b9e..4e4eafb1db 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
@@ -18,8 +18,10 @@ import unittest
from datetime import datetime, timezone
from gravitino.api.job.job_handle import JobHandle
+from gravitino.api.job.job_template import JobType
from gravitino.dto.audit_dto import AuditDTO
from gravitino.dto.job.job_dto import JobDTO
+from gravitino.dto.job.shell_job_template_dto import ShellJobTemplateDTO
class TestJobDTOSerDe(unittest.TestCase):
@@ -107,3 +109,45 @@ class TestJobDTOSerDe(unittest.TestCase):
)
self.assertIsNone(job_dto.started_at())
self.assertIsNone(job_dto.finished_at())
+
+ def test_ser_de_with_runtime_job_template(self):
+ audit = AuditDTO(_creator="test",
_create_time=datetime.now(timezone.utc))
+ runtime_job_template = ShellJobTemplateDTO(
+ _job_type=JobType.SHELL,
+ _name="resolved-template",
+ _executable="/bin/echo",
+ _comment="A resolved job template",
+ _arguments=["resolved-arg"],
+ _environments={},
+ _custom_fields={},
+ _scripts=["/path/to/script.sh"],
+ _audit=audit,
+ )
+ job_dto = JobDTO(
+ _job_id="job-1001",
+ _job_template_name="test_template",
+ _status=JobHandle.Status.SUCCEEDED,
+ _audit=audit,
+ _runtime_job_template=runtime_job_template,
+ )
+
+ json_str = job_dto.to_json()
+ self.assertIn("runtimeJobTemplate", json_str)
+
+ deser_job_dto = JobDTO.from_json(json_str)
+ self.assertEqual(job_dto, deser_job_dto)
+ self.assertEqual(runtime_job_template,
deser_job_dto.runtime_job_template())
+ self.assertIsInstance(deser_job_dto.runtime_job_template(),
ShellJobTemplateDTO)
+
+ def test_ser_de_with_none_runtime_job_template(self):
+ job_dto = JobDTO(
+ _job_id="job-1002",
+ _job_template_name="test_template",
+ _status=JobHandle.Status.QUEUED,
+ _audit=AuditDTO(_creator="test",
_create_time=datetime.now(timezone.utc)),
+ )
+
+ json_str = job_dto.to_json()
+ deser_job_dto = JobDTO.from_json(json_str)
+ self.assertEqual(job_dto, deser_job_dto)
+ self.assertIsNone(deser_job_dto.runtime_job_template())
diff --git a/clients/client-python/tests/unittests/test_supports_jobs.py
b/clients/client-python/tests/unittests/test_supports_jobs.py
index 0ca6367490..1167491a28 100644
--- a/clients/client-python/tests/unittests/test_supports_jobs.py
+++ b/clients/client-python/tests/unittests/test_supports_jobs.py
@@ -26,8 +26,10 @@ from gravitino.api.job.job_template import JobType
from gravitino.api.job.job_template_change import JobTemplateChange,
ShellTemplateUpdate
from gravitino.api.job.shell_job_template import ShellJobTemplate
from gravitino.api.job.spark_job_template import SparkJobTemplate
+from gravitino.client.dto_converters import DTOConverters
from gravitino.dto.audit_dto import AuditDTO
from gravitino.dto.job.job_dto import JobDTO
+from gravitino.dto.job.job_template_dto import JobTemplateDTO
from gravitino.dto.job.shell_job_template_dto import ShellJobTemplateDTO
from gravitino.dto.job.spark_job_template_dto import SparkJobTemplateDTO
from gravitino.dto.responses.base_response import BaseResponse
@@ -228,6 +230,39 @@ class TestSupportsJobs(unittest.TestCase):
with self.assertRaises(ValueError):
gravitino_client.get_job(None)
+ def test_get_job_with_runtime_job_template(self, *mock_methods):
+ gravitino_client = GravitinoClient(
+ uri="http://localhost:8090";,
+ metalake_name=self._metalake_name,
+ )
+
+ job_template_name = "test_shell_job"
+ runtime_job_template = self._new_shell_job_template_dto(
+ self._new_shell_job_template()
+ )
+ job_dto = self._new_job_dto(
+ job_template_name,
+ finished_at=datetime.now(timezone.utc),
+ started_at=datetime.now(timezone.utc),
+ runtime_job_template=runtime_job_template,
+ )
+ resp = JobResponse(_job=job_dto, _code=0)
+ mock_resp = self._mock_http_response(resp.to_json())
+
+ with patch(
+ "gravitino.utils.http_client.HTTPClient.get",
return_value=mock_resp
+ ):
+ job_handle = gravitino_client.get_job(job_dto.job_id())
+ self._compare_job_handle(job_handle, job_dto)
+
+ # The handle must expose the resolved runtime template as an
api-level JobTemplate
+ # (not the wire DTO), converted via
DTOConverters.from_job_template_dto.
+ self.assertIsNotNone(job_handle.runtime_job_template())
+ self.assertEqual(
+ DTOConverters.from_job_template_dto(runtime_job_template),
+ job_handle.runtime_job_template(),
+ )
+
def test_cancel_job(self, *mock_methods):
gravitino_client = GravitinoClient(
uri="http://localhost:8090";,
@@ -321,6 +356,7 @@ class TestSupportsJobs(unittest.TestCase):
job_template_name: str,
finished_at: Optional[datetime] = None,
started_at: Optional[datetime] = None,
+ runtime_job_template: Optional[JobTemplateDTO] = None,
) -> JobDTO:
return JobDTO(
_job_id="job-123",
@@ -330,6 +366,7 @@ class TestSupportsJobs(unittest.TestCase):
_queued_at=datetime(2023, 10, 1, tzinfo=timezone.utc),
_started_at=started_at,
_finished_at=finished_at,
+ _runtime_job_template=runtime_job_template,
)
def _compare_job_handle(self, job_handle: JobHandle, job_dto: JobDTO):
@@ -339,3 +376,11 @@ class TestSupportsJobs(unittest.TestCase):
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())
+ expected_runtime_job_template = (
+ None
+ if job_dto.runtime_job_template() is None
+ else
DTOConverters.from_job_template_dto(job_dto.runtime_job_template())
+ )
+ self.assertEqual(
+ job_handle.runtime_job_template(), expected_runtime_job_template
+ )
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 3c810a0866..54b4abaf11 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
@@ -68,9 +68,12 @@ public class JobDTO {
@JsonProperty("finishedAt")
private final Instant finishedAt;
+ @JsonProperty("runtimeJobTemplate")
+ private final JobTemplateDTO runtimeJobTemplate;
+
/** Default constructor for Jackson deserialization. */
private JobDTO() {
- this(null, null, null, null, null, null, null);
+ this(null, null, null, null, null, null, null, null);
}
/**
@@ -85,6 +88,9 @@ public class JobDTO {
* execution yet.
* @param finishedAt The time when the job finished execution, or null if
the job has not finished
* execution yet.
+ * @param runtimeJobTemplate The resolved job template that was actually
submitted for execution,
+ * with placeholders replaced and referenced files downloaded, or null
for jobs run before
+ * this field was introduced.
*/
public JobDTO(
String jobId,
@@ -93,7 +99,8 @@ public class JobDTO {
AuditDTO audit,
Instant queuedAt,
Instant startedAt,
- Instant finishedAt) {
+ Instant finishedAt,
+ JobTemplateDTO runtimeJobTemplate) {
this.jobId = jobId;
this.jobTemplateName = jobTemplateName;
this.status = status;
@@ -101,6 +108,7 @@ public class JobDTO {
this.queuedAt = queuedAt;
this.startedAt = startedAt;
this.finishedAt = finishedAt;
+ this.runtimeJobTemplate = runtimeJobTemplate;
}
/**
diff --git
a/common/src/main/java/org/apache/gravitino/dto/util/DTOConverters.java
b/common/src/main/java/org/apache/gravitino/dto/util/DTOConverters.java
index b30ba533af..7ea792a255 100644
--- a/common/src/main/java/org/apache/gravitino/dto/util/DTOConverters.java
+++ b/common/src/main/java/org/apache/gravitino/dto/util/DTOConverters.java
@@ -20,6 +20,7 @@ package org.apache.gravitino.dto.util;
import static
org.apache.gravitino.rel.expressions.transforms.Transforms.NAME_OF_IDENTITY;
+import com.fasterxml.jackson.core.JsonProcessingException;
import java.util.Arrays;
import java.util.Map;
import java.util.Optional;
@@ -97,6 +98,7 @@ import org.apache.gravitino.function.Function;
import org.apache.gravitino.job.JobTemplate;
import org.apache.gravitino.job.ShellJobTemplate;
import org.apache.gravitino.job.SparkJobTemplate;
+import org.apache.gravitino.json.JsonUtils;
import org.apache.gravitino.messaging.Topic;
import org.apache.gravitino.model.Model;
import org.apache.gravitino.model.ModelVersion;
@@ -1413,6 +1415,87 @@ public class DTOConverters {
}
}
+ /**
+ * Converts a JobTemplate to a JobTemplateDTO.
+ *
+ * @param jobTemplate The job template to be converted.
+ * @param audit The audit information to attach to the DTO. A bare {@link
JobTemplate} carries no
+ * audit info of its own, so the caller supplies it (e.g. the
originating template entity's
+ * audit info, when serializing a resolved runtime template).
+ * @return The job template DTO.
+ */
+ public static JobTemplateDTO toDTO(JobTemplate jobTemplate, AuditDTO audit) {
+ switch (jobTemplate.jobType()) {
+ case SHELL:
+ ShellJobTemplate shellJobTemplate = (ShellJobTemplate) jobTemplate;
+ return ShellJobTemplateDTO.builder()
+ .withName(shellJobTemplate.name())
+ .withComment(shellJobTemplate.comment())
+ .withJobType(shellJobTemplate.jobType())
+ .withExecutable(shellJobTemplate.executable())
+ .withArguments(shellJobTemplate.arguments())
+ .withEnvironments(shellJobTemplate.environments())
+ .withCustomFields(shellJobTemplate.customFields())
+ .withScripts(shellJobTemplate.scripts())
+ .withAudit(audit)
+ .build();
+
+ case SPARK:
+ SparkJobTemplate sparkJobTemplate = (SparkJobTemplate) jobTemplate;
+ return SparkJobTemplateDTO.builder()
+ .withName(sparkJobTemplate.name())
+ .withComment(sparkJobTemplate.comment())
+ .withJobType(sparkJobTemplate.jobType())
+ .withExecutable(sparkJobTemplate.executable())
+ .withArguments(sparkJobTemplate.arguments())
+ .withEnvironments(sparkJobTemplate.environments())
+ .withCustomFields(sparkJobTemplate.customFields())
+ .withClassName(sparkJobTemplate.className())
+ .withJars(sparkJobTemplate.jars())
+ .withFiles(sparkJobTemplate.files())
+ .withArchives(sparkJobTemplate.archives())
+ .withConfigs(sparkJobTemplate.configs())
+ .withAudit(audit)
+ .build();
+
+ default:
+ throw new IllegalArgumentException(
+ "Unsupported job template type: " + jobTemplate.jobType());
+ }
+ }
+
+ /**
+ * Deserializes a job entity's stored runtime job template JSON, if any,
back into a {@link
+ * JobTemplateDTO}. {@link JobTemplateDTO}'s {@code @JsonTypeInfo} handles
the Shell/Spark
+ * dispatch automatically.
+ *
+ * @param runtimeJobTemplateJson The serialized runtime job template, or
null if the job has none.
+ * @param jobName The name of the job the template belongs to, used in the
error message on
+ * failure.
+ * @return The deserialized job template DTO, or null if
runtimeJobTemplateJson is null.
+ */
+ public static JobTemplateDTO fromRuntimeJobTemplateJson(
+ String runtimeJobTemplateJson, String jobName) {
+ if (runtimeJobTemplateJson == null) {
+ return null;
+ }
+
+ try {
+ return JsonUtils.anyFieldMapper().readValue(runtimeJobTemplateJson,
JobTemplateDTO.class);
+ } catch (JsonProcessingException e) {
+ // Deliberately excludes the raw JSON content from the message: the
resolved template can
+ // carry sensitive values (env vars, custom fields, credentials
substituted from jobConf),
+ // and untrusted content in a log/exception message is also a
log-injection risk. The
+ // content length plus the cause's own message (which Jackson scopes to
the syntax error,
+ // not the full payload) is enough to debug without echoing arbitrary
stored data.
+ throw new RuntimeException(
+ String.format(
+ "Failed to deserialize the runtime job template for job %s (%d
chars)",
+ jobName, runtimeJobTemplateJson.length()),
+ e);
+ }
+ }
+
/**
* Converts a PolicyContentDTO to a PolicyContent.
*
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 bf5e78fa45..8065893bb0 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
@@ -19,9 +19,11 @@
package org.apache.gravitino.dto.job;
import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.collect.Lists;
import java.time.Instant;
import org.apache.gravitino.dto.AuditDTO;
import org.apache.gravitino.job.JobHandle;
+import org.apache.gravitino.job.JobTemplate;
import org.apache.gravitino.json.JsonUtils;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -41,7 +43,8 @@ public class TestJobDTO {
AuditDTO.builder().withCreator("test").withCreateTime(Instant.now()).build(),
queuedAt,
startedAt,
- finishedAt);
+ finishedAt,
+ null);
Assertions.assertDoesNotThrow(jobDTO::validate);
@@ -68,6 +71,7 @@ public class TestJobDTO {
AuditDTO.builder().withCreator("test").withCreateTime(Instant.now()).build(),
queuedAt,
null,
+ null,
null);
Assertions.assertDoesNotThrow(jobDTO::validate);
@@ -94,7 +98,8 @@ public class TestJobDTO {
AuditDTO.builder().withCreator("test").withCreateTime(createTime).build(),
queuedAt,
startedAt,
- finishedAt);
+ finishedAt,
+ null);
String serJson = JsonUtils.objectMapper().writeValueAsString(jobDTO);
@@ -147,4 +152,59 @@ public class TestJobDTO {
Assertions.assertNull(jobDTO.startedAt());
Assertions.assertNull(jobDTO.finishedAt());
}
+
+ @Test
+ public void testSerDeWithRuntimeJobTemplate() throws JsonProcessingException
{
+ JobTemplateDTO runtimeJobTemplate =
+ ShellJobTemplateDTO.builder()
+ .withJobType(JobTemplate.JobType.SHELL)
+ .withName("resolved-template")
+ .withComment("A resolved job template")
+ .withExecutable("/bin/echo")
+ .withArguments(Lists.newArrayList("resolved-arg"))
+ .withScripts(Lists.newArrayList("/path/to/script.sh"))
+
.withAudit(AuditDTO.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .build();
+ JobDTO jobDTO =
+ new JobDTO(
+ "job-1001",
+ "testTemplate",
+ JobHandle.Status.SUCCEEDED,
+
AuditDTO.builder().withCreator("test").withCreateTime(Instant.now()).build(),
+ Instant.now(),
+ Instant.now(),
+ Instant.now(),
+ runtimeJobTemplate);
+
+ Assertions.assertDoesNotThrow(jobDTO::validate);
+
+ String serJson = JsonUtils.objectMapper().writeValueAsString(jobDTO);
+ Assertions.assertTrue(serJson.contains("\"runtimeJobTemplate\""));
+
+ JobDTO deserJobDTO = JsonUtils.objectMapper().readValue(serJson,
JobDTO.class);
+ Assertions.assertEquals(jobDTO, deserJobDTO);
+ Assertions.assertEquals(runtimeJobTemplate,
deserJobDTO.runtimeJobTemplate());
+ Assertions.assertInstanceOf(ShellJobTemplateDTO.class,
deserJobDTO.runtimeJobTemplate());
+ }
+
+ @Test
+ public void testSerDeWithNullRuntimeJobTemplate() throws
JsonProcessingException {
+ JobDTO jobDTO =
+ new JobDTO(
+ "job-1002",
+ "testTemplate",
+ JobHandle.Status.QUEUED,
+
AuditDTO.builder().withCreator("test").withCreateTime(Instant.now()).build(),
+ Instant.now(),
+ null,
+ null,
+ null);
+
+ Assertions.assertDoesNotThrow(jobDTO::validate);
+
+ String serJson = JsonUtils.objectMapper().writeValueAsString(jobDTO);
+ JobDTO deserJobDTO = JsonUtils.objectMapper().readValue(serJson,
JobDTO.class);
+ Assertions.assertEquals(jobDTO, deserJobDTO);
+ Assertions.assertNull(deserJobDTO.runtimeJobTemplate());
+ }
}
diff --git
a/common/src/test/java/org/apache/gravitino/dto/util/TestDTOConverters.java
b/common/src/test/java/org/apache/gravitino/dto/util/TestDTOConverters.java
index 46a2ffa11b..479bc3a340 100644
--- a/common/src/test/java/org/apache/gravitino/dto/util/TestDTOConverters.java
+++ b/common/src/test/java/org/apache/gravitino/dto/util/TestDTOConverters.java
@@ -19,9 +19,16 @@
package org.apache.gravitino.dto.util;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
+import java.time.Instant;
import java.util.Arrays;
import java.util.Collections;
import java.util.Map;
+import org.apache.gravitino.dto.AuditDTO;
+import org.apache.gravitino.dto.job.JobTemplateDTO;
+import org.apache.gravitino.dto.job.ShellJobTemplateDTO;
+import org.apache.gravitino.dto.job.SparkJobTemplateDTO;
import org.apache.gravitino.dto.rel.expressions.LiteralDTO;
import org.apache.gravitino.dto.rel.partitioning.ListPartitioningDTO;
import org.apache.gravitino.dto.rel.partitioning.RangePartitioningDTO;
@@ -29,6 +36,9 @@ import
org.apache.gravitino.dto.rel.partitions.IdentityPartitionDTO;
import org.apache.gravitino.dto.rel.partitions.ListPartitionDTO;
import org.apache.gravitino.dto.rel.partitions.PartitionDTO;
import org.apache.gravitino.dto.rel.partitions.RangePartitionDTO;
+import org.apache.gravitino.job.JobTemplate;
+import org.apache.gravitino.job.ShellJobTemplate;
+import org.apache.gravitino.job.SparkJobTemplate;
import org.apache.gravitino.rel.expressions.literals.Literal;
import org.apache.gravitino.rel.expressions.literals.Literals;
import org.apache.gravitino.rel.expressions.transforms.Transform;
@@ -246,4 +256,116 @@ public class TestDTOConverters {
Types.StringType.get().simpleString(),
listPartitionAssignments[0].lists()[0][0].value());
Assertions.assertEquals(properties,
listPartitionAssignments[0].properties());
}
+
+ @Test
+ void testJobTemplateToDTOConvertsShellJobTemplate() {
+
+ // given
+ JobTemplate shellJobTemplate =
+ ShellJobTemplate.builder()
+ .withName("shell-job-template")
+ .withComment("This is a shell job template")
+ .withExecutable("/bin/echo")
+ .withArguments(Lists.newArrayList("Hello, World!"))
+ .withEnvironments(ImmutableMap.of("ENV_VAR", "value"))
+ .withScripts(Lists.newArrayList("/path/to/script.sh"))
+ .build();
+ AuditDTO audit =
AuditDTO.builder().withCreator("test").withCreateTime(Instant.now()).build();
+
+ // when
+ JobTemplateDTO jobTemplateDTO = DTOConverters.toDTO(shellJobTemplate,
audit);
+
+ // then
+ Assertions.assertInstanceOf(ShellJobTemplateDTO.class, jobTemplateDTO);
+ Assertions.assertEquals(JobTemplate.JobType.SHELL,
jobTemplateDTO.jobType());
+ Assertions.assertEquals(shellJobTemplate.name(), jobTemplateDTO.name());
+ Assertions.assertEquals(shellJobTemplate.comment(),
jobTemplateDTO.comment());
+ Assertions.assertEquals(shellJobTemplate.executable(),
jobTemplateDTO.executable());
+ Assertions.assertEquals(shellJobTemplate.arguments(),
jobTemplateDTO.arguments());
+ Assertions.assertEquals(shellJobTemplate.environments(),
jobTemplateDTO.environments());
+ Assertions.assertEquals(shellJobTemplate.customFields(),
jobTemplateDTO.customFields());
+ Assertions.assertEquals(
+ ((ShellJobTemplate) shellJobTemplate).scripts(),
+ ((ShellJobTemplateDTO) jobTemplateDTO).scripts());
+ Assertions.assertEquals(audit, jobTemplateDTO.audit());
+
+ // Round-tripping back through fromDTO must reproduce the original
template.
+ Assertions.assertEquals(shellJobTemplate,
DTOConverters.fromDTO(jobTemplateDTO));
+ }
+
+ @Test
+ void testJobTemplateToDTOConvertsSparkJobTemplate() {
+
+ // given
+ JobTemplate sparkJobTemplate =
+ SparkJobTemplate.builder()
+ .withName("spark-job-template")
+ .withComment("This is a spark job template")
+ .withExecutable("/path/to/spark-demo.jar")
+ .withClassName("org.example.SparkDemo")
+ .withArguments(
+ Lists.newArrayList("--input", "/path/to/input", "--output",
"/path/to/output"))
+ .withEnvironments(ImmutableMap.of("SPARK_ENV_VAR", "value"))
+ .withConfigs(ImmutableMap.of("spark.executor.memory", "2g"))
+ .withJars(Lists.newArrayList("/path/to/dependency.jar"))
+ .withFiles(Lists.newArrayList("/path/to/config.yaml"))
+ .withArchives(Lists.newArrayList("/path/to/archive.zip"))
+ .build();
+ AuditDTO audit =
AuditDTO.builder().withCreator("test").withCreateTime(Instant.now()).build();
+
+ // when
+ JobTemplateDTO jobTemplateDTO = DTOConverters.toDTO(sparkJobTemplate,
audit);
+
+ // then
+ Assertions.assertInstanceOf(SparkJobTemplateDTO.class, jobTemplateDTO);
+ Assertions.assertEquals(JobTemplate.JobType.SPARK,
jobTemplateDTO.jobType());
+ Assertions.assertEquals(sparkJobTemplate.name(), jobTemplateDTO.name());
+ Assertions.assertEquals(sparkJobTemplate.comment(),
jobTemplateDTO.comment());
+ Assertions.assertEquals(sparkJobTemplate.executable(),
jobTemplateDTO.executable());
+ Assertions.assertEquals(sparkJobTemplate.arguments(),
jobTemplateDTO.arguments());
+ Assertions.assertEquals(sparkJobTemplate.environments(),
jobTemplateDTO.environments());
+ Assertions.assertEquals(sparkJobTemplate.customFields(),
jobTemplateDTO.customFields());
+ SparkJobTemplate expectedSpark = (SparkJobTemplate) sparkJobTemplate;
+ SparkJobTemplateDTO actualSparkDTO = (SparkJobTemplateDTO) jobTemplateDTO;
+ Assertions.assertEquals(expectedSpark.className(),
actualSparkDTO.className());
+ Assertions.assertEquals(expectedSpark.jars(), actualSparkDTO.jars());
+ Assertions.assertEquals(expectedSpark.files(), actualSparkDTO.files());
+ Assertions.assertEquals(expectedSpark.archives(),
actualSparkDTO.archives());
+ Assertions.assertEquals(expectedSpark.configs(), actualSparkDTO.configs());
+ Assertions.assertEquals(audit, jobTemplateDTO.audit());
+
+ // Round-tripping back through fromDTO must reproduce the original
template.
+ Assertions.assertEquals(sparkJobTemplate,
DTOConverters.fromDTO(jobTemplateDTO));
+ }
+
+ @Test
+ public void testFromRuntimeJobTemplateJsonNull() {
+ Assertions.assertNull(DTOConverters.fromRuntimeJobTemplateJson(null,
"job-1"));
+ }
+
+ @Test
+ public void testFromRuntimeJobTemplateJson() {
+ String runtimeJobTemplateJson =
+
"{\"jobType\":\"shell\",\"name\":\"shell_template_1\",\"executable\":\"/bin/echo\"}";
+ JobTemplateDTO jobTemplateDTO =
+ DTOConverters.fromRuntimeJobTemplateJson(runtimeJobTemplateJson,
"job-1");
+ Assertions.assertInstanceOf(ShellJobTemplateDTO.class, jobTemplateDTO);
+ Assertions.assertEquals("shell_template_1", jobTemplateDTO.name());
+ Assertions.assertEquals("/bin/echo", ((ShellJobTemplateDTO)
jobTemplateDTO).executable());
+ }
+
+ @Test
+ public void testFromRuntimeJobTemplateJsonMalformedMessageOmitsRawContent() {
+ // The message must name the job (for debugging which row is affected) but
must not echo the
+ // raw stored content - the resolved template can carry sensitive values
(env vars, custom
+ // fields, credentials substituted from jobConf), and untrusted content in
a log/exception
+ // message is also a log-injection risk.
+ String malformedJson = "{not-valid-json";
+ RuntimeException exception =
+ Assertions.assertThrows(
+ RuntimeException.class,
+ () -> DTOConverters.fromRuntimeJobTemplateJson(malformedJson,
"job-1"));
+ Assertions.assertTrue(exception.getMessage().contains("job-1"));
+ Assertions.assertFalse(exception.getMessage().contains(malformedJson));
+ }
}
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 42df607c5e..6db68afb14 100644
--- a/core/src/main/java/org/apache/gravitino/job/JobManager.java
+++ b/core/src/main/java/org/apache/gravitino/job/JobManager.java
@@ -21,6 +21,7 @@ package org.apache.gravitino.job;
import static org.apache.gravitino.metalake.MetalakeManager.checkMetalake;
+import com.fasterxml.jackson.core.JsonProcessingException;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import java.io.File;
@@ -49,11 +50,14 @@ import org.apache.gravitino.EntityStore;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.connector.job.JobExecutor;
+import org.apache.gravitino.dto.job.JobTemplateDTO;
+import org.apache.gravitino.dto.util.DTOConverters;
import org.apache.gravitino.exceptions.InUseException;
import org.apache.gravitino.exceptions.JobTemplateAlreadyExistsException;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NoSuchJobException;
import org.apache.gravitino.exceptions.NoSuchJobTemplateException;
+import org.apache.gravitino.json.JsonUtils;
import org.apache.gravitino.lock.LockType;
import org.apache.gravitino.lock.TreeLockUtils;
import org.apache.gravitino.meta.AuditInfo;
@@ -443,6 +447,19 @@ public class JobManager implements JobOperationDispatcher {
// also downloading any necessary files from the URIs specified in the job
template.
JobTemplate jobTemplate = createRuntimeJobTemplate(jobTemplateEntity,
jobConf, jobStagingDir);
+ // Serialize the resolved (placeholder-replaced) job template so callers
can later see exactly
+ // what was submitted for execution, not just the original template. This
is done before
+ // submission so that a serialization failure never leaves a job running
on the executor
+ // without a corresponding JobEntity.
+ JobTemplateDTO runtimeJobTemplateDTO =
+ DTOConverters.toDTO(jobTemplate,
DTOConverters.toDTO(jobTemplateEntity.auditInfo()));
+ String runtimeJobTemplateJson;
+ try {
+ runtimeJobTemplateJson =
JsonUtils.anyFieldMapper().writeValueAsString(runtimeJobTemplateDTO);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException("Failed to serialize the runtime job
template", e);
+ }
+
// Submit the job template to the job executor
String jobExecutionId;
try {
@@ -468,6 +485,7 @@ public class JobManager implements JobOperationDispatcher {
// A newly submitted job is queued, not started or finished yet.
.withStartedAt(0L)
.withFinishedAt(0L)
+ .withRuntimeJobTemplate(runtimeJobTemplateJson)
.build();
try {
@@ -559,6 +577,8 @@ public class JobManager implements JobOperationDispatcher {
// job already had.
.withStartedAt(latestJobEntity.startedAt())
.withFinishedAt(latestJobEntity.finishedAt())
+ // The runtime job template is fixed at job creation and never changes.
+ .withRuntimeJobTemplate(latestJobEntity.runtimeJobTemplate())
.build();
}
@@ -729,6 +749,8 @@ public class JobManager implements JobOperationDispatcher {
.build())
.withStartedAt(startedAt)
.withFinishedAt(finishedAt)
+ // The runtime job template is fixed at job creation and never changes.
+ .withRuntimeJobTemplate(latestJobEntity.runtimeJobTemplate())
.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 121c1b0112..cf6b9be02f 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
@@ -23,8 +23,13 @@ import java.time.Instant;
import javax.annotation.Nullable;
import org.apache.gravitino.Audit;
import org.apache.gravitino.annotation.DeveloperApi;
+import org.apache.gravitino.dto.job.JobTemplateDTO;
+import org.apache.gravitino.dto.util.DTOConverters;
import org.apache.gravitino.job.JobHandle;
+import org.apache.gravitino.job.JobTemplate;
import org.apache.gravitino.meta.JobEntity;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Represents information about a job, including its ID, template name,
status, and audit details.
@@ -32,6 +37,8 @@ import org.apache.gravitino.meta.JobEntity;
@DeveloperApi
public final class JobInfo {
+ private static final Logger LOG = LoggerFactory.getLogger(JobInfo.class);
+
private final String jobId;
private final String jobTemplateName;
@@ -44,19 +51,23 @@ public final class JobInfo {
private final Instant finishedAt;
+ private final JobTemplate runtimeJobTemplate;
+
private JobInfo(
String jobId,
String jobTemplateName,
JobHandle.Status jobStatus,
Audit audit,
Instant startedAt,
- Instant finishedAt) {
+ Instant finishedAt,
+ JobTemplate runtimeJobTemplate) {
this.jobId = jobId;
this.jobTemplateName = jobTemplateName;
this.jobStatus = jobStatus;
this.audit = audit;
this.startedAt = startedAt;
this.finishedAt = finishedAt;
+ this.runtimeJobTemplate = runtimeJobTemplate;
}
/**
@@ -72,7 +83,32 @@ public final class JobInfo {
jobEntity.status(),
jobEntity.auditInfo(),
jobEntity.startedAtAsInstant(),
- jobEntity.finishedAtAsInstant());
+ jobEntity.finishedAtAsInstant(),
+ toRuntimeJobTemplate(jobEntity));
+ }
+
+ /**
+ * Deserializes the job entity's stored runtime job template JSON, if any,
back into a {@link
+ * JobTemplate}. This is called from inside the event dispatcher's try block
for getJob/runJob/
+ * cancelJob, after the underlying operation has already succeeded (or, for
cancelJob, after the
+ * external cancel call and entity update have already happened) - a
malformed or
+ * forward-incompatible stored value (e.g. a job type unknown to this server
version) must not
+ * turn that already-completed operation into a 500, so failures here are
logged and swallowed
+ * rather than propagated.
+ */
+ private static JobTemplate toRuntimeJobTemplate(JobEntity jobEntity) {
+ try {
+ JobTemplateDTO runtimeJobTemplateDTO =
+ DTOConverters.fromRuntimeJobTemplateJson(
+ jobEntity.runtimeJobTemplate(), jobEntity.name());
+ return runtimeJobTemplateDTO == null ? null :
DTOConverters.fromDTO(runtimeJobTemplateDTO);
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to build the runtime job template for job {}, omitting it
from the event",
+ jobEntity.name(),
+ e);
+ return null;
+ }
}
/**
@@ -140,4 +176,15 @@ public final class JobInfo {
public Instant finishedAt() {
return finishedAt;
}
+
+ /**
+ * Returns the resolved job template that was actually submitted for
execution, with placeholders
+ * replaced and referenced files downloaded.
+ *
+ * @return the runtime job template, or null for jobs run before this field
was introduced
+ */
+ @Nullable
+ public JobTemplate runtimeJobTemplate() {
+ return runtimeJobTemplate;
+ }
}
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 c7c2a0bec5..ff1e1f81a6 100644
--- a/core/src/main/java/org/apache/gravitino/meta/JobEntity.java
+++ b/core/src/main/java/org/apache/gravitino/meta/JobEntity.java
@@ -62,6 +62,13 @@ public class JobEntity implements Entity, Auditable,
HasIdentifier {
Long.class,
"The time when the job finished execution, using the storage layer's
"
+ "\"not finished\" sentinel (<= 0) when the job has not
finished execution yet.");
+ public static final Field RUNTIME_JOB_TEMPLATE =
+ Field.optional(
+ "runtime_job_template",
+ String.class,
+ "The resolved job template that was actually submitted for
execution, serialized as "
+ + "JSON, with placeholders replaced and referenced files
downloaded. Null for jobs "
+ + "run before this field was introduced.");
private Long id;
private String jobExecutionId;
@@ -71,6 +78,7 @@ public class JobEntity implements Entity, Auditable,
HasIdentifier {
private AuditInfo auditInfo;
private Long startedAt;
private Long finishedAt;
+ private String runtimeJobTemplate;
private JobEntity() {}
@@ -84,6 +92,7 @@ public class JobEntity implements Entity, Auditable,
HasIdentifier {
fields.put(AUDIT_INFO, auditInfo);
fields.put(STARTED_AT, startedAt);
fields.put(FINISHED_AT, finishedAt);
+ fields.put(RUNTIME_JOB_TEMPLATE, runtimeJobTemplate);
return Collections.unmodifiableMap(fields);
}
@@ -148,6 +157,18 @@ public class JobEntity implements Entity, Auditable,
HasIdentifier {
return (finishedAt == null || finishedAt <= 0) ? null :
Instant.ofEpochMilli(finishedAt);
}
+ /**
+ * Returns the resolved job template that was actually submitted for
execution, serialized as JSON
+ * (placeholders replaced, referenced files downloaded).
+ *
+ * @return the serialized runtime job template, or {@code null} for jobs run
before this field was
+ * introduced
+ */
+ @Nullable
+ public String runtimeJobTemplate() {
+ return runtimeJobTemplate;
+ }
+
@Override
public AuditInfo auditInfo() {
return auditInfo;
@@ -175,13 +196,22 @@ public class JobEntity implements Entity, Auditable,
HasIdentifier {
&& Objects.equals(namespace, that.namespace)
&& Objects.equals(auditInfo, that.auditInfo)
&& Objects.equals(startedAt, that.startedAt)
- && Objects.equals(finishedAt, that.finishedAt);
+ && Objects.equals(finishedAt, that.finishedAt)
+ && Objects.equals(runtimeJobTemplate, that.runtimeJobTemplate);
}
@Override
public int hashCode() {
return Objects.hash(
- id, jobExecutionId, namespace, status, jobTemplateName, auditInfo,
startedAt, finishedAt);
+ id,
+ jobExecutionId,
+ namespace,
+ status,
+ jobTemplateName,
+ auditInfo,
+ startedAt,
+ finishedAt,
+ runtimeJobTemplate);
}
public static Builder builder() {
@@ -235,6 +265,11 @@ public class JobEntity implements Entity, Auditable,
HasIdentifier {
return this;
}
+ public Builder withRuntimeJobTemplate(String runtimeJobTemplate) {
+ jobEntity.runtimeJobTemplate = runtimeJobTemplate;
+ return this;
+ }
+
public JobEntity build() {
jobEntity.validate();
return jobEntity;
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 a18d45e73b..cf8661b452 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,7 +31,8 @@ public class JobMetaBaseSQLProvider {
return "INSERT INTO "
+ JobMetaMapper.TABLE_NAME
+ " (job_run_id, job_template_id, metalake_id,"
- + " job_execution_id, job_run_status, job_started_at, job_finished_at,
audit_info,"
+ + " job_execution_id, job_run_status, job_started_at, job_finished_at,"
+ + " runtime_job_template, audit_info,"
+ " current_version, last_version, deleted_at)"
+ " VALUES (#{jobMeta.jobRunId},"
+ " (SELECT job_template_id FROM "
@@ -40,15 +41,16 @@ public class JobMetaBaseSQLProvider {
+ " AND metalake_id = #{jobMeta.metalakeId} AND deleted_at = 0),"
+ " #{jobMeta.metalakeId}, #{jobMeta.jobExecutionId},"
+ " #{jobMeta.jobRunStatus}, #{jobMeta.jobStartedAt},
#{jobMeta.jobFinishedAt},"
- + " #{jobMeta.auditInfo}, #{jobMeta.currentVersion},
#{jobMeta.lastVersion},"
- + " #{jobMeta.deletedAt})";
+ + " #{jobMeta.runtimeJobTemplate}, #{jobMeta.auditInfo},
#{jobMeta.currentVersion},"
+ + " #{jobMeta.lastVersion}, #{jobMeta.deletedAt})";
}
public String insertJobMetaOnDuplicateKeyUpdate(@Param("jobMeta") JobPO
jobPO) {
return "INSERT INTO "
+ JobMetaMapper.TABLE_NAME
+ " (job_run_id, job_template_id, metalake_id,"
- + " job_execution_id, job_run_status, job_started_at, job_finished_at,
audit_info,"
+ + " job_execution_id, job_run_status, job_started_at, job_finished_at,"
+ + " runtime_job_template, audit_info,"
+ " current_version, last_version, deleted_at)"
+ " VALUES (#{jobMeta.jobRunId},"
+ " (SELECT job_template_id FROM "
@@ -57,8 +59,8 @@ public class JobMetaBaseSQLProvider {
+ " AND metalake_id = #{jobMeta.metalakeId} AND deleted_at = 0),"
+ " #{jobMeta.metalakeId}, #{jobMeta.jobExecutionId},"
+ " #{jobMeta.jobRunStatus}, #{jobMeta.jobStartedAt},
#{jobMeta.jobFinishedAt},"
- + " #{jobMeta.auditInfo}, #{jobMeta.currentVersion},
#{jobMeta.lastVersion},"
- + " #{jobMeta.deletedAt})"
+ + " #{jobMeta.runtimeJobTemplate}, #{jobMeta.auditInfo},
#{jobMeta.currentVersion},"
+ + " #{jobMeta.lastVersion}, #{jobMeta.deletedAt})"
+ " ON DUPLICATE KEY UPDATE"
+ " job_template_id = (SELECT job_template_id FROM "
+ JobTemplateMetaMapper.TABLE_NAME
@@ -69,6 +71,7 @@ public class JobMetaBaseSQLProvider {
+ " job_run_status = #{jobMeta.jobRunStatus},"
+ " job_started_at = #{jobMeta.jobStartedAt},"
+ " job_finished_at = #{jobMeta.jobFinishedAt},"
+ + " runtime_job_template = #{jobMeta.runtimeJobTemplate},"
+ " audit_info = #{jobMeta.auditInfo},"
+ " current_version = #{jobMeta.currentVersion},"
+ " last_version = #{jobMeta.lastVersion},"
@@ -80,6 +83,7 @@ public class JobMetaBaseSQLProvider {
+ " jrm.metalake_id AS metalakeId, jrm.job_execution_id AS
jobExecutionId,"
+ " jrm.job_run_status AS jobRunStatus, jrm.job_started_at AS
jobStartedAt,"
+ " jrm.job_finished_at AS jobFinishedAt,"
+ + " jrm.runtime_job_template AS runtimeJobTemplate,"
+ " jrm.audit_info AS auditInfo,"
+ " jrm.current_version AS currentVersion, jrm.last_version AS
lastVersion,"
+ " jrm.deleted_at AS deletedAt"
@@ -102,6 +106,7 @@ public class JobMetaBaseSQLProvider {
+ " jrm.metalake_id AS metalakeId, jrm.job_execution_id AS
jobExecutionId,"
+ " jrm.job_run_status AS jobRunStatus, jrm.job_started_at AS
jobStartedAt,"
+ " jrm.job_finished_at AS jobFinishedAt,"
+ + " jrm.runtime_job_template AS runtimeJobTemplate,"
+ " jrm.audit_info AS auditInfo,"
+ " jrm.current_version AS currentVersion, jrm.last_version AS
lastVersion,"
+ " jrm.deleted_at AS deletedAt"
@@ -123,6 +128,7 @@ public class JobMetaBaseSQLProvider {
+ " jrm.metalake_id AS metalakeId, jrm.job_execution_id AS
jobExecutionId,"
+ " jrm.job_run_status AS jobRunStatus, jrm.job_started_at AS
jobStartedAt,"
+ " jrm.job_finished_at AS jobFinishedAt,"
+ + " jrm.runtime_job_template AS runtimeJobTemplate,"
+ " jrm.audit_info AS auditInfo,"
+ " jrm.current_version AS currentVersion, jrm.last_version AS
lastVersion,"
+ " jrm.deleted_at AS deletedAt"
@@ -146,6 +152,7 @@ public class JobMetaBaseSQLProvider {
+ " job_run_status = #{newJobMeta.jobRunStatus},"
+ " job_started_at = #{newJobMeta.jobStartedAt},"
+ " job_finished_at = #{newJobMeta.jobFinishedAt},"
+ + " runtime_job_template = #{newJobMeta.runtimeJobTemplate},"
+ " audit_info = #{newJobMeta.auditInfo},"
+ " current_version = #{newJobMeta.currentVersion},"
+ " last_version = #{newJobMeta.lastVersion}"
@@ -211,6 +218,7 @@ public class JobMetaBaseSQLProvider {
+ " jrm.metalake_id AS metalakeId, jrm.job_execution_id AS
jobExecutionId,"
+ " jrm.job_run_status AS jobRunStatus, jrm.job_started_at AS
jobStartedAt,"
+ " jrm.job_finished_at AS jobFinishedAt,"
+ + " jrm.runtime_job_template AS runtimeJobTemplate,"
+ " 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 ffe4ca8ea8..73d2364cc1 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,7 +32,8 @@ 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_started_at, job_finished_at,
audit_info,"
+ + " job_execution_id, job_run_status, job_started_at, job_finished_at,"
+ + " runtime_job_template, audit_info,"
+ " current_version, last_version, deleted_at)"
+ " VALUES (#{jobMeta.jobRunId},"
+ " (SELECT job_template_id FROM "
@@ -41,8 +42,8 @@ public class JobMetaPostgreSQLProvider extends
JobMetaBaseSQLProvider {
+ " AND metalake_id = #{jobMeta.metalakeId} AND deleted_at = 0),"
+ " #{jobMeta.metalakeId}, #{jobMeta.jobExecutionId},"
+ " #{jobMeta.jobRunStatus}, #{jobMeta.jobStartedAt},
#{jobMeta.jobFinishedAt},"
- + " #{jobMeta.auditInfo}, #{jobMeta.currentVersion},
#{jobMeta.lastVersion},"
- + " #{jobMeta.deletedAt})"
+ + " #{jobMeta.runtimeJobTemplate}, #{jobMeta.auditInfo},
#{jobMeta.currentVersion},"
+ + " #{jobMeta.lastVersion}, #{jobMeta.deletedAt})"
+ " ON CONFLICT (job_run_id) DO UPDATE SET"
+ " job_template_id = (SELECT job_template_id FROM "
+ JobTemplateMetaMapper.TABLE_NAME
@@ -53,6 +54,7 @@ public class JobMetaPostgreSQLProvider extends
JobMetaBaseSQLProvider {
+ " job_run_status = #{jobMeta.jobRunStatus},"
+ " job_started_at = #{jobMeta.jobStartedAt},"
+ " job_finished_at = #{jobMeta.jobFinishedAt},"
+ + " runtime_job_template = #{jobMeta.runtimeJobTemplate},"
+ " audit_info = #{jobMeta.auditInfo},"
+ " current_version = #{jobMeta.currentVersion},"
+ " last_version = #{jobMeta.lastVersion},"
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 bf0dbdf41d..871ef074a1 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
@@ -47,6 +47,7 @@ public class JobPO {
private String jobRunStatus;
private Long jobStartedAt;
private Long jobFinishedAt;
+ private String runtimeJobTemplate;
private String auditInfo;
private Long currentVersion;
private Long lastVersion;
@@ -65,6 +66,7 @@ public class JobPO {
String jobRunStatus,
Long jobStartedAt,
Long jobFinishedAt,
+ String runtimeJobTemplate,
String auditInfo,
Long currentVersion,
Long lastVersion,
@@ -79,6 +81,8 @@ public class JobPO {
StringUtils.isNotBlank(jobRunStatus), "jobRunStatus cannot be blank");
Preconditions.checkArgument(jobStartedAt != null, "jobStartedAt cannot be
null");
Preconditions.checkArgument(jobFinishedAt != null, "jobFinishedAt cannot
be null");
+ // runtimeJobTemplate is legitimately nullable: rows created before this
field was introduced
+ // have no resolved template to backfill.
Preconditions.checkArgument(StringUtils.isNotBlank(auditInfo), "auditInfo
cannot be blank");
Preconditions.checkArgument(currentVersion != null, "currentVersion cannot
be null");
Preconditions.checkArgument(lastVersion != null, "lastVersion cannot be
null");
@@ -91,6 +95,7 @@ public class JobPO {
this.jobRunStatus = jobRunStatus;
this.jobStartedAt = jobStartedAt;
this.jobFinishedAt = jobFinishedAt;
+ this.runtimeJobTemplate = runtimeJobTemplate;
this.auditInfo = auditInfo;
this.currentVersion = currentVersion;
this.lastVersion = lastVersion;
@@ -116,6 +121,7 @@ public class JobPO {
.withJobRunStatus(jobEntity.status().name())
.withJobStartedAt(jobEntity.startedAt())
.withJobFinishedAt(jobEntity.finishedAt())
+ .withRuntimeJobTemplate(jobEntity.runtimeJobTemplate())
.withAuditInfo(JsonUtils.anyFieldMapper().writeValueAsString(jobEntity.auditInfo()))
.withCurrentVersion(INIT_VERSION)
.withLastVersion(INIT_VERSION)
@@ -128,9 +134,13 @@ public class JobPO {
/**
* Builds the {@link JobPO} to persist for an update, carrying forward the
identity fields ({@code
- * jobRunId}, {@code jobTemplateName}) from the old PO since a job's
template is immutable, and
- * bumping the version counters. This does not perform any
optimistic-concurrency check; the
- * caller is responsible for any such guarantee.
+ * jobRunId}, {@code jobTemplateName}) from the old PO since a job's
template is immutable once
+ * the job is created, and bumping the version counters. Unlike those
identity fields, {@code
+ * runtimeJobTemplate} is taken from {@code newJobEntity} rather than
carried forward from the old
+ * PO - this layer stores whatever the caller passes, it does not enforce
that the resolved
+ * runtime template never changes. That invariant is the caller's
responsibility (see {@code
+ * JobManager}'s updater functions, which always carry the existing value
forward). This does not
+ * perform any optimistic-concurrency check; the caller is responsible for
any such guarantee.
*
* @param oldJobPO the existing {@link JobPO} being updated
* @param newJobEntity the {@link JobEntity} with the updated
status/timestamps/audit info
@@ -149,6 +159,7 @@ public class JobPO {
.withJobRunStatus(newJobEntity.status().name())
.withJobStartedAt(newJobEntity.startedAt())
.withJobFinishedAt(newJobEntity.finishedAt())
+ .withRuntimeJobTemplate(newJobEntity.runtimeJobTemplate())
.withAuditInfo(JsonUtils.anyFieldMapper().writeValueAsString(newJobEntity.auditInfo()))
.withCurrentVersion(currentVersion)
.withLastVersion(lastVersion)
@@ -170,6 +181,7 @@ public class JobPO {
.withAuditInfo(JsonUtils.anyFieldMapper().readValue(jobPO.auditInfo,
AuditInfo.class))
.withStartedAt(jobPO.jobStartedAt())
.withFinishedAt(jobPO.jobFinishedAt())
+ .withRuntimeJobTemplate(jobPO.runtimeJobTemplate())
.build();
} catch (JsonProcessingException e) {
throw new RuntimeException("Failed to deserialize job PO", 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 c1345d0f77..ac88c2974e 100644
--- a/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
+++ b/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
@@ -61,6 +61,8 @@ import org.apache.gravitino.GravitinoEnv;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.connector.job.JobExecutor;
+import org.apache.gravitino.dto.job.JobTemplateDTO;
+import org.apache.gravitino.dto.job.ShellJobTemplateDTO;
import org.apache.gravitino.exceptions.InUseException;
import org.apache.gravitino.exceptions.JobTemplateAlreadyExistsException;
import org.apache.gravitino.exceptions.MetalakeInUseException;
@@ -68,6 +70,7 @@ import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NoSuchJobException;
import org.apache.gravitino.exceptions.NoSuchJobTemplateException;
import org.apache.gravitino.exceptions.NoSuchMetalakeException;
+import org.apache.gravitino.json.JsonUtils;
import org.apache.gravitino.lock.LockManager;
import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.meta.BaseMetalake;
@@ -519,6 +522,59 @@ public class TestJobManager {
() -> jobManager.runJob(metalake, "shell_job",
Collections.emptyMap()));
}
+ @Test
+ public void testRunJobPopulatesResolvedRuntimeJobTemplate() throws
IOException {
+ mockedMetalake
+ .when(() -> MetalakeManager.checkMetalake(metalakeIdent, entityStore))
+ .thenAnswer(a -> null);
+
+ ShellJobTemplate templateWithPlaceholder =
+ ShellJobTemplate.builder()
+ .withName("shell_job_with_placeholder")
+ .withComment("A shell job template with a placeholder")
+ .withExecutable("/bin/echo")
+ .withArguments(Lists.newArrayList("{{greeting}}"))
+ .build();
+ JobTemplateEntity jobTemplateEntity =
+ JobTemplateEntity.builder()
+ .withId(new Random().nextLong())
+ .withName(templateWithPlaceholder.name())
+ .withNamespace(NamespaceUtil.ofJobTemplate(metalake))
+ .withTemplateContent(
+
JobTemplateEntity.TemplateContent.fromJobTemplate(templateWithPlaceholder))
+ .withComment(templateWithPlaceholder.comment())
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .build();
+ when(jobManager.getJobTemplate(metalake, jobTemplateEntity.name()))
+ .thenReturn(jobTemplateEntity);
+
+ when(jobExecutor.submitJob(any())).thenReturn("job_execution_id_for_test");
+ doNothing().when(entityStore).put(any(JobEntity.class), anyBoolean());
+
+ JobEntity jobEntity =
+ jobManager.runJob(
+ metalake, jobTemplateEntity.name(),
Collections.singletonMap("greeting", "Hello!"));
+
+ Assertions.assertNotNull(jobEntity.runtimeJobTemplate());
+ ShellJobTemplateDTO runtimeJobTemplateDTO =
+ (ShellJobTemplateDTO)
+ JsonUtils.anyFieldMapper()
+ .readValue(jobEntity.runtimeJobTemplate(),
JobTemplateDTO.class);
+
+ // The resolved runtime template must carry the actual value substituted
for the placeholder,
+ // not the original template's raw {{greeting}} string.
+ Assertions.assertEquals(Lists.newArrayList("Hello!"),
runtimeJobTemplateDTO.arguments());
+ Assertions.assertEquals(jobTemplateEntity.name(),
runtimeJobTemplateDTO.name());
+ Assertions.assertEquals(jobTemplateEntity.comment(),
runtimeJobTemplateDTO.comment());
+ // createRuntimeJobTemplate() also resolves the executable by fetching it
into the job's
+ // staging directory, so it ends up as a local staging-dir path rather
than the original
+ // "/bin/echo" - just confirm it was actually resolved to something under
that directory.
+ Assertions.assertTrue(
+ runtimeJobTemplateDTO.executable().endsWith("echo"),
+ () -> "Unexpected resolved executable: " +
runtimeJobTemplateDTO.executable());
+ }
+
@Test
public void testRunJobSucceedsWhenStagingDirectoryAlreadyExists() throws
Exception {
mockedMetalake
@@ -692,6 +748,37 @@ public class TestJobManager {
Assertions.assertEquals(0L, result.finishedAt());
}
+ @Test
+ public void testCancelJobPreservesRuntimeJobTemplate() throws IOException {
+ mockedMetalake
+ .when(() -> MetalakeManager.checkMetalake(metalakeIdent, entityStore))
+ .thenAnswer(a -> null);
+
+ String runtimeJobTemplateJson =
+
"{\"jobType\":\"shell\",\"name\":\"shell_job\",\"executable\":\"/bin/echo\"}";
+ JobEntity job =
+ JobEntity.builder()
+ .withId(new Random().nextLong())
+ .withJobExecutionId(new Random().nextLong() + "")
+ .withNamespace(NamespaceUtil.ofJob(metalake))
+ .withJobTemplateName("shell_job")
+ .withStatus(JobHandle.Status.QUEUED)
+ .withStartedAt(0L)
+ .withFinishedAt(0L)
+ .withRuntimeJobTemplate(runtimeJobTemplateJson)
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .build();
+ when(jobManager.getJob(metalake, job.name())).thenReturn(job);
+ doNothing().when(jobExecutor).cancelJob(job.jobExecutionId());
+ stubEntityStoreUpdateToApply(job);
+
+ // The runtime job template is fixed at job creation, so cancelling must
carry it forward
+ // unchanged rather than dropping it while rebuilding the entity for the
CANCELLING status.
+ JobEntity cancelledJob = jobManager.cancelJob(metalake, job.name());
+ Assertions.assertEquals(runtimeJobTemplateJson,
cancelledJob.runtimeJobTemplate());
+ }
+
@Test
public void testPullJobStatus() throws IOException {
JobEntity job = newJobEntity("shell_job", JobHandle.Status.QUEUED);
@@ -728,6 +815,47 @@ public class TestJobManager {
Assertions.assertTrue(updatedJob.finishedAt() > 0);
}
+ @Test
+ public void testPullJobStatusPreservesRuntimeJobTemplate() throws
IOException {
+ String runtimeJobTemplateJson =
+
"{\"jobType\":\"shell\",\"name\":\"shell_job\",\"executable\":\"/bin/echo\"}";
+ JobEntity job =
+ JobEntity.builder()
+ .withId(new Random().nextLong())
+ .withJobExecutionId(new Random().nextLong() + "")
+ .withNamespace(NamespaceUtil.ofJob(metalake))
+ .withJobTemplateName("shell_job")
+ .withStatus(JobHandle.Status.QUEUED)
+ .withStartedAt(0L)
+ .withFinishedAt(0L)
+ .withRuntimeJobTemplate(runtimeJobTemplateJson)
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .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));
+
when(jobExecutor.getJobStatus(job.jobExecutionId())).thenReturn(JobHandle.Status.SUCCEEDED);
+ stubEntityStoreUpdateToApply(job);
+
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+
+ // The runtime job template is fixed at job creation, so a status-poll
update must carry it
+ // forward unchanged rather than dropping it while rebuilding the entity
for the new status.
+ JobEntity updatedJob = captureUpdatedJobEntity(job);
+ Assertions.assertEquals(runtimeJobTemplateJson,
updatedJob.runtimeJobTemplate());
+ }
+
@Test
public void testPullJobStatusStartedAt() throws IOException {
JobEntity job =
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
index 833a37fbdc..ef16fe4de9 100644
--- 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
@@ -18,8 +18,14 @@
*/
package org.apache.gravitino.listener.api.info;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
import java.time.Instant;
+import org.apache.gravitino.dto.util.DTOConverters;
import org.apache.gravitino.job.JobHandle;
+import org.apache.gravitino.job.JobTemplate;
+import org.apache.gravitino.job.ShellJobTemplate;
+import org.apache.gravitino.json.JsonUtils;
import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.meta.JobEntity;
import org.apache.gravitino.utils.NamespaceUtil;
@@ -76,4 +82,88 @@ public class TestJobInfo {
Assertions.assertEquals(Instant.ofEpochMilli(startedAt),
jobInfo.startedAt());
Assertions.assertEquals(Instant.ofEpochMilli(finishedAt),
jobInfo.finishedAt());
}
+
+ @Test
+ public void testFromJobEntityWithoutRuntimeJobTemplate() {
+ 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(Instant.now()).build())
+ .withStartedAt(0L)
+ .withFinishedAt(0L)
+ .build();
+
+ JobInfo jobInfo = JobInfo.fromJobEntity(jobEntity);
+
+ // No runtime job template stored (e.g. a job run before this field was
introduced) - must
+ // round-trip as null rather than failing to convert.
+ Assertions.assertNull(jobInfo.runtimeJobTemplate());
+ }
+
+ @Test
+ public void testFromJobEntityWithRuntimeJobTemplate() throws Exception {
+ JobTemplate resolvedTemplate =
+ ShellJobTemplate.builder()
+ .withName("test-job-template")
+ .withComment("resolved")
+ .withExecutable("/bin/echo")
+ .withArguments(Lists.newArrayList("resolved-arg"))
+ .withEnvironments(ImmutableMap.of("ENV_VAR", "resolved-value"))
+ .build();
+ AuditInfo auditInfo =
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build();
+ String runtimeJobTemplateJson =
+ JsonUtils.anyFieldMapper()
+ .writeValueAsString(
+ DTOConverters.toDTO(resolvedTemplate,
DTOConverters.toDTO(auditInfo)));
+
+ JobEntity jobEntity =
+ JobEntity.builder()
+ .withId(1L)
+ .withJobExecutionId("job-execution-1")
+ .withJobTemplateName("test-job-template")
+ .withStatus(JobHandle.Status.QUEUED)
+ .withNamespace(NamespaceUtil.ofJob("test"))
+ .withAuditInfo(auditInfo)
+ .withStartedAt(0L)
+ .withFinishedAt(0L)
+ .withRuntimeJobTemplate(runtimeJobTemplateJson)
+ .build();
+
+ JobInfo jobInfo = JobInfo.fromJobEntity(jobEntity);
+
+ Assertions.assertEquals(resolvedTemplate, jobInfo.runtimeJobTemplate());
+ }
+
+ @Test
+ public void testFromJobEntityWithMalformedRuntimeJobTemplateDoesNotThrow() {
+ // fromJobEntity() is called inside JobEventDispatcher's try block for
getJob/runJob/
+ // cancelJob, after the underlying operation has already succeeded - a
malformed or
+ // forward-incompatible stored runtime job template (e.g. a job type
unknown to this server
+ // version) must not turn that already-completed operation into a failure.
It should just be
+ // omitted from the built JobInfo.
+ 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(Instant.now()).build())
+ .withStartedAt(0L)
+ .withFinishedAt(0L)
+ .withRuntimeJobTemplate("{not-valid-json")
+ .build();
+
+ JobInfo jobInfo = Assertions.assertDoesNotThrow(() ->
JobInfo.fromJobEntity(jobEntity));
+
+ Assertions.assertNull(jobInfo.runtimeJobTemplate());
+ Assertions.assertEquals(jobEntity.name(), jobInfo.jobId());
+ }
}
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 cc9854849f..24e8d729d5 100644
--- a/core/src/test/java/org/apache/gravitino/meta/TestJobEntity.java
+++ b/core/src/test/java/org/apache/gravitino/meta/TestJobEntity.java
@@ -318,4 +318,90 @@ public class TestJobEntity {
Assertions.assertNotEquals(notFinished, finished);
Assertions.assertNotEquals(notFinished.hashCode(), finished.hashCode());
}
+
+ @Test
+ public void testRuntimeJobTemplateDefaultsToNullWhenNotSet() {
+ // Unlike startedAt/finishedAt, runtimeJobTemplate is optional - jobs run
before this field
+ // was introduced have no resolved template to backfill, so building
without it must not
+ // throw.
+ JobEntity jobEntity =
+ 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(jobEntity.runtimeJobTemplate());
+ }
+
+ @Test
+ public void testRuntimeJobTemplate() {
+ String runtimeJobTemplateJson =
"{\"jobType\":\"shell\",\"name\":\"test-job-template\"}";
+ JobEntity jobEntity =
+ 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)
+ .withRuntimeJobTemplate(runtimeJobTemplateJson)
+ .build();
+
+ Assertions.assertEquals(runtimeJobTemplateJson,
jobEntity.runtimeJobTemplate());
+ }
+
+ @Test
+ public void testEqualsAndHashCodeIncludeRuntimeJobTemplate() {
+ JobEntity withoutTemplate =
+ 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 sameWithoutTemplate =
+ 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(withoutTemplate, sameWithoutTemplate);
+ Assertions.assertEquals(withoutTemplate.hashCode(),
sameWithoutTemplate.hashCode());
+
+ // Same identity/status/audit but a resolved runtime template must not
compare equal.
+ JobEntity withTemplate =
+ 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)
+
.withRuntimeJobTemplate("{\"jobType\":\"shell\",\"name\":\"test-job-template\"}")
+ .build();
+
+ Assertions.assertNotEquals(withoutTemplate, withTemplate);
+ Assertions.assertNotEquals(withoutTemplate.hashCode(),
withTemplate.hashCode());
+ }
}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestJobMetaBaseSQLProvider.java
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestJobMetaBaseSQLProvider.java
new file mode 100644
index 0000000000..3a332141c7
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestJobMetaBaseSQLProvider.java
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.storage.relational.mapper.provider.base;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestJobMetaBaseSQLProvider {
+
+ private final JobMetaBaseSQLProvider provider = new JobMetaBaseSQLProvider();
+
+ @Test
+ void testInsertJobMetaIncludesRuntimeJobTemplateColumn() {
+ String sql = provider.insertJobMeta(null);
+
+ Assertions.assertTrue(
+ sql.contains("runtime_job_template"),
+ () -> "Column list must include runtime_job_template, but got: " +
sql);
+ Assertions.assertTrue(
+ sql.contains("#{jobMeta.runtimeJobTemplate}"),
+ () -> "VALUES clause must bind runtimeJobTemplate, but got: " + sql);
+ }
+
+ @Test
+ void testInsertJobMetaOnDuplicateKeyUpdateIncludesRuntimeJobTemplateColumn()
{
+ String sql = provider.insertJobMetaOnDuplicateKeyUpdate(null);
+ String onDuplicateClause = sql.substring(sql.indexOf("ON DUPLICATE KEY
UPDATE"));
+
+ Assertions.assertTrue(
+ sql.substring(0,
sql.indexOf("VALUES")).contains("runtime_job_template"),
+ () -> "Column list must include runtime_job_template, but got: " +
sql);
+ Assertions.assertTrue(
+ sql.contains("#{jobMeta.runtimeJobTemplate}"),
+ () -> "VALUES clause must bind runtimeJobTemplate, but got: " + sql);
+ Assertions.assertTrue(
+ onDuplicateClause.contains("runtime_job_template =
#{jobMeta.runtimeJobTemplate}"),
+ () ->
+ "ON DUPLICATE KEY UPDATE must overwrite runtime_job_template, but
got: "
+ + onDuplicateClause);
+ }
+
+ @Test
+ void testListJobPOsByMetalakeSelectsRuntimeJobTemplate() {
+ assertSelectsRuntimeJobTemplate(provider.listJobPOsByMetalake("metalake"));
+ }
+
+ @Test
+ void testListJobPOsByMetalakeAndTemplateSelectsRuntimeJobTemplate() {
+ assertSelectsRuntimeJobTemplate(
+ provider.listJobPOsByMetalakeAndTemplate("metalake", "template"));
+ }
+
+ @Test
+ void testSelectJobPOByMetalakeAndRunIdSelectsRuntimeJobTemplate() {
+
assertSelectsRuntimeJobTemplate(provider.selectJobPOByMetalakeAndRunId("metalake",
1L));
+ }
+
+ @Test
+ void testBatchSelectJobByRunIdsSelectsRuntimeJobTemplate() {
+
assertSelectsRuntimeJobTemplate(provider.batchSelectJobByRunIds("metalake",
null));
+ }
+
+ private void assertSelectsRuntimeJobTemplate(String sql) {
+ Assertions.assertTrue(
+ sql.contains("jrm.runtime_job_template AS runtimeJobTemplate"),
+ () -> "SELECT list must project runtime_job_template, but got: " +
sql);
+ }
+}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TestJobMetaPostgreSQLProvider.java
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TestJobMetaPostgreSQLProvider.java
new file mode 100644
index 0000000000..8b36e827f0
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TestJobMetaPostgreSQLProvider.java
@@ -0,0 +1,43 @@
+/*
+ * 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.storage.relational.mapper.provider.postgresql;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestJobMetaPostgreSQLProvider {
+
+ @Test
+ void testInsertJobMetaOnDuplicateKeyUpdateIncludesRuntimeJobTemplateColumn()
{
+ String sql = new
JobMetaPostgreSQLProvider().insertJobMetaOnDuplicateKeyUpdate(null);
+ String conflictClause = sql.substring(sql.indexOf("ON CONFLICT"));
+
+ // PostgreSQL's ON CONFLICT rewrite (as opposed to MySQL/H2's ON DUPLICATE
KEY UPDATE) is a
+ // separate override from the base provider, so the new column has to be
added here too.
+ Assertions.assertTrue(
+ sql.substring(0,
sql.indexOf("VALUES")).contains("runtime_job_template"),
+ () -> "Column list must include runtime_job_template, but got: " +
sql);
+ Assertions.assertTrue(
+ sql.contains("#{jobMeta.runtimeJobTemplate}"),
+ () -> "VALUES clause must bind runtimeJobTemplate, but got: " + sql);
+ Assertions.assertTrue(
+ conflictClause.contains("runtime_job_template =
#{jobMeta.runtimeJobTemplate}"),
+ () -> "ON CONFLICT DO UPDATE SET must overwrite runtime_job_template,
but got: " + sql);
+ }
+}
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 9ad14ed737..bc3d4cd9d6 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
@@ -201,8 +201,64 @@ public class TestJobPO {
Assertions.assertEquals(finishedAt, resultEntity.finishedAt());
}
+ @Test
+ public void testJobPORuntimeJobTemplateDefaultsToNull() {
+ // runtimeJobTemplate is optional, unlike startedAt/finishedAt - jobs run
before this field
+ // was introduced have no resolved template to backfill, so it must
round-trip as null when
+ // not set, without JobPO's constructor rejecting it.
+ 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(Instant.now()).build())
+ .withStartedAt(0L)
+ .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.assertNull(jobPO.runtimeJobTemplate());
+ Assertions.assertNull(resultEntity.runtimeJobTemplate());
+ }
+
+ @Test
+ public void testJobPORuntimeJobTemplate() {
+ // initializeJobPO must trust the resolved template already set on the
entity by the caller
+ // (JobManager, at runJob time), and fromJobPO must round-trip it
unchanged.
+ String runtimeJobTemplateJson =
+
"{\"jobType\":\"shell\",\"name\":\"test-job-template\",\"executable\":\"/bin/echo\"}";
+ 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(Instant.now()).build())
+ .withStartedAt(0L)
+ .withFinishedAt(0L)
+ .withRuntimeJobTemplate(runtimeJobTemplateJson)
+ .build();
+
+ JobPO.JobPOBuilder builder = JobPO.builder().withMetalakeId(1L);
+ JobPO jobPO = JobPO.initializeJobPO(jobEntity, builder);
+ JobEntity resultEntity = JobPO.fromJobPO(jobPO,
NamespaceUtil.ofJob("test"));
+
+ Assertions.assertEquals(runtimeJobTemplateJson,
jobPO.runtimeJobTemplate());
+ Assertions.assertEquals(runtimeJobTemplateJson,
resultEntity.runtimeJobTemplate());
+ }
+
@Test
public void testUpdateJobPO() {
+ String originalRuntimeJobTemplateJson =
+
"{\"jobType\":\"shell\",\"name\":\"test-job-template\",\"executable\":\"/bin/echo\"}";
JobEntity oldEntity =
JobEntity.builder()
.withId(1L)
@@ -214,6 +270,7 @@ public class TestJobPO {
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
.withStartedAt(0L)
.withFinishedAt(0L)
+ .withRuntimeJobTemplate(originalRuntimeJobTemplateJson)
.build();
JobPO.JobPOBuilder initBuilder = JobPO.builder().withMetalakeId(1L);
@@ -229,7 +286,13 @@ public class TestJobPO {
.build();
// Deliberately try to change jobTemplateName - updateJobPO must ignore it
and keep the old
- // PO's value, since a job's template is immutable once the job is created.
+ // PO's value, since a job's template is immutable once the job is
created. By contrast,
+ // deliberately set a *different* runtimeJobTemplate - unlike
jobTemplateName, updateJobPO
+ // trusts whatever the caller passes here; it is JobManager's updater
functions, not this
+ // storage-layer method, that are responsible for keeping it unchanged in
practice.
+ String newRuntimeJobTemplateJson =
+
"{\"jobType\":\"shell\",\"name\":\"test-job-template\",\"executable\":\"/bin/echo\","
+ + "\"arguments\":[\"resolved\"]}";
JobEntity newEntity =
JobEntity.builder()
.withId(oldEntity.id())
@@ -240,6 +303,7 @@ public class TestJobPO {
.withAuditInfo(updatedAuditInfo)
.withStartedAt(startedAt)
.withFinishedAt(0L)
+ .withRuntimeJobTemplate(newRuntimeJobTemplateJson)
.build();
JobPO.JobPOBuilder updateBuilder =
JobPO.builder().withMetalakeId(oldJobPO.metalakeId());
@@ -251,6 +315,7 @@ public class TestJobPO {
Assertions.assertEquals(JobHandle.Status.STARTED.name(),
newJobPO.jobRunStatus());
Assertions.assertEquals(startedAt, newJobPO.jobStartedAt());
Assertions.assertEquals(0L, newJobPO.jobFinishedAt());
+ Assertions.assertEquals(newRuntimeJobTemplateJson,
newJobPO.runtimeJobTemplate());
Assertions.assertEquals(oldJobPO.currentVersion() + 1,
newJobPO.currentVersion());
Assertions.assertEquals(oldJobPO.lastVersion() + 1,
newJobPO.lastVersion());
@@ -258,6 +323,7 @@ public class TestJobPO {
Assertions.assertEquals(JobHandle.Status.STARTED, resultEntity.status());
Assertions.assertEquals(startedAt, resultEntity.startedAt());
Assertions.assertEquals("test-job-template",
resultEntity.jobTemplateName());
+ Assertions.assertEquals(newRuntimeJobTemplateJson,
resultEntity.runtimeJobTemplate());
Assertions.assertEquals(
updatedAuditInfo.lastModifier(),
resultEntity.auditInfo().lastModifier());
}
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 fd17917d93..9ddbfe51c4 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
@@ -314,6 +314,137 @@ public class TestJobMetaService extends TestJDBCBackend {
Assertions.assertEquals(finishedJob, fetchedJob);
}
+ @TestTemplate
+ public void testInsertAndGetJobWithRuntimeJobTemplate() throws IOException {
+ BaseMetalake metalake =
+ createBaseMakeLake(RandomIdGenerator.INSTANCE.nextId(), METALAKE_NAME,
AUDIT_INFO);
+ backend.insert(metalake, false);
+
+ JobTemplateEntity jobTemplate =
+ TestJobTemplateMetaService.newShellJobTemplateEntity(
+ "test_job_template", "test_comment", METALAKE_NAME);
+ JobTemplateMetaService.getInstance().insertJobTemplate(jobTemplate, false);
+
+ String runtimeJobTemplateJson =
+
"{\"jobType\":\"shell\",\"name\":\"test_job_template\",\"executable\":\"/bin/echo\"}";
+ JobEntity job =
+ JobEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withJobExecutionId("job-execution-runtime-template")
+ .withJobTemplateName(jobTemplate.name())
+ .withStatus(JobHandle.Status.QUEUED)
+ .withNamespace(NamespaceUtil.ofJob(METALAKE_NAME))
+ .withAuditInfo(AUDIT_INFO)
+ .withStartedAt(0L)
+ .withFinishedAt(0L)
+ .withRuntimeJobTemplate(runtimeJobTemplateJson)
+ .build();
+ Assertions.assertDoesNotThrow(() ->
JobMetaService.getInstance().insertJob(job, false));
+
+ JobEntity retrievedJob =
+ JobMetaService.getInstance()
+ .getJobByIdentifier(NameIdentifierUtil.ofJob(METALAKE_NAME,
job.name()));
+ Assertions.assertEquals(runtimeJobTemplateJson,
retrievedJob.runtimeJobTemplate());
+ Assertions.assertEquals(job, retrievedJob);
+
+ // A job inserted without a runtime job template (e.g. a row from before
this field existed)
+ // must round-trip as null rather than failing to insert/select.
+ JobEntity jobWithoutTemplate =
+ TestJobTemplateMetaService.newJobEntity(
+ jobTemplate.name(), JobHandle.Status.QUEUED, METALAKE_NAME);
+ Assertions.assertDoesNotThrow(
+ () -> JobMetaService.getInstance().insertJob(jobWithoutTemplate,
false));
+ JobEntity retrievedJobWithoutTemplate =
+ JobMetaService.getInstance()
+ .getJobByIdentifier(NameIdentifierUtil.ofJob(METALAKE_NAME,
jobWithoutTemplate.name()));
+ Assertions.assertNull(retrievedJobWithoutTemplate.runtimeJobTemplate());
+
+ // Overwriting a job must also overwrite its stored runtime job template.
+ String updatedRuntimeJobTemplateJson =
+
"{\"jobType\":\"shell\",\"name\":\"test_job_template\",\"executable\":\"/bin/echo\","
+ + "\"arguments\":[\"resolved\"]}";
+ JobEntity jobOverwrite =
+ JobEntity.builder()
+ .withId(job.id())
+ .withJobExecutionId(job.jobExecutionId())
+ .withStatus(JobHandle.Status.STARTED)
+ .withNamespace(job.namespace())
+ .withAuditInfo(job.auditInfo())
+ .withJobTemplateName(job.jobTemplateName())
+ .withStartedAt(System.currentTimeMillis())
+ .withFinishedAt(0L)
+ .withRuntimeJobTemplate(updatedRuntimeJobTemplateJson)
+ .build();
+ Assertions.assertDoesNotThrow(() ->
JobMetaService.getInstance().insertJob(jobOverwrite, true));
+
+ JobEntity retrievedOverwrittenJob =
+ JobMetaService.getInstance()
+ .getJobByIdentifier(NameIdentifierUtil.ofJob(METALAKE_NAME,
jobOverwrite.name()));
+ Assertions.assertEquals(
+ updatedRuntimeJobTemplateJson,
retrievedOverwrittenJob.runtimeJobTemplate());
+ }
+
+ @TestTemplate
+ public void testUpdateJobPersistsRuntimeJobTemplateChange() throws
IOException {
+ // Unlike jobTemplateName, this storage layer does not enforce that the
runtime job template
+ // never changes - it persists whatever the updater lambda's returned
entity says, the same
+ // way it persists status/timestamps/audit info. Keeping the resolved
template unchanged
+ // across status transitions is JobManager's responsibility (its updater
functions always
+ // carry the existing value forward), not something guarded here.
+ BaseMetalake metalake =
+ createBaseMakeLake(RandomIdGenerator.INSTANCE.nextId(), METALAKE_NAME,
AUDIT_INFO);
+ backend.insert(metalake, false);
+
+ JobTemplateEntity jobTemplate =
+ TestJobTemplateMetaService.newShellJobTemplateEntity(
+ "test_job_template", "test_comment", METALAKE_NAME);
+ JobTemplateMetaService.getInstance().insertJobTemplate(jobTemplate, false);
+
+ String originalRuntimeJobTemplateJson =
+
"{\"jobType\":\"shell\",\"name\":\"test_job_template\",\"executable\":\"/bin/echo\"}";
+ JobEntity job =
+ JobEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withJobExecutionId("job-execution-update-changes-template")
+ .withJobTemplateName(jobTemplate.name())
+ .withStatus(JobHandle.Status.QUEUED)
+ .withNamespace(NamespaceUtil.ofJob(METALAKE_NAME))
+ .withAuditInfo(AUDIT_INFO)
+ .withStartedAt(0L)
+ .withFinishedAt(0L)
+ .withRuntimeJobTemplate(originalRuntimeJobTemplateJson)
+ .build();
+ JobMetaService.getInstance().insertJob(job, false);
+
+ String newRuntimeJobTemplateJson =
+
"{\"jobType\":\"shell\",\"name\":\"test_job_template\",\"executable\":\"/bin/echo\","
+ + "\"arguments\":[\"resolved\"]}";
+ JobEntity updatedJob =
+ JobMetaService.getInstance()
+ .updateJob(
+ NameIdentifierUtil.ofJob(METALAKE_NAME, job.name()),
+ (JobEntity oldJob) ->
+ JobEntity.builder()
+ .withId(oldJob.id())
+ .withJobExecutionId(oldJob.jobExecutionId())
+ .withJobTemplateName(oldJob.jobTemplateName())
+ .withNamespace(oldJob.namespace())
+ .withStatus(JobHandle.Status.STARTED)
+ .withAuditInfo(oldJob.auditInfo())
+ .withStartedAt(System.currentTimeMillis())
+ .withFinishedAt(oldJob.finishedAt())
+ .withRuntimeJobTemplate(newRuntimeJobTemplateJson)
+ .build());
+ Assertions.assertEquals(newRuntimeJobTemplateJson,
updatedJob.runtimeJobTemplate());
+
+ // The change must actually be persisted, not just returned.
+ JobEntity persistedJob =
+ JobMetaService.getInstance()
+ .getJobByIdentifier(NameIdentifierUtil.ofJob(METALAKE_NAME,
job.name()));
+ Assertions.assertEquals(JobHandle.Status.STARTED, persistedJob.status());
+ Assertions.assertEquals(newRuntimeJobTemplateJson,
persistedJob.runtimeJobTemplate());
+ }
+
@TestTemplate
public void testUpdateNonExistentJobThrowsNoSuchEntityException() throws
IOException {
BaseMetalake metalake =
diff --git a/docs/open-api/jobs.yaml b/docs/open-api/jobs.yaml
index ab4f692f0b..7b7a75d6ab 100644
--- a/docs/open-api/jobs.yaml
+++ b/docs/open-api/jobs.yaml
@@ -586,6 +586,12 @@ components:
format: date-time
nullable: true
description: The time when the job finished execution, or null if
the job has not finished execution yet
+ runtimeJobTemplate:
+ description: The resolved job template that was actually submitted
for execution, with placeholders replaced and referenced files downloaded. Null
for jobs run before this field was introduced, or if the stored value could not
be read back
+ type: object
+ nullable: true
+ allOf:
+ - $ref: "#/components/schemas/JobTemplate"
TemplateUpdate:
@@ -1005,7 +1011,23 @@ components:
},
"queuedAt": "2025-08-12T02:14:28.205023Z",
"startedAt": "2025-08-12T02:14:31.098231Z",
- "finishedAt": "2025-08-12T02:15:03.512847Z"
+ "finishedAt": "2025-08-12T02:15:03.512847Z",
+ "runtimeJobTemplate": {
+ "name": "test_run_get",
+ "jobType": "shell",
+ "comment": "Test shell job template",
+ "executable":
"/tmp/gravitino-job-staging/job-12345/test-job.sh",
+ "arguments": ["value1", "value2"],
+ "environments": { },
+ "customFields": { },
+ "scripts": [
+ "/tmp/gravitino-job-staging/job-12345/common.sh"
+ ],
+ "audit": {
+ "createTime": "2025-08-12T02:00:00.000000Z",
+ "creator": "anonymous"
+ }
+ }
},
{
"jobId": "job-67890",
@@ -1043,7 +1065,23 @@ components:
},
"queuedAt": "2025-08-12T02:14:28.205023Z",
"startedAt": "2025-08-12T02:14:31.098231Z",
- "finishedAt": "2025-08-12T02:15:03.512847Z"
+ "finishedAt": "2025-08-12T02:15:03.512847Z",
+ "runtimeJobTemplate": {
+ "name": "test_run_get",
+ "jobType": "shell",
+ "comment": "Test shell job template",
+ "executable": "/tmp/gravitino-job-staging/job-12345/test-job.sh",
+ "arguments": ["value1", "value2"],
+ "environments": { },
+ "customFields": { },
+ "scripts": [
+ "/tmp/gravitino-job-staging/job-12345/common.sh"
+ ],
+ "audit": {
+ "createTime": "2025-08-12T02:00:00.000000Z",
+ "creator": "anonymous"
+ }
+ }
}
}
diff --git a/scripts/h2/schema-2.0.0-h2.sql b/scripts/h2/schema-2.0.0-h2.sql
index 2ac5e66998..d5a4e5baf6 100644
--- a/scripts/h2/schema-2.0.0-h2.sql
+++ b/scripts/h2/schema-2.0.0-h2.sql
@@ -487,6 +487,7 @@ CREATE TABLE IF NOT EXISTS `job_run_meta` (
`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',
+ `runtime_job_template` CLOB DEFAULT NULL COMMENT 'job run runtime job
template',
`audit_info` CLOB NOT NULL COMMENT 'job run audit info',
`current_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'job run current
version',
`last_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'job run last
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 8175e0c630..e3238121b7 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
@@ -58,6 +58,8 @@ CREATE TABLE IF NOT EXISTS `policy_tag_relation_meta` (
KEY `policy_tag_relation_meta_idx_tag_id` (`tag_id`)
) ENGINE=InnoDB;
+ALTER TABLE `job_run_meta` ADD COLUMN `runtime_job_template` CLOB DEFAULT NULL
COMMENT 'job run runtime job template' AFTER `job_finished_at`;
+
CREATE TABLE IF NOT EXISTS `semantic_model_meta` (
`semantic_model_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'semantic model
id',
`semantic_model_name` VARCHAR(128) NOT NULL COMMENT 'semantic model name',
diff --git a/scripts/mysql/schema-2.0.0-mysql.sql
b/scripts/mysql/schema-2.0.0-mysql.sql
index b8042ef534..c51e99f937 100644
--- a/scripts/mysql/schema-2.0.0-mysql.sql
+++ b/scripts/mysql/schema-2.0.0-mysql.sql
@@ -478,6 +478,7 @@ CREATE TABLE IF NOT EXISTS `job_run_meta` (
`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',
+ `runtime_job_template` MEDIUMTEXT DEFAULT NULL COMMENT 'job run runtime
job template',
`audit_info` MEDIUMTEXT NOT NULL COMMENT 'job run audit info',
`current_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'job run current
version',
`last_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'job run last
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 67fba282a2..7342a0905d 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
@@ -118,6 +118,9 @@ CREATE TABLE IF NOT EXISTS `policy_tag_relation_meta` (
KEY `policy_tag_relation_meta_idx_tag_id` (`tag_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'policy
tag relation';
+ALTER TABLE `job_run_meta`
+ ADD COLUMN `runtime_job_template` MEDIUMTEXT DEFAULT NULL COMMENT 'job run
runtime job template' AFTER `job_finished_at`;
+
CREATE TABLE IF NOT EXISTS `semantic_model_meta` (
`semantic_model_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'semantic model
id',
`semantic_model_name` VARCHAR(128) NOT NULL COMMENT 'semantic model name',
diff --git a/scripts/postgresql/schema-2.0.0-postgresql.sql
b/scripts/postgresql/schema-2.0.0-postgresql.sql
index 2461b957af..9c8da3beee 100644
--- a/scripts/postgresql/schema-2.0.0-postgresql.sql
+++ b/scripts/postgresql/schema-2.0.0-postgresql.sql
@@ -843,6 +843,7 @@ CREATE TABLE IF NOT EXISTS job_run_meta (
job_run_status VARCHAR(64) NOT NULL,
job_started_at BIGINT NOT NULL DEFAULT 0,
job_finished_at BIGINT NOT NULL DEFAULT 0,
+ runtime_job_template TEXT DEFAULT NULL,
audit_info TEXT NOT NULL,
current_version INT NOT NULL DEFAULT 1,
last_version INT NOT NULL DEFAULT 1,
@@ -861,6 +862,7 @@ 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.runtime_job_template IS 'job run runtime job
template';
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';
COMMENT ON COLUMN job_run_meta.last_version IS 'job run last 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 7a0a12d2b5..79c1a4f032 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
@@ -78,6 +78,9 @@ COMMENT ON COLUMN policy_tag_relation_meta.current_version IS
'policy tag relati
COMMENT ON COLUMN policy_tag_relation_meta.last_version IS 'policy tag
relation last version';
COMMENT ON COLUMN policy_tag_relation_meta.deleted_at IS 'policy tag relation
deleted at';
+ALTER TABLE job_run_meta ADD COLUMN IF NOT EXISTS runtime_job_template TEXT
DEFAULT NULL;
+COMMENT ON COLUMN job_run_meta.runtime_job_template IS 'job run runtime job
template';
+
CREATE TABLE IF NOT EXISTS semantic_model_meta (
semantic_model_id BIGINT NOT NULL,
semantic_model_name VARCHAR(128) NOT NULL,
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 0df9b9bf74..ded98a34b4 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
@@ -525,7 +525,29 @@ public class JobOperations {
DTOConverters.toDTO(jobEntity.auditInfo()),
jobEntity.auditInfo().createTime(),
jobEntity.startedAtAsInstant(),
- jobEntity.finishedAtAsInstant());
+ jobEntity.finishedAtAsInstant(),
+ toRuntimeJobTemplateDTO(jobEntity));
+ }
+
+ /**
+ * Deserializes the job entity's stored runtime job template JSON, if any. A
malformed or
+ * forward-incompatible stored value (e.g. a job type unknown to this server
version) must not
+ * make the job unreadable or uncancellable through the API - failures here
are logged and
+ * swallowed rather than propagated, so callers of {@link #toDTO(JobEntity)}
(get/run/cancel/list
+ * job) always get a usable response with just the runtime job template
omitted.
+ */
+ private static JobTemplateDTO toRuntimeJobTemplateDTO(JobEntity jobEntity) {
+ try {
+ return DTOConverters.fromRuntimeJobTemplateJson(
+ jobEntity.runtimeJobTemplate(), jobEntity.name());
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to deserialize the runtime job template for job {}, omitting
it from the "
+ + "response",
+ jobEntity.name(),
+ e);
+ return null;
+ }
}
private static List<JobDTO> toJobDTOs(List<JobEntity> jobEntities) {
diff --git
a/server/src/test/java/org/apache/gravitino/server/web/rest/TestJobOperations.java
b/server/src/test/java/org/apache/gravitino/server/web/rest/TestJobOperations.java
index fe19825831..e39cafa70c 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
@@ -47,6 +47,7 @@ import org.apache.gravitino.Config;
import org.apache.gravitino.GravitinoEnv;
import org.apache.gravitino.dto.job.JobDTO;
import org.apache.gravitino.dto.job.JobTemplateDTO;
+import org.apache.gravitino.dto.job.ShellJobTemplateDTO;
import org.apache.gravitino.dto.job.ShellTemplateUpdateDTO;
import org.apache.gravitino.dto.requests.JobRunRequest;
import org.apache.gravitino.dto.requests.JobTemplateRegisterRequest;
@@ -1162,6 +1163,42 @@ public class TestJobOperations extends JerseyTest {
Assertions.assertEquals(NoSuchJobException.class.getSimpleName(),
errorResp.getType());
}
+ @Test
+ public void testCancelJobWithMalformedRuntimeJobTemplateDoesNotFail() {
+ // By the time toDTO() runs here, jobOperationDispatcher.cancelJob() has
already cancelled
+ // the job and updated its stored entity - a malformed stored runtime job
template must not
+ // turn that already-completed cancellation into a 500 for the caller. The
response should
+ // just omit the runtime job template.
+ JobEntity job =
+ JobEntity.builder()
+ .withId(new Random().nextLong())
+ .withJobExecutionId("job-execution-cancel-malformed")
+ .withNamespace(NamespaceUtil.ofJob(metalake))
+ .withJobTemplateName("shell_template_1")
+ .withStatus(JobHandle.Status.CANCELLED)
+ .withStartedAt(0L)
+ .withFinishedAt(Instant.now().toEpochMilli())
+ .withRuntimeJobTemplate("{not-valid-json")
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .build();
+
+ when(jobOperationDispatcher.cancelJob(metalake,
job.name())).thenReturn(job);
+
+ Response resp =
+ target(jobRunPath())
+ .path(job.name())
+ .request(APPLICATION_JSON_TYPE)
+ .accept("application/vnd.gravitino.v1+json")
+ .post(null);
+
+ Assertions.assertEquals(Response.Status.OK.getStatusCode(),
resp.getStatus());
+ JobResponse jobResp = resp.readEntity(JobResponse.class);
+ Assertions.assertEquals(0, jobResp.getCode());
+ Assertions.assertEquals(JobHandle.Status.CANCELLED,
jobResp.getJob().status());
+ Assertions.assertNull(jobResp.getJob().runtimeJobTemplate());
+ }
+
@Test
public void testToDTOFinishedAt() {
// Sentinel value (<= 0) used by the storage layer means "not finished".
@@ -1201,6 +1238,95 @@ public class TestJobOperations extends JerseyTest {
Assertions.assertNotNull(jobDTO.queuedAt());
}
+ @Test
+ public void testToDTORuntimeJobTemplate() {
+ // No runtime job template stored (e.g. a job run before this field was
introduced) - must
+ // round-trip as null rather than failing to convert.
+ JobEntity jobWithoutTemplate = newJobEntity("shell_template_1",
JobHandle.Status.QUEUED);
+ JobDTO jobDTOWithoutTemplate = JobOperations.toDTO(jobWithoutTemplate);
+ Assertions.assertNull(jobDTOWithoutTemplate.runtimeJobTemplate());
+
+ // A stored runtime job template must be deserialized back into a
JobTemplateDTO, with
+ // Shell/Spark dispatch handled automatically by JobTemplateDTO's
@JsonTypeInfo.
+ String runtimeJobTemplateJson =
+
"{\"jobType\":\"shell\",\"name\":\"shell_template_1\",\"comment\":\"resolved\","
+ + "\"executable\":\"/bin/echo\",\"arguments\":[\"resolved-arg\"]}";
+ JobEntity jobWithTemplate =
+ JobEntity.builder()
+ .withId(new Random().nextLong())
+ .withJobExecutionId("job-execution-with-template")
+ .withNamespace(NamespaceUtil.ofJob(metalake))
+ .withJobTemplateName("shell_template_1")
+ .withStatus(JobHandle.Status.QUEUED)
+ .withStartedAt(0L)
+ .withFinishedAt(0L)
+ .withRuntimeJobTemplate(runtimeJobTemplateJson)
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .build();
+
+ JobDTO jobDTO = JobOperations.toDTO(jobWithTemplate);
+
+ Assertions.assertNotNull(jobDTO.runtimeJobTemplate());
+ Assertions.assertInstanceOf(ShellJobTemplateDTO.class,
jobDTO.runtimeJobTemplate());
+ ShellJobTemplateDTO runtimeJobTemplateDTO = (ShellJobTemplateDTO)
jobDTO.runtimeJobTemplate();
+ Assertions.assertEquals("shell_template_1", runtimeJobTemplateDTO.name());
+ Assertions.assertEquals("resolved", runtimeJobTemplateDTO.comment());
+ Assertions.assertEquals("/bin/echo", runtimeJobTemplateDTO.executable());
+ Assertions.assertEquals(Lists.newArrayList("resolved-arg"),
runtimeJobTemplateDTO.arguments());
+ }
+
+ @Test
+ public void
testListJobsWithMalformedRuntimeJobTemplateDoesNotFailWholeList() {
+ // A single job whose stored runtime job template fails to deserialize
(e.g. corrupted or
+ // written by a future, incompatible version) must not fail the entire
listJobs response -
+ // it should come back with a null runtimeJobTemplate while every other
job is unaffected.
+ String templateName = "shell_template_1";
+ JobEntity healthyJob = newJobEntity(templateName, JobHandle.Status.QUEUED);
+ JobEntity malformedJob =
+ JobEntity.builder()
+ .withId(new Random().nextLong())
+ .withJobExecutionId("job-execution-malformed")
+ .withNamespace(NamespaceUtil.ofJob(metalake))
+ .withJobTemplateName(templateName)
+ .withStatus(JobHandle.Status.QUEUED)
+ .withStartedAt(0L)
+ .withFinishedAt(0L)
+ .withRuntimeJobTemplate("{not-valid-json")
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .build();
+
+ when(jobOperationDispatcher.listJobs(metalake, Optional.empty()))
+ .thenReturn(Lists.newArrayList(healthyJob, malformedJob));
+
+ Response resp =
+ target(jobRunPath())
+ .request(APPLICATION_JSON_TYPE)
+ .accept("application/vnd.gravitino.v1+json")
+ .get();
+
+ Assertions.assertEquals(Response.Status.OK.getStatusCode(),
resp.getStatus());
+ JobListResponse jobListResponse = resp.readEntity(JobListResponse.class);
+ Assertions.assertEquals(0, jobListResponse.getCode());
+ Assertions.assertEquals(2, jobListResponse.getJobs().size());
+
+ JobDTO healthyJobDTO =
+ jobListResponse.getJobs().stream()
+ .filter(dto -> dto.jobId().equals(healthyJob.name()))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("Healthy job missing from
response"));
+ Assertions.assertEquals(JobOperations.toDTO(healthyJob), healthyJobDTO);
+
+ JobDTO malformedJobDTO =
+ jobListResponse.getJobs().stream()
+ .filter(dto -> dto.jobId().equals(malformedJob.name()))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("Malformed job missing from
response"));
+ Assertions.assertNull(malformedJobDTO.runtimeJobTemplate());
+ Assertions.assertEquals(JobHandle.Status.QUEUED, malformedJobDTO.status());
+ }
+
private String jobTemplatePath() {
return "/metalakes/" + metalake + "/jobs/templates";
}
