This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 0c6c9c29f2 [#12521] feat(job): Support filtering and sorting jobs by
time in listJobs (#12522)
0c6c9c29f2 is described below
commit 0c6c9c29f2fa16baba0c98c8463d0d83fcbf35e4
Author: Jerry Shao <[email protected]>
AuthorDate: Thu Aug 20 22:24:33 2026 +0800
[#12521] feat(job): Support filtering and sorting jobs by time in listJobs
(#12522)
### What changes were proposed in this pull request?
Add `queuedAfter`/`startedAfter`/`finishedAfter`/`sortBy`/`sortOrder`
query parameters to the list jobs REST API (`GET
/metalakes/{metalake}/jobs/runs`):
- `queuedAfter`/`startedAfter`/`finishedAfter` — ISO-8601 instant
strings, inclusive lower bounds, AND-combined; a job missing the
relevant timestamp is excluded by that filter.
- `sortBy` (`queuedAt`|`startedAt`|`finishedAt`, default `queuedAt`) /
`sortOrder` (`asc`|`desc`, default `desc`) — jobs missing the sort field
always sort last, regardless of direction.
- Implemented entirely in `JobOperations` (server REST layer):
filtering/sorting applied in-memory after the existing
`jobOperationDispatcher.listJobs(...)` fetch and authorization filter —
no dispatcher/manager/storage/SQL changes.
- All query params are validated up front, before the dispatcher fetch,
so invalid input fails fast with 400; `sortBy`/`sortOrder` are matched
exactly against the OpenAPI-documented casing.
- OpenAPI spec (`docs/open-api/jobs.yaml`) updated with the five new
parameters.
### Why are the changes needed?
Callers currently have to fetch the entire job list and filter/sort
client-side to answer time-scoped questions like "what ran in the last
24 hours," which doesn't scale as job history grows.
Fix: #12521
### Does this PR introduce _any_ user-facing change?
Yes: new
`queuedAfter`/`startedAfter`/`finishedAfter`/`sortBy`/`sortOrder` query
parameters on the `GET /metalakes/{metalake}/jobs/runs` REST endpoint.
No changes to existing fields or client APIs.
### How was this patch tested?
Unit tests in `TestJobOperations` covering: each filter individually and
AND-combined, inclusive boundary matching, sort by each field in both
directions with null-handling (unset timestamps sort last),
`sortBy`/`sortOrder` validation (invalid values, wrong casing), invalid
ISO-8601 timestamps, and `jobTemplateName` combined with time
filters/sort together.
*Note: this branch is stacked on `feat/job-queued-started-at` (#12509),
which is still open — the diff below will include those commits until
#12509 merges into main.*
---------
Co-authored-by: Claude Sonnet 5 <[email protected]>
---
.../apache/gravitino/client/TestSupportsJobs.java | 23 +-
.../gravitino/dto/responses/JobListResponse.java | 24 +-
docs/open-api/jobs.yaml | 79 ++++-
.../gravitino/server/web/rest/JobOperations.java | 170 ++++++++++-
.../server/web/rest/TestJobOperations.java | 339 ++++++++++++++++++++-
5 files changed, 610 insertions(+), 25 deletions(-)
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 4fa6469fa1..f811728cb0 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
@@ -23,7 +23,9 @@ import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import java.time.Instant;
import java.util.Collections;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
import java.util.stream.Collectors;
import org.apache.gravitino.dto.AuditDTO;
import org.apache.gravitino.dto.job.JobDTO;
@@ -243,7 +245,7 @@ public class TestSupportsJobs extends TestBase {
List<JobDTO> jobs =
Lists.newArrayList(newJobDTO(jobId1, jobTemplateName),
newJobDTO(jobId2, jobTemplateName));
- JobListResponse resp = new JobListResponse(jobs);
+ JobListResponse resp = new JobListResponse(jobs, ImmutableMap.of());
buildMockResource(Method.GET, jobRunsPath(), null, resp, HttpStatus.SC_OK);
@@ -274,6 +276,25 @@ public class TestSupportsJobs extends TestBase {
compare(jobs.get(1), jobsByTemplate.get(1));
}
+ @Test
+ public void testListJobsAgainstServerWithoutStatusCounts() throws
JsonProcessingException {
+ // Simulate an older server whose response predates the statusCounts field
entirely (the key
+ // is absent, not just null) - a new client must still be able to parse it
without
+ // JobListResponse.validate() failing.
+ String jobTemplateName = "shell-job-template";
+ JobDTO job = newJobDTO("job-1", jobTemplateName);
+
+ Map<String, Object> legacyRespBody = new LinkedHashMap<>();
+ legacyRespBody.put("code", 0);
+ legacyRespBody.put("jobs", Lists.newArrayList(job));
+
+ buildMockResource(Method.GET, jobRunsPath(), null, legacyRespBody,
HttpStatus.SC_OK);
+
+ List<JobHandle> actualJobs = metalake.listJobs();
+ Assertions.assertEquals(1, actualJobs.size());
+ compare(job, actualJobs.get(0));
+ }
+
@Test
public void testGetJob() throws JsonProcessingException {
String jobId = "job-1";
diff --git
a/common/src/main/java/org/apache/gravitino/dto/responses/JobListResponse.java
b/common/src/main/java/org/apache/gravitino/dto/responses/JobListResponse.java
index 50c4c422b5..af1902083f 100644
---
a/common/src/main/java/org/apache/gravitino/dto/responses/JobListResponse.java
+++
b/common/src/main/java/org/apache/gravitino/dto/responses/JobListResponse.java
@@ -21,6 +21,7 @@ package org.apache.gravitino.dto.responses;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.base.Preconditions;
import java.util.List;
+import java.util.Map;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import org.apache.gravitino.dto.job.JobDTO;
@@ -33,19 +34,36 @@ public class JobListResponse extends BaseResponse {
@JsonProperty("jobs")
private final List<JobDTO> jobs;
+ @JsonProperty("statusCounts")
+ private final Map<String, Long> statusCounts;
+
/**
- * Creates a new JobListResponse with the specified list of jobs.
+ * Creates a new JobListResponse with the specified list of jobs and no
per-status counts.
*
* @param jobs The list of jobs to include in the response.
*/
public JobListResponse(List<JobDTO> jobs) {
+ this(jobs, null);
+ }
+
+ /**
+ * Creates a new JobListResponse with the specified list of jobs and
per-status counts.
+ *
+ * @param jobs The list of jobs to include in the response.
+ * @param statusCounts The number of jobs in {@code jobs}, keyed by
lower-case status name (e.g.
+ * "queued", "started"), with every {@link
org.apache.gravitino.job.JobHandle.Status} value
+ * present even when its count is zero. May be {@code null} when
deserialized from an older
+ * server that predates this field.
+ */
+ public JobListResponse(List<JobDTO> jobs, Map<String, Long> statusCounts) {
super(0);
this.jobs = jobs;
+ this.statusCounts = statusCounts;
}
/** Default constructor for Jackson deserialization. */
private JobListResponse() {
- this(null);
+ this(null, null);
}
@Override
@@ -54,5 +72,7 @@ public class JobListResponse extends BaseResponse {
Preconditions.checkArgument(jobs != null, "\"jobs\" must not be null");
jobs.forEach(JobDTO::validate);
+ // statusCounts is intentionally not required: an older server that
predates this field
+ // won't include it, and a new client must still be able to talk to it.
}
}
diff --git a/docs/open-api/jobs.yaml b/docs/open-api/jobs.yaml
index 5237d5e958..ab4f692f0b 100644
--- a/docs/open-api/jobs.yaml
+++ b/docs/open-api/jobs.yaml
@@ -195,6 +195,11 @@ paths:
operationId: listJobs
parameters:
- $ref: "#/components/parameters/jobTemplateName"
+ - $ref: "#/components/parameters/queuedAfter"
+ - $ref: "#/components/parameters/startedAfter"
+ - $ref: "#/components/parameters/finishedAfter"
+ - $ref: "#/components/parameters/sortBy"
+ - $ref: "#/components/parameters/sortOrder"
responses:
"200":
description: Returns the list of job objects
@@ -352,6 +357,58 @@ components:
required: true
schema:
type: string
+ queuedAfter:
+ name: queuedAfter
+ in: query
+ description: >-
+ Only return jobs queued at or after this ISO-8601 instant (e.g.
2026-08-18T00:00:00Z)
+ required: false
+ schema:
+ type: string
+ format: date-time
+ startedAfter:
+ name: startedAfter
+ in: query
+ description: >-
+ Only return jobs started at or after this ISO-8601 instant (e.g.
2026-08-18T00:00:00Z).
+ Jobs that have not started yet are excluded
+ required: false
+ schema:
+ type: string
+ format: date-time
+ finishedAfter:
+ name: finishedAfter
+ in: query
+ description: >-
+ Only return jobs finished at or after this ISO-8601 instant (e.g.
2026-08-18T00:00:00Z).
+ Jobs that have not finished yet are excluded
+ required: false
+ schema:
+ type: string
+ format: date-time
+ sortBy:
+ name: sortBy
+ in: query
+ description: The field to sort the returned jobs by
+ required: false
+ schema:
+ type: string
+ enum:
+ - queuedAt
+ - startedAt
+ - finishedAt
+ default: queuedAt
+ sortOrder:
+ name: sortOrder
+ in: query
+ description: The sort order for the returned jobs
+ required: false
+ schema:
+ type: string
+ enum:
+ - asc
+ - desc
+ default: desc
schemas:
@@ -512,7 +569,7 @@ components:
- "failed"
- "succeeded"
- "cancelling"
- - "canceled"
+ - "cancelled"
audit:
$ref: "./openapi.yaml#/components/schemas/Audit"
queuedAt:
@@ -773,6 +830,16 @@ components:
description: A list of job objects
items:
$ref: "#/components/schemas/Job"
+ statusCounts:
+ type: object
+ description: >-
+ The number of jobs in "jobs", keyed by lower-case status name.
Every status is
+ present, even at zero. Counted after
queuedAfter/startedAfter/finishedAfter filtering,
+ so a time filter can make some statuses structurally zero - e.g.
startedAfter always
+ yields queued: 0, since a queued job has no startedAt yet.
+ additionalProperties:
+ type: integer
+ format: int64
JobResponse:
type: object
@@ -952,7 +1019,15 @@ components:
"startedAt": "2025-08-12T02:14:35.442190Z",
"finishedAt": "2025-08-12T02:15:47.891023Z"
}
- ]
+ ],
+ "statusCounts": {
+ "queued": 0,
+ "started": 0,
+ "failed": 1,
+ "succeeded": 1,
+ "cancelling": 0,
+ "cancelled": 0
+ }
}
JobResponse:
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 74d1e0e5fd..55f8b578e8 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
@@ -21,12 +21,19 @@ package org.apache.gravitino.server.web.rest;
import com.codahale.metrics.annotation.ResponseMetered;
import com.codahale.metrics.annotation.Timed;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import java.time.Instant;
+import java.time.OffsetDateTime;
+import java.time.format.DateTimeParseException;
import java.util.Collections;
+import java.util.Comparator;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Optional;
+import java.util.function.Function;
import java.util.stream.Collectors;
import javax.inject.Inject;
import javax.servlet.http.HttpServletRequest;
@@ -59,6 +66,7 @@ import
org.apache.gravitino.dto.responses.JobTemplateListResponse;
import org.apache.gravitino.dto.responses.JobTemplateResponse;
import org.apache.gravitino.dto.responses.NameListResponse;
import org.apache.gravitino.dto.util.DTOConverters;
+import org.apache.gravitino.job.JobHandle;
import org.apache.gravitino.job.JobOperationDispatcher;
import org.apache.gravitino.job.JobTemplateChange;
import org.apache.gravitino.meta.AuditInfo;
@@ -285,13 +293,36 @@ public class JobOperations {
public Response listJobs(
@PathParam("metalake") @AuthorizationMetadata(type =
Entity.EntityType.METALAKE)
String metalake,
- @QueryParam("jobTemplateName") String jobTemplateName) {
+ @QueryParam("jobTemplateName") String jobTemplateName,
+ @QueryParam("queuedAfter") String queuedAfter,
+ @QueryParam("startedAfter") String startedAfter,
+ @QueryParam("finishedAfter") String finishedAfter,
+ @QueryParam("sortBy") @DefaultValue("queuedAt") String sortBy,
+ @QueryParam("sortOrder") @DefaultValue("desc") String sortOrder) {
LOG.info(
- "Received request to list jobs in metalake {}{}",
+ "Received request to list jobs in metalake {}{}, queuedAfter: {},
startedAfter: {},"
+ + " finishedAfter: {}, sortBy: {}, sortOrder: {}",
metalake,
- jobTemplateName != null ? " for job template " + jobTemplateName : "");
+ jobTemplateName != null ? " for job template " + jobTemplateName : "",
+ queuedAfter,
+ startedAfter,
+ finishedAfter,
+ sortBy,
+ sortOrder);
try {
+ // Parse/validate query params up front so a bad request fails fast,
before paying for the
+ // dispatcher fetch and authorization filtering below. @DefaultValue
only applies when a
+ // param is absent, not when a client sends it empty (e.g.
"?sortBy=&sortOrder="), so blank
+ // values are treated as "not set" here too.
+ Instant queuedAfterInstant = parseInstant("queuedAfter", queuedAfter);
+ Instant startedAfterInstant = parseInstant("startedAfter", startedAfter);
+ Instant finishedAfterInstant = parseInstant("finishedAfter",
finishedAfter);
+ Comparator<JobEntity> comparator =
+ buildJobComparator(
+ Strings.isNullOrEmpty(sortBy) ? "queuedAt" : sortBy,
+ Strings.isNullOrEmpty(sortOrder) ? "desc" : sortOrder);
+
return Utils.doAs(
httpRequest,
() -> {
@@ -305,10 +336,18 @@ public class JobOperations {
.listJobs(metalake,
Optional.ofNullable(jobTemplateName))
.toArray(new JobEntity[0]),
jobEntity -> NameIdentifierUtil.ofJob(metalake,
jobEntity.name())));
- List<JobDTO> jobDTOs = toJobDTOs(jobEntities);
-
- LOG.info("Listed {} jobs in metalake {}", jobEntities.size(),
metalake);
- return Utils.ok(new JobListResponse(jobDTOs));
+ List<JobEntity> filteredAndSortedJobs =
+ filterAndSortJobs(
+ jobEntities,
+ queuedAfterInstant,
+ startedAfterInstant,
+ finishedAfterInstant,
+ comparator);
+ List<JobDTO> jobDTOs = toJobDTOs(filteredAndSortedJobs);
+ Map<String, Long> statusCounts =
countJobsByStatus(filteredAndSortedJobs);
+
+ LOG.info("Listed {} jobs in metalake {}", jobDTOs.size(),
metalake);
+ return Utils.ok(new JobListResponse(jobDTOs, statusCounts));
});
} catch (Exception e) {
@@ -485,4 +524,121 @@ public class JobOperations {
private static List<JobDTO> toJobDTOs(List<JobEntity> jobEntities) {
return
jobEntities.stream().map(JobOperations::toDTO).collect(Collectors.toList());
}
+
+ @VisibleForTesting
+ static Map<String, Long> countJobsByStatus(List<JobEntity> jobEntities) {
+ // Every status is present, even at zero, so callers get a stable set of
keys to render
+ // (e.g. a status histogram) without having to special-case missing
entries.
+ Map<String, Long> statusCounts = new LinkedHashMap<>();
+ for (JobHandle.Status status : JobHandle.Status.values()) {
+ statusCounts.put(status.name().toLowerCase(Locale.ROOT), 0L);
+ }
+ for (JobEntity jobEntity : jobEntities) {
+ statusCounts.merge(jobEntity.status().name().toLowerCase(Locale.ROOT),
1L, Long::sum);
+ }
+ return statusCounts;
+ }
+
+ @VisibleForTesting
+ static List<JobEntity> filterAndSortJobs(
+ List<JobEntity> jobEntities,
+ Instant queuedAfter,
+ Instant startedAfter,
+ Instant finishedAfter,
+ Comparator<JobEntity> comparator) {
+ return jobEntities.stream()
+ .filter(
+ jobEntity -> matchesTimeFilters(jobEntity, queuedAfter,
startedAfter, finishedAfter))
+ .sorted(comparator)
+ .collect(Collectors.toList());
+ }
+
+ @VisibleForTesting
+ static Instant parseInstant(String paramName, String value) {
+ if (Strings.isNullOrEmpty(value)) {
+ return null;
+ }
+ try {
+ // OffsetDateTime.parse (RFC 3339, matching the OpenAPI `format:
date-time`) rather than
+ // Instant.parse (strict ISO_INSTANT, `Z`/zero-offset only):
Instant.parse of a numeric
+ // offset like "+08:00" throws on JDK 8/11 and only started accepting it
on JDK 12+
+ // (JDK-8166138), so the same request would 400 or succeed depending on
the server's JDK.
+ return OffsetDateTime.parse(value).toInstant();
+ } catch (DateTimeParseException e) {
+ throw new IllegalArgumentException(
+ "Invalid "
+ + paramName
+ + " value: "
+ + value
+ + ", must be an ISO-8601 instant, e.g. 2026-08-18T00:00:00Z",
+ e);
+ }
+ }
+
+ // queuedAfter/startedAfter/finishedAfter are inclusive lower bounds (>=),
AND-combined. A job
+ // missing the relevant timestamp (e.g. not started yet) never matches a
startedAfter/
+ // finishedAfter filter.
+ private static boolean matchesTimeFilters(
+ JobEntity jobEntity, Instant queuedAfter, Instant startedAfter, Instant
finishedAfter) {
+ return matchesTimeFilter(jobEntity.auditInfo().createTime(), queuedAfter)
+ && matchesTimeFilter(jobEntity.startedAtAsInstant(), startedAfter)
+ && matchesTimeFilter(jobEntity.finishedAtAsInstant(), finishedAfter);
+ }
+
+ private static boolean matchesTimeFilter(Instant actual, Instant after) {
+ if (after == null) {
+ return true;
+ }
+ return actual != null && !actual.isBefore(after);
+ }
+
+ // sortBy/sortOrder are matched exactly against the OpenAPI-documented enum
values (queuedAt,
+ // startedAt, finishedAt, asc, desc) rather than case-insensitively, so
accepted input always
+ // matches what the spec advertises.
+ @VisibleForTesting
+ static Comparator<JobEntity> buildJobComparator(String sortBy, String
sortOrder) {
+ Function<JobEntity, Instant> keyExtractor;
+ switch (sortBy) {
+ case "queuedAt":
+ keyExtractor = jobEntity -> jobEntity.auditInfo().createTime();
+ break;
+ case "startedAt":
+ keyExtractor = JobEntity::startedAtAsInstant;
+ break;
+ case "finishedAt":
+ keyExtractor = JobEntity::finishedAtAsInstant;
+ break;
+ default:
+ throw new IllegalArgumentException(
+ "Invalid sortBy value: " + sortBy + ", must be one of queuedAt,
startedAt, finishedAt");
+ }
+
+ boolean ascending;
+ switch (sortOrder) {
+ case "asc":
+ ascending = true;
+ break;
+ case "desc":
+ ascending = false;
+ break;
+ default:
+ throw new IllegalArgumentException(
+ "Invalid sortOrder value: " + sortOrder + ", must be one of asc,
desc");
+ }
+
+ // Jobs missing the sort key (e.g. not started/finished yet) always sort
last, regardless of
+ // sortOrder - flipping their position based on direction would be
confusing.
+ return (a, b) -> {
+ Instant ta = keyExtractor.apply(a);
+ Instant tb = keyExtractor.apply(b);
+ if (ta == null && tb == null) {
+ return 0;
+ } else if (ta == null) {
+ return 1;
+ } else if (tb == null) {
+ return -1;
+ }
+ return ascending ? ta.compareTo(tb) : tb.compareTo(ta);
+ };
+ }
}
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 cdd53e6cd8..130fae9434 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
@@ -32,6 +32,9 @@ import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import java.io.IOException;
import java.time.Instant;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Random;
@@ -640,11 +643,14 @@ public class TestJobOperations extends JerseyTest {
@Test
public void testListJobs() {
String templateName = "shell_template_1";
- JobEntity job1 = newJobEntity(templateName, JobHandle.Status.QUEUED);
+ // Fixed, strictly-increasing queuedAt values rather than back-to-back
Instant.now() calls -
+ // the latter can collide (millisecond clock resolution on some
JDKs/OSes), which would make
+ // the desc-sort assertions below flaky.
+ JobEntity job1 = newJobEntityWithQueuedAt(templateName,
JobHandle.Status.QUEUED, 1000L, 0L, 0L);
JobEntity job2 =
- newJobEntity(templateName, JobHandle.Status.STARTED,
Instant.now().toEpochMilli(), 0L);
+ newJobEntityWithQueuedAt(templateName, JobHandle.Status.STARTED,
2000L, 2500L, 0L);
JobEntity job3 =
- newJobEntity("spark_template_1", JobHandle.Status.SUCCEEDED,
Instant.now().toEpochMilli());
+ newJobEntityWithQueuedAt("spark_template_1",
JobHandle.Status.SUCCEEDED, 3000L, 0L, 3500L);
when(jobOperationDispatcher.listJobs(metalake, Optional.empty()))
.thenReturn(Lists.newArrayList(job1, job2, job3));
@@ -661,28 +667,41 @@ public class TestJobOperations extends JerseyTest {
JobListResponse jobListResponse = resp.readEntity(JobListResponse.class);
Assertions.assertEquals(0, jobListResponse.getCode());
+ // statusCounts reflects the returned jobs: one QUEUED, one STARTED, one
SUCCEEDED, and every
+ // other status present at zero.
+ Map<String, Long> expectedStatusCounts = new HashMap<>();
+ expectedStatusCounts.put("queued", 1L);
+ expectedStatusCounts.put("started", 1L);
+ expectedStatusCounts.put("failed", 0L);
+ expectedStatusCounts.put("succeeded", 1L);
+ expectedStatusCounts.put("cancelling", 0L);
+ expectedStatusCounts.put("cancelled", 0L);
+ Assertions.assertEquals(expectedStatusCounts,
jobListResponse.getStatusCounts());
+
+ // Default sort is queuedAt desc (newest first); job1/job2/job3 have
strictly increasing
+ // queuedAt (1000L < 2000L < 3000L), so the response reverses that order.
Assertions.assertEquals(3, jobListResponse.getJobs().size());
- Assertions.assertEquals(JobOperations.toDTO(job1),
jobListResponse.getJobs().get(0));
+ Assertions.assertEquals(JobOperations.toDTO(job3),
jobListResponse.getJobs().get(0));
Assertions.assertEquals(JobOperations.toDTO(job2),
jobListResponse.getJobs().get(1));
- Assertions.assertEquals(JobOperations.toDTO(job3),
jobListResponse.getJobs().get(2));
+ Assertions.assertEquals(JobOperations.toDTO(job1),
jobListResponse.getJobs().get(2));
- // Not-yet-finished jobs round-trip finishedAt as null over the wire.
- Assertions.assertNull(jobListResponse.getJobs().get(0).finishedAt());
- Assertions.assertNull(jobListResponse.getJobs().get(1).finishedAt());
// A finished job round-trips its finishedAt as an Instant over the wire.
Assertions.assertEquals(
- Instant.ofEpochMilli(job3.finishedAt()),
jobListResponse.getJobs().get(2).finishedAt());
+ Instant.ofEpochMilli(job3.finishedAt()),
jobListResponse.getJobs().get(0).finishedAt());
+ // Not-yet-finished jobs round-trip finishedAt as null over the wire.
+ Assertions.assertNull(jobListResponse.getJobs().get(1).finishedAt());
+ Assertions.assertNull(jobListResponse.getJobs().get(2).finishedAt());
// queuedAt is always present, regardless of status.
Assertions.assertNotNull(jobListResponse.getJobs().get(0).queuedAt());
Assertions.assertNotNull(jobListResponse.getJobs().get(1).queuedAt());
Assertions.assertNotNull(jobListResponse.getJobs().get(2).queuedAt());
- // A not-yet-started job round-trips startedAt as null over the wire.
- Assertions.assertNull(jobListResponse.getJobs().get(0).startedAt());
// A started job round-trips its startedAt as an Instant over the wire.
Assertions.assertEquals(
Instant.ofEpochMilli(job2.startedAt()),
jobListResponse.getJobs().get(1).startedAt());
+ // A not-yet-started job round-trips startedAt as null over the wire.
+ Assertions.assertNull(jobListResponse.getJobs().get(2).startedAt());
// Test list jobs by template name
when(jobOperationDispatcher.listJobs(metalake, Optional.of(templateName)))
@@ -701,8 +720,9 @@ public class TestJobOperations extends JerseyTest {
JobListResponse jobListResponse1 = resp1.readEntity(JobListResponse.class);
Assertions.assertEquals(0, jobListResponse1.getCode());
Assertions.assertEquals(2, jobListResponse1.getJobs().size());
- Assertions.assertEquals(JobOperations.toDTO(job1),
jobListResponse1.getJobs().get(0));
- Assertions.assertEquals(JobOperations.toDTO(job2),
jobListResponse1.getJobs().get(1));
+ // Default sort is queuedAt desc: job2 (queued after job1) comes first.
+ Assertions.assertEquals(JobOperations.toDTO(job2),
jobListResponse1.getJobs().get(0));
+ Assertions.assertEquals(JobOperations.toDTO(job1),
jobListResponse1.getJobs().get(1));
// Test throw NoSuchMetalakeException
doThrow(new NoSuchMetalakeException("mock error"))
@@ -772,6 +792,276 @@ public class TestJobOperations extends JerseyTest {
Assertions.assertEquals(RuntimeException.class.getSimpleName(),
errorResp4.getType());
}
+ @Test
+ public void testListJobsWithTimeFiltersAndSort() {
+ String templateName = "shell_template_1";
+ // job1: queued only. job2: queued+started. job3: queued+started+finished,
queued earliest.
+ JobEntity job1 = newJobEntityWithQueuedAt(templateName,
JobHandle.Status.QUEUED, 3000L, 0L, 0L);
+ JobEntity job2 =
+ newJobEntityWithQueuedAt(templateName, JobHandle.Status.STARTED,
2000L, 2500L, 0L);
+ JobEntity job3 =
+ newJobEntityWithQueuedAt(templateName, JobHandle.Status.SUCCEEDED,
1000L, 1500L, 1800L);
+
+ when(jobOperationDispatcher.listJobs(metalake, Optional.empty()))
+ .thenReturn(Lists.newArrayList(job1, job2, job3));
+
+ // startedAfter excludes job1 (never started); default sort is queuedAt
desc.
+ Response resp =
+ target(jobRunPath())
+ .queryParam("startedAfter", Instant.ofEpochMilli(1000L).toString())
+ .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(2, jobListResponse.getJobs().size());
+ Assertions.assertEquals(JobOperations.toDTO(job2),
jobListResponse.getJobs().get(0));
+ Assertions.assertEquals(JobOperations.toDTO(job3),
jobListResponse.getJobs().get(1));
+
+ // statusCounts is scoped to the filtered set: job1 (QUEUED) is excluded
by startedAfter,
+ // so its status contributes zero here, even though the job itself exists.
+ Map<String, Long> expectedFilteredStatusCounts = new HashMap<>();
+ expectedFilteredStatusCounts.put("queued", 0L);
+ expectedFilteredStatusCounts.put("started", 1L);
+ expectedFilteredStatusCounts.put("failed", 0L);
+ expectedFilteredStatusCounts.put("succeeded", 1L);
+ expectedFilteredStatusCounts.put("cancelling", 0L);
+ expectedFilteredStatusCounts.put("cancelled", 0L);
+ Assertions.assertEquals(expectedFilteredStatusCounts,
jobListResponse.getStatusCounts());
+
+ // sortBy=startedAt, sortOrder=asc: job3 (1500) < job2 (2500) < job1
(null, sorts last).
+ Response resp2 =
+ target(jobRunPath())
+ .queryParam("sortBy", "startedAt")
+ .queryParam("sortOrder", "asc")
+ .request(APPLICATION_JSON_TYPE)
+ .accept("application/vnd.gravitino.v1+json")
+ .get();
+
+ Assertions.assertEquals(Response.Status.OK.getStatusCode(),
resp2.getStatus());
+ JobListResponse jobListResponse2 = resp2.readEntity(JobListResponse.class);
+ Assertions.assertEquals(3, jobListResponse2.getJobs().size());
+ Assertions.assertEquals(JobOperations.toDTO(job3),
jobListResponse2.getJobs().get(0));
+ Assertions.assertEquals(JobOperations.toDTO(job2),
jobListResponse2.getJobs().get(1));
+ Assertions.assertEquals(JobOperations.toDTO(job1),
jobListResponse2.getJobs().get(2));
+
+ // Invalid sortBy is rejected with 400.
+ Response resp3 =
+ target(jobRunPath())
+ .queryParam("sortBy", "bogus")
+ .request(APPLICATION_JSON_TYPE)
+ .accept("application/vnd.gravitino.v1+json")
+ .get();
+
+ Assertions.assertEquals(Response.Status.BAD_REQUEST.getStatusCode(),
resp3.getStatus());
+ ErrorResponse errorResp3 = resp3.readEntity(ErrorResponse.class);
+ Assertions.assertEquals(ErrorConstants.ILLEGAL_ARGUMENTS_CODE,
errorResp3.getCode());
+ Assertions.assertEquals(IllegalArgumentException.class.getSimpleName(),
errorResp3.getType());
+
+ // Invalid sortOrder is rejected with 400.
+ Response resp4 =
+ target(jobRunPath())
+ .queryParam("sortOrder", "sideways")
+ .request(APPLICATION_JSON_TYPE)
+ .accept("application/vnd.gravitino.v1+json")
+ .get();
+
+ Assertions.assertEquals(Response.Status.BAD_REQUEST.getStatusCode(),
resp4.getStatus());
+ ErrorResponse errorResp4 = resp4.readEntity(ErrorResponse.class);
+ Assertions.assertEquals(ErrorConstants.ILLEGAL_ARGUMENTS_CODE,
errorResp4.getCode());
+ Assertions.assertEquals(IllegalArgumentException.class.getSimpleName(),
errorResp4.getType());
+
+ // Non-ISO-8601 time value is rejected with 400.
+ Response resp5 =
+ target(jobRunPath())
+ .queryParam("queuedAfter", "not-a-timestamp")
+ .request(APPLICATION_JSON_TYPE)
+ .accept("application/vnd.gravitino.v1+json")
+ .get();
+
+ Assertions.assertEquals(Response.Status.BAD_REQUEST.getStatusCode(),
resp5.getStatus());
+ ErrorResponse errorResp5 = resp5.readEntity(ErrorResponse.class);
+ Assertions.assertEquals(ErrorConstants.ILLEGAL_ARGUMENTS_CODE,
errorResp5.getCode());
+ Assertions.assertEquals(IllegalArgumentException.class.getSimpleName(),
errorResp5.getType());
+
+ // Blank query params (e.g. "?sortBy=&sortOrder=&queuedAfter="), as
templated/generated
+ // clients sometimes send, fall back to "not set" rather than 400ing:
@DefaultValue only
+ // applies when a param is absent, not when it's present-but-empty.
+ Response resp6 =
+ target(jobRunPath())
+ .queryParam("sortBy", "")
+ .queryParam("sortOrder", "")
+ .queryParam("queuedAfter", "")
+ .request(APPLICATION_JSON_TYPE)
+ .accept("application/vnd.gravitino.v1+json")
+ .get();
+
+ Assertions.assertEquals(Response.Status.OK.getStatusCode(),
resp6.getStatus());
+ JobListResponse jobListResponse6 = resp6.readEntity(JobListResponse.class);
+ Assertions.assertEquals(3, jobListResponse6.getJobs().size());
+ Assertions.assertEquals(JobOperations.toDTO(job1),
jobListResponse6.getJobs().get(0));
+ Assertions.assertEquals(JobOperations.toDTO(job2),
jobListResponse6.getJobs().get(1));
+ Assertions.assertEquals(JobOperations.toDTO(job3),
jobListResponse6.getJobs().get(2));
+ }
+
+ @Test
+ public void testListJobsCombinesTemplateFilterWithTimeFiltersAndSort() {
+ String templateName = "shell_template_1";
+ JobEntity job1 = newJobEntityWithQueuedAt(templateName,
JobHandle.Status.QUEUED, 3000L, 0L, 0L);
+ JobEntity job2 =
+ newJobEntityWithQueuedAt(templateName, JobHandle.Status.STARTED,
2000L, 2500L, 0L);
+ JobEntity job3 =
+ newJobEntityWithQueuedAt(templateName, JobHandle.Status.SUCCEEDED,
1000L, 1500L, 1800L);
+
+ // The dispatcher already narrows to this template's jobs; jobs from other
templates (e.g. a
+ // spark_template_1 job) are never in this list, so they can't leak in via
the time filter.
+ when(jobOperationDispatcher.listJobs(metalake, Optional.of(templateName)))
+ .thenReturn(Lists.newArrayList(job1, job2, job3));
+
+ Response resp =
+ target(jobRunPath())
+ .queryParam("jobTemplateName", templateName)
+ .queryParam("startedAfter", Instant.ofEpochMilli(1000L).toString())
+ .queryParam("sortBy", "startedAt")
+ .queryParam("sortOrder", "asc")
+ .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);
+ // job1 is excluded (never started); job3 (startedAt=1500) sorts before
job2 (startedAt=2500).
+ Assertions.assertEquals(2, jobListResponse.getJobs().size());
+ Assertions.assertEquals(JobOperations.toDTO(job3),
jobListResponse.getJobs().get(0));
+ Assertions.assertEquals(JobOperations.toDTO(job2),
jobListResponse.getJobs().get(1));
+ }
+
+ @Test
+ public void testFilterAndSortJobs() {
+ String templateName = "shell_template_1";
+ JobEntity job1 = newJobEntityWithQueuedAt(templateName,
JobHandle.Status.QUEUED, 1000L, 0L, 0L);
+ JobEntity job2 =
+ newJobEntityWithQueuedAt(templateName, JobHandle.Status.STARTED,
2000L, 3000L, 0L);
+ JobEntity job3 =
+ newJobEntityWithQueuedAt(templateName, JobHandle.Status.SUCCEEDED,
3000L, 4000L, 5000L);
+ List<JobEntity> jobs = Lists.newArrayList(job1, job2, job3);
+ Comparator<JobEntity> queuedAtAsc =
JobOperations.buildJobComparator("queuedAt", "asc");
+
+ // queuedAfter filters out job1.
+ List<JobEntity> filteredByQueuedAfter =
+ JobOperations.filterAndSortJobs(jobs, instant(1500L), null, null,
queuedAtAsc);
+ Assertions.assertEquals(Lists.newArrayList(job2, job3),
filteredByQueuedAfter);
+
+ // startedAfter excludes not-yet-started job1 and jobs started before the
bound.
+ List<JobEntity> filteredByStartedAfter =
+ JobOperations.filterAndSortJobs(jobs, null, instant(3500L), null,
queuedAtAsc);
+ Assertions.assertEquals(Lists.newArrayList(job3), filteredByStartedAfter);
+
+ // finishedAfter excludes not-yet-finished jobs.
+ List<JobEntity> filteredByFinishedAfter =
+ JobOperations.filterAndSortJobs(jobs, null, null, instant(1L),
queuedAtAsc);
+ Assertions.assertEquals(Lists.newArrayList(job3), filteredByFinishedAfter);
+
+ // Filters are AND-combined.
+ List<JobEntity> filteredByAll =
+ JobOperations.filterAndSortJobs(
+ jobs, instant(1500L), instant(3500L), instant(1L), queuedAtAsc);
+ Assertions.assertEquals(Lists.newArrayList(job3), filteredByAll);
+
+ // Boundary is inclusive (>=).
+ List<JobEntity> filteredInclusive =
+ JobOperations.filterAndSortJobs(jobs, instant(2000L), null, null,
queuedAtAsc);
+ Assertions.assertEquals(Lists.newArrayList(job2, job3), filteredInclusive);
+
+ // sortBy=queuedAt desc.
+ List<JobEntity> sortedByQueuedDesc =
+ JobOperations.filterAndSortJobs(
+ jobs, null, null, null,
JobOperations.buildJobComparator("queuedAt", "desc"));
+ Assertions.assertEquals(Lists.newArrayList(job3, job2, job1),
sortedByQueuedDesc);
+
+ // sortBy=startedAt asc, jobs without startedAt sort last regardless of
direction.
+ List<JobEntity> sortedByStartedAsc =
+ JobOperations.filterAndSortJobs(
+ jobs, null, null, null,
JobOperations.buildJobComparator("startedAt", "asc"));
+ Assertions.assertEquals(Lists.newArrayList(job2, job3, job1),
sortedByStartedAsc);
+
+ List<JobEntity> sortedByStartedDesc =
+ JobOperations.filterAndSortJobs(
+ jobs, null, null, null,
JobOperations.buildJobComparator("startedAt", "desc"));
+ Assertions.assertEquals(Lists.newArrayList(job3, job2, job1),
sortedByStartedDesc);
+ }
+
+ @Test
+ public void testCountJobsByStatus() {
+ String templateName = "shell_template_1";
+ JobEntity queuedJob1 = newJobEntity(templateName, JobHandle.Status.QUEUED);
+ JobEntity queuedJob2 = newJobEntity(templateName, JobHandle.Status.QUEUED);
+ JobEntity startedJob =
+ newJobEntity(templateName, JobHandle.Status.STARTED,
Instant.now().toEpochMilli(), 0L);
+ JobEntity succeededJob =
+ newJobEntity(templateName, JobHandle.Status.SUCCEEDED,
Instant.now().toEpochMilli());
+
+ Map<String, Long> counts =
+ JobOperations.countJobsByStatus(
+ Lists.newArrayList(queuedJob1, queuedJob2, startedJob,
succeededJob));
+
+ // Every JobHandle.Status is present, even at zero.
+ Map<String, Long> expected = new HashMap<>();
+ expected.put("queued", 2L);
+ expected.put("started", 1L);
+ expected.put("failed", 0L);
+ expected.put("succeeded", 1L);
+ expected.put("cancelling", 0L);
+ expected.put("cancelled", 0L);
+ Assertions.assertEquals(expected, counts);
+
+ // An empty job list still reports every status at zero.
+ Map<String, Long> emptyCounts =
JobOperations.countJobsByStatus(Lists.newArrayList());
+ Assertions.assertEquals(6, emptyCounts.size());
+ emptyCounts.values().forEach(count -> Assertions.assertEquals(0L, count));
+ }
+
+ @Test
+ public void testParseInstant() {
+ Assertions.assertNull(JobOperations.parseInstant("queuedAfter", null));
+ Assertions.assertNull(JobOperations.parseInstant("queuedAfter", ""));
+
+ Instant expected = Instant.ofEpochMilli(1500L);
+ Assertions.assertEquals(
+ expected, JobOperations.parseInstant("queuedAfter",
expected.toString()));
+
+ // RFC 3339 numeric offsets (as advertised by the OpenAPI `format:
date-time`) are accepted,
+ // not just the strict ISO_INSTANT `Z` form.
+ Assertions.assertEquals(
+ Instant.parse("2026-08-17T16:00:00Z"),
+ JobOperations.parseInstant("queuedAfter",
"2026-08-18T00:00:00+08:00"));
+
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> JobOperations.parseInstant("queuedAfter", "not-a-timestamp"));
+ }
+
+ @Test
+ public void testBuildJobComparatorValidation() {
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
JobOperations.buildJobComparator("bogus", "asc"));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> JobOperations.buildJobComparator("queuedAt", "bogus"));
+
+ // sortBy/sortOrder are matched exactly against the OpenAPI-documented
casing, not
+ // case-insensitively.
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
JobOperations.buildJobComparator("queuedat", "asc"));
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
JobOperations.buildJobComparator("queuedAt", "ASC"));
+ }
+
+ private static Instant instant(long epochMilli) {
+ return Instant.ofEpochMilli(epochMilli);
+ }
+
@Test
public void testRunJob() {
String templateName = "shell_template_1";
@@ -967,4 +1257,27 @@ public class TestJobOperations extends JerseyTest {
.withFinishedAt(finishedAt)
.build();
}
+
+ private JobEntity newJobEntityWithQueuedAt(
+ String templateName,
+ JobHandle.Status status,
+ long queuedAtEpochMilli,
+ Long startedAt,
+ Long finishedAt) {
+ Random rand = new Random();
+ return JobEntity.builder()
+ .withId(rand.nextLong())
+ .withJobExecutionId(rand.nextLong() + "")
+ .withNamespace(NamespaceUtil.ofJob(metalake))
+ .withJobTemplateName(templateName)
+ .withStatus(status)
+ .withAuditInfo(
+ AuditInfo.builder()
+ .withCreator("test")
+ .withCreateTime(Instant.ofEpochMilli(queuedAtEpochMilli))
+ .build())
+ .withStartedAt(startedAt)
+ .withFinishedAt(finishedAt)
+ .build();
+ }
}
