Skip to content

Commit 7699d69

Browse files
chesches
committed
core: Make isTerminal an instance property of JobStatus
1 parent 7903dc7 commit 7699d69

7 files changed

Lines changed: 38 additions & 36 deletions

File tree

‎core/src/main/java/feast/core/job/dataflow/DataflowJobManager.java‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -193,16 +193,15 @@ public void abortJob(String dataflowJobId) {
193193
}
194194

195195
/**
196-
* Restart a restart dataflow job. Dataflow should ensure continuity between during the restart,
197-
* so no data should be lost during the restart operation.
196+
* Restart a Dataflow job. Dataflow should ensure continuity such that no data should be lost
197+
* during the restart operation.
198198
*
199199
* @param job job to restart
200200
* @return the restarted job
201201
*/
202202
@Override
203203
public Job restartJob(Job job) {
204-
JobStatus status = job.getStatus();
205-
if (JobStatus.getTerminalState().contains(status)) {
204+
if (job.getStatus().isTerminal()) {
206205
// job yet not running: just start job
207206
return this.startJob(job);
208207
} else {

‎core/src/main/java/feast/core/job/direct/DirectRunnerJobManager.java‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -166,8 +166,7 @@ public PipelineResult runPipeline(ImportOptions pipelineOptions) throws IOExcept
166166
*/
167167
@Override
168168
public Job restartJob(Job job) {
169-
JobStatus status = job.getStatus();
170-
if (JobStatus.getTerminalState().contains(status)) {
169+
if (job.getStatus().isTerminal()) {
171170
// job yet not running: just start job
172171
return this.startJob(job);
173172
} else {

‎core/src/main/java/feast/core/model/Job.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -110,6 +110,10 @@ public Job(
110110
this.status = jobStatus;
111111
}
112112

113+
public boolean hasTerminated() {
114+
return getStatus().isTerminal();
115+
}
116+
113117
public void updateMetrics(List<Metrics> newMetrics) {
114118
metrics.clear();
115119
metrics.addAll(newMetrics);

‎core/src/main/java/feast/core/model/JobStatus.java‎

Lines changed: 22 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -17,10 +17,8 @@
1717
package feast.core.model;
1818

1919
import feast.core.IngestionJobProto.IngestionJobStatus;
20-
import java.util.Arrays;
21-
import java.util.Collection;
22-
import java.util.Collections;
2320
import java.util.Map;
21+
import java.util.Set;
2422

2523
public enum JobStatus {
2624
/** Job status is not known. */
@@ -53,33 +51,41 @@ public enum JobStatus {
5351
/** job has been suspended */
5452
SUSPENDED;
5553

56-
private static final Collection<JobStatus> TERMINAL_STATE =
57-
Collections.unmodifiableList(Arrays.asList(COMPLETED, ABORTED, ERROR));
54+
private static final Set<JobStatus> TERMINAL_STATES = Set.of(COMPLETED, ABORTED, ERROR);
5855

5956
/**
60-
* Get a collection of terminal job state.
57+
* Get the set of terminal job states.
6158
*
62-
* <p>Terminal job state is final and will not change to any other state.
59+
* <p>A terminal job state is final and will not change to any other state.
6360
*
64-
* @return collection of terminal job state.
61+
* @return set of terminal job states.
6562
*/
66-
public static Collection<JobStatus> getTerminalState() {
67-
return TERMINAL_STATE;
63+
public static Set<JobStatus> getTerminalStates() {
64+
return TERMINAL_STATES;
6865
}
6966

70-
private static final Collection<JobStatus> TRANSITIONAL_STATES =
71-
Collections.unmodifiableList(Arrays.asList(PENDING, ABORTING, SUSPENDING));
67+
private static final Set<JobStatus> TRANSITIONAL_STATES = Set.of(PENDING, ABORTING, SUSPENDING);
7268

7369
/**
74-
* Get Transitional Job Status states. Transitionals states are assigned to jobs that
70+
* Get Transitional Job Status states. Transitional states are assigned to jobs that are
7571
* transitioning to a more stable state (ie SUSPENDED, ABORTED etc.)
7672
*
77-
* @return Collection of transitional Job Status states.
73+
* @return set of transitional Job Status states.
7874
*/
79-
public static final Collection<JobStatus> getTransitionalStates() {
75+
public static Set<JobStatus> getTransitionalStates() {
8076
return TRANSITIONAL_STATES;
8177
}
8278

79+
/** @return true if this {@code JobStatus} is a terminal state. */
80+
public boolean isTerminal() {
81+
return getTerminalStates().contains(this);
82+
}
83+
84+
/** @return true if this {@code JobStatus} is a transitional state. */
85+
public boolean isTransitional() {
86+
return getTransitionalStates().contains(this);
87+
}
88+
8389
private static final Map<JobStatus, IngestionJobStatus> INGESTION_JOB_STATUS_MAP =
8490
Map.of(
8591
JobStatus.UNKNOWN, IngestionJobStatus.UNKNOWN,
@@ -95,7 +101,7 @@ public static final Collection<JobStatus> getTransitionalStates() {
95101
/**
96102
* Convert a Job Status to Ingestion Job Status proto
97103
*
98-
* @return IngestionJobStatus proto derieved from this job status
104+
* @return IngestionJobStatus proto derived from this job status
99105
*/
100106
public IngestionJobStatus toProto() {
101107
// maps job models job status to ingestion job status

‎core/src/main/java/feast/core/service/JobCoordinatorService.java‎

Lines changed: 2 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -189,11 +189,8 @@ public Optional<Job> getJob(Source source, Store store) {
189189
List<Job> jobs =
190190
jobRepository.findBySourceIdAndStoreNameOrderByLastUpdatedDesc(
191191
source.getId(), store.getName());
192-
jobs =
193-
jobs.stream()
194-
.filter(job -> !JobStatus.getTerminalState().contains(job.getStatus()))
195-
.collect(Collectors.toList());
196-
if (jobs.size() == 0) {
192+
jobs = jobs.stream().filter(job -> !job.hasTerminated()).collect(Collectors.toList());
193+
if (jobs.isEmpty()) {
197194
return Optional.empty();
198195
}
199196
// return the latest

‎core/src/main/java/feast/core/service/JobService.java‎

Lines changed: 4 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -158,16 +158,15 @@ public RestartIngestionJobResponse restartJob(RestartIngestionJobRequest request
158158
// check job exists
159159
Optional<Job> getJob = this.jobRepository.findById(request.getId());
160160
if (getJob.isEmpty()) {
161+
// FIXME: if getJob.isEmpty then constructing this error message will always throw an error...
161162
throw new NoSuchElementException(
162163
"Attempted to stop nonexistent job with id: " + getJob.get().getId());
163164
}
164165

165166
// check job status is valid for restarting
166167
Job job = getJob.get();
167168
JobStatus status = job.getStatus();
168-
if (JobStatus.getTransitionalStates().contains(status)
169-
|| JobStatus.getTerminalState().contains(status)
170-
|| status.equals(JobStatus.UNKNOWN)) {
169+
if (status.isTransitional() || status.isTerminal() || status == JobStatus.UNKNOWN) {
171170
throw new UnsupportedOperationException(
172171
"Restarting a job with a transitional, terminal or unknown status is unsupported");
173172
}
@@ -208,11 +207,10 @@ public StopIngestionJobResponse stopJob(StopIngestionJobRequest request)
208207
// check job status is valid for stopping
209208
Job job = getJob.get();
210209
JobStatus status = job.getStatus();
211-
if (JobStatus.getTerminalState().contains(status)) {
210+
if (status.isTerminal()) {
212211
// do nothing - job is already stopped
213212
return StopIngestionJobResponse.newBuilder().build();
214-
} else if (JobStatus.getTransitionalStates().contains(status)
215-
|| status.equals(JobStatus.UNKNOWN)) {
213+
} else if (status.isTransitional() || status == JobStatus.UNKNOWN) {
216214
throw new UnsupportedOperationException(
217215
"Stopping a job with a transitional or unknown status is unsupported");
218216
}

‎core/src/test/java/feast/core/service/JobServiceTest.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -341,10 +341,9 @@ public void testStopJobForId() {
341341
}
342342

343343
@Test
344-
public void testStopAlreadyStop() {
344+
public void testStopAlreadyStopped() {
345345
// check that stop jobs does not trying to stop jobs that are not already stopped
346-
List<JobStatus> doNothingStatuses = new ArrayList<>();
347-
doNothingStatuses.addAll(JobStatus.getTerminalState());
346+
List<JobStatus> doNothingStatuses = new ArrayList<>(JobStatus.getTerminalStates());
348347

349348
JobStatus prevStatus = this.job.getStatus();
350349
for (JobStatus status : doNothingStatuses) {

0 commit comments

Comments
 (0)