Skip to content

Commit 78b6024

Browse files
chesches
authored
core: Use Runner enum type instead of string for Job model (#651)
#575 sought to clear up inconsistencies between uses of `Runner#name()` (the standard final method of `java.lang.Enum` that returns the value's enum constant name) and the riskily-named `Runner#getName()` defined in Feast for human-readable Beam Runner names. The latter is used as runner name users can set in config. The former is used for values of the runner column of the jobs table in SQL (as it should be). But it relied on careful coding to use the right one when constructing `Job` instances. This is error prone, as #578 demonstrates. There is a more robust way: use the enum instead of stringly-typed programming. It's one of the reasons we have enums :-) This also renames the internal identifier in the Runner definition to `humanName`, to distinguish it further from `Enum#name()`.
1 parent e5bc18c commit 78b6024

11 files changed

Lines changed: 90 additions & 42 deletions

File tree

‎core/src/main/java/feast/core/job/JobUpdateTask.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -144,7 +144,7 @@ private Job startJob(
144144
new Job(
145145
jobId,
146146
"",
147-
jobManager.getRunnerType().name(),
147+
jobManager.getRunnerType(),
148148
Source.fromProto(source),
149149
Store.fromProto(sinkSpec),
150150
featureSets,

‎core/src/main/java/feast/core/job/Runner.java‎

Lines changed: 15 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -16,33 +16,37 @@
1616
*/
1717
package feast.core.job;
1818

19+
import java.util.NoSuchElementException;
20+
21+
/**
22+
* An Apache Beam Runner, for which Feast Core supports managing ingestion jobs.
23+
*
24+
* @see <a href="https://beam.apache.org/documentation/#runners">Beam Runners</a>
25+
*/
1926
public enum Runner {
2027
DATAFLOW("DataflowRunner"),
2128
FLINK("FlinkRunner"),
2229
DIRECT("DirectRunner");
2330

24-
private final String name;
31+
private final String humanName;
2532

26-
Runner(String name) {
27-
this.name = name;
33+
Runner(String humanName) {
34+
this.humanName = humanName;
2835
}
2936

30-
/**
31-
* Get the human readable name of this runner. Returns a human readable name of the runner that
32-
* can be used for logging/config files/etc.
33-
*/
37+
/** Returns the human readable name of this runner, usable in logging, config files, etc. */
3438
@Override
3539
public String toString() {
36-
return name;
40+
return humanName;
3741
}
3842

3943
/** Parses a runner from its human readable name. */
40-
public static Runner fromString(String runner) {
44+
public static Runner fromString(String humanName) {
4145
for (Runner r : Runner.values()) {
42-
if (r.toString().equals(runner)) {
46+
if (r.toString().equals(humanName)) {
4347
return r;
4448
}
4549
}
46-
throw new IllegalArgumentException("Unknown value: " + runner);
50+
throw new NoSuchElementException("Unknown Runner value: " + humanName);
4751
}
4852
}

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -220,7 +220,7 @@ public Job restartJob(Job job) {
220220
*/
221221
@Override
222222
public JobStatus getJobStatus(Job job) {
223-
if (!Runner.DATAFLOW.name().equals(job.getRunner())) {
223+
if (job.getRunner() != RUNNER_TYPE) {
224224
return job.getStatus();
225225
}
226226

@@ -252,7 +252,7 @@ private Job submitDataflowJob(
252252
return new Job(
253253
jobName,
254254
jobId,
255-
getRunnerType().name(),
255+
getRunnerType(),
256256
Source.fromProto(source),
257257
Store.fromProto(sink),
258258
featureSets,

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

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import com.google.protobuf.InvalidProtocolBufferException;
2020
import feast.core.FeatureSetProto;
2121
import feast.core.IngestionJobProto;
22+
import feast.core.job.Runner;
2223
import java.util.ArrayList;
2324
import java.util.List;
2425
import javax.persistence.CascadeType;
@@ -55,9 +56,9 @@ public class Job extends AbstractTimestampEntity {
5556
private String extId;
5657

5758
// Runner type
58-
// Use Runner.name() when converting a Runner to string to assign to this property.
59+
@Enumerated(EnumType.STRING)
5960
@Column(name = "runner")
60-
private String runner;
61+
private Runner runner;
6162

6263
// Source id
6364
@ManyToOne
@@ -96,7 +97,7 @@ public Job() {
9697
public Job(
9798
String id,
9899
String extId,
99-
String runner,
100+
Runner runner,
100101
Source source,
101102
Store sink,
102103
List<FeatureSet> featureSets,

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

Lines changed: 7 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
import feast.core.IngestionJobProto;
3030
import feast.core.dao.JobRepository;
3131
import feast.core.job.JobManager;
32+
import feast.core.job.Runner;
3233
import feast.core.log.Action;
3334
import feast.core.log.AuditLogger;
3435
import feast.core.log.Resource;
@@ -50,13 +51,13 @@
5051
import org.springframework.stereotype.Service;
5152
import org.springframework.transaction.annotation.Transactional;
5253

53-
/** Defines a Job Managemenent Service that allows users to manage feast ingestion jobs. */
54+
/** A Job Management Service that allows users to manage Feast ingestion jobs. */
5455
@Slf4j
5556
@Service
5657
public class JobService {
57-
private JobRepository jobRepository;
58-
private SpecService specService;
59-
private Map<String, JobManager> jobManagers;
58+
private final JobRepository jobRepository;
59+
private final SpecService specService;
60+
private final Map<Runner, JobManager> jobManagers;
6061

6162
@Autowired
6263
public JobService(
@@ -66,13 +67,13 @@ public JobService(
6667

6768
this.jobManagers = new HashMap<>();
6869
for (JobManager manager : jobManagerList) {
69-
this.jobManagers.put(manager.getRunnerType().name(), manager);
70+
this.jobManagers.put(manager.getRunnerType(), manager);
7071
}
7172
}
7273

7374
/* Job Service API */
7475
/**
75-
* List Ingestion Jobs in feast matching the given request. See CoreService protobuf documentation
76+
* List Ingestion Jobs in Feast matching the given request. See CoreService protobuf documentation
7677
* for more detailed documentation.
7778
*
7879
* @param request list ingestion jobs request specifying which jobs to include

‎core/src/test/java/feast/core/job/JobUpdateTaskTest.java‎

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -102,7 +102,7 @@ public void shouldUpdateJobIfPresent() {
102102
new Job(
103103
"job",
104104
"old_ext",
105-
Runner.DATAFLOW.name(),
105+
Runner.DATAFLOW,
106106
feast.core.model.Source.fromProto(source),
107107
feast.core.model.Store.fromProto(store),
108108
Arrays.asList(FeatureSet.fromProto(featureSet1)),
@@ -119,7 +119,7 @@ public void shouldUpdateJobIfPresent() {
119119
new Job(
120120
"job",
121121
"old_ext",
122-
Runner.DATAFLOW.name(),
122+
Runner.DATAFLOW,
123123
feast.core.model.Source.fromProto(source),
124124
feast.core.model.Store.fromProto(store),
125125
Arrays.asList(FeatureSet.fromProto(featureSet1), FeatureSet.fromProto(featureSet2)),
@@ -129,7 +129,7 @@ public void shouldUpdateJobIfPresent() {
129129
new Job(
130130
"job",
131131
"new_ext",
132-
Runner.DATAFLOW.name(),
132+
Runner.DATAFLOW,
133133
Source.fromProto(source),
134134
Store.fromProto(store),
135135
Arrays.asList(FeatureSet.fromProto(featureSet1), FeatureSet.fromProto(featureSet2)),
@@ -163,7 +163,7 @@ public void shouldCreateJobIfNotPresent() {
163163
new Job(
164164
"job",
165165
"",
166-
Runner.DATAFLOW.name(),
166+
Runner.DATAFLOW,
167167
feast.core.model.Source.fromProto(source),
168168
feast.core.model.Store.fromProto(store),
169169
Arrays.asList(FeatureSet.fromProto(featureSet1)),
@@ -173,7 +173,7 @@ public void shouldCreateJobIfNotPresent() {
173173
new Job(
174174
"job",
175175
"ext",
176-
Runner.DATAFLOW.name(),
176+
Runner.DATAFLOW,
177177
feast.core.model.Source.fromProto(source),
178178
feast.core.model.Store.fromProto(store),
179179
Arrays.asList(FeatureSet.fromProto(featureSet1)),
@@ -202,7 +202,7 @@ public void shouldUpdateJobStatusIfNotCreateOrUpdate() {
202202
new Job(
203203
"job",
204204
"ext",
205-
Runner.DATAFLOW.name(),
205+
Runner.DATAFLOW,
206206
feast.core.model.Source.fromProto(source),
207207
feast.core.model.Store.fromProto(store),
208208
Arrays.asList(FeatureSet.fromProto(featureSet1)),
@@ -216,7 +216,7 @@ public void shouldUpdateJobStatusIfNotCreateOrUpdate() {
216216
new Job(
217217
"job",
218218
"ext",
219-
Runner.DATAFLOW.name(),
219+
Runner.DATAFLOW,
220220
Source.fromProto(source),
221221
Store.fromProto(store),
222222
Arrays.asList(FeatureSet.fromProto(featureSet1)),
@@ -248,7 +248,7 @@ public void shouldReturnJobWithErrorStatusIfFailedToSubmit() {
248248
new Job(
249249
"job",
250250
"",
251-
Runner.DATAFLOW.name(),
251+
Runner.DATAFLOW,
252252
feast.core.model.Source.fromProto(source),
253253
feast.core.model.Store.fromProto(store),
254254
Arrays.asList(FeatureSet.fromProto(featureSet1)),
@@ -258,7 +258,7 @@ public void shouldReturnJobWithErrorStatusIfFailedToSubmit() {
258258
new Job(
259259
"job",
260260
"",
261-
Runner.DATAFLOW.name(),
261+
Runner.DATAFLOW,
262262
feast.core.model.Source.fromProto(source),
263263
feast.core.model.Store.fromProto(store),
264264
Arrays.asList(FeatureSet.fromProto(featureSet1)),

‎core/src/test/java/feast/core/job/RunnerTest.java‎

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,42 @@
1+
/*
2+
* SPDX-License-Identifier: Apache-2.0
3+
* Copyright 2018-2020 The Feast Authors
4+
*
5+
* Licensed under the Apache License, Version 2.0 (the "License");
6+
* you may not use this file except in compliance with the License.
7+
* You may obtain a copy of the License at
8+
*
9+
* https://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package feast.core.job;
18+
19+
import static org.hamcrest.Matchers.is;
20+
import static org.junit.Assert.assertThat;
21+
22+
import java.util.NoSuchElementException;
23+
import org.junit.Test;
24+
25+
public class RunnerTest {
26+
27+
@Test
28+
public void toStringReturnsHumanReadableName() {
29+
assertThat(Runner.DATAFLOW.toString(), is("DataflowRunner"));
30+
}
31+
32+
@Test
33+
public void fromStringLoadsValueFromHumanReadableName() {
34+
var humanName = Runner.DATAFLOW.toString();
35+
assertThat(Runner.fromString(humanName), is(Runner.DATAFLOW));
36+
}
37+
38+
@Test(expected = NoSuchElementException.class)
39+
public void fromStringThrowsNoSuchElementExceptionForUnknownValue() {
40+
Runner.fromString("this is not a valid Runner");
41+
}
42+
}

‎core/src/test/java/feast/core/job/dataflow/DataflowJobManagerTest.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -158,7 +158,7 @@ public void shouldStartJobWithCorrectPipelineOptions() throws IOException {
158158
new Job(
159159
jobName,
160160
"",
161-
Runner.DATAFLOW.name(),
161+
Runner.DATAFLOW,
162162
Source.fromProto(source),
163163
Store.fromProto(store),
164164
Lists.newArrayList(FeatureSet.fromProto(featureSet)),
@@ -239,7 +239,7 @@ public void shouldThrowExceptionWhenJobStateTerminal() throws IOException {
239239
new Job(
240240
"job",
241241
"",
242-
Runner.DATAFLOW.name(),
242+
Runner.DATAFLOW,
243243
Source.fromProto(source),
244244
Store.fromProto(store),
245245
Lists.newArrayList(FeatureSet.fromProto(featureSet)),

‎core/src/test/java/feast/core/job/direct/DirectRunnerJobManagerTest.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -144,7 +144,7 @@ public void shouldStartDirectJobAndRegisterPipelineResult() throws IOException {
144144
new Job(
145145
expectedJobId,
146146
"",
147-
Runner.DIRECT.name(),
147+
Runner.DIRECT,
148148
Source.fromProto(source),
149149
Store.fromProto(store),
150150
Lists.newArrayList(FeatureSet.fromProto(featureSet)),

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

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -164,7 +164,7 @@ public void shouldGenerateAndSubmitJobsIfAny() throws InvalidProtocolBufferExcep
164164
new Job(
165165
"",
166166
"",
167-
Runner.DATAFLOW.name(),
167+
Runner.DATAFLOW,
168168
feast.core.model.Source.fromProto(source),
169169
feast.core.model.Store.fromProto(store),
170170
Arrays.asList(FeatureSet.fromProto(featureSet1), FeatureSet.fromProto(featureSet2)),
@@ -174,7 +174,7 @@ public void shouldGenerateAndSubmitJobsIfAny() throws InvalidProtocolBufferExcep
174174
new Job(
175175
"some_id",
176176
extId,
177-
Runner.DATAFLOW.name(),
177+
Runner.DATAFLOW,
178178
feast.core.model.Source.fromProto(source),
179179
feast.core.model.Store.fromProto(store),
180180
Arrays.asList(FeatureSet.fromProto(featureSet1), FeatureSet.fromProto(featureSet2)),
@@ -264,7 +264,7 @@ public void shouldGroupJobsBySource() throws InvalidProtocolBufferException {
264264
new Job(
265265
"name1",
266266
"",
267-
Runner.DATAFLOW.name(),
267+
Runner.DATAFLOW,
268268
feast.core.model.Source.fromProto(source1),
269269
feast.core.model.Store.fromProto(store),
270270
Arrays.asList(FeatureSet.fromProto(featureSet1)),
@@ -274,7 +274,7 @@ public void shouldGroupJobsBySource() throws InvalidProtocolBufferException {
274274
new Job(
275275
"name1",
276276
"extId1",
277-
Runner.DATAFLOW.name(),
277+
Runner.DATAFLOW,
278278
feast.core.model.Source.fromProto(source1),
279279
feast.core.model.Store.fromProto(store),
280280
Arrays.asList(FeatureSet.fromProto(featureSet1)),
@@ -284,7 +284,7 @@ public void shouldGroupJobsBySource() throws InvalidProtocolBufferException {
284284
new Job(
285285
"",
286286
"extId2",
287-
Runner.DATAFLOW.name(),
287+
Runner.DATAFLOW,
288288
feast.core.model.Source.fromProto(source2),
289289
feast.core.model.Store.fromProto(store),
290290
Arrays.asList(FeatureSet.fromProto(featureSet2)),
@@ -294,7 +294,7 @@ public void shouldGroupJobsBySource() throws InvalidProtocolBufferException {
294294
new Job(
295295
"name2",
296296
"extId2",
297-
Runner.DATAFLOW.name(),
297+
Runner.DATAFLOW,
298298
feast.core.model.Source.fromProto(source2),
299299
feast.core.model.Store.fromProto(store),
300300
Arrays.asList(FeatureSet.fromProto(featureSet2)),

0 commit comments

Comments
 (0)