Skip to content

Commit 901e260

Browse files
mrzzyZhu Zhanyan
mrzzy
and
Zhu Zhanyan
authored
Fix runner to string inconsistency (#575)
* Changed Runner.getName() to Runner.toString() when passing to Job.runner in DataflowJobManager This is necessary to standardise the use of Runner.toString() when passing to the Job.runner, so that code dependending on Job.runner would know what to expect. * Document how & when Runner.toString() or Runner.getName() should be used * Convert getName() to toString(). Use name() for Job.runner. Use toString() to render human readable strings while using the non overriding name() for code dependencies. Co-authored-by: Zhu Zhanyan <zhu.zhanyan@gojek.com>
1 parent 9facd2e commit 901e260

8 files changed

Lines changed: 35 additions & 28 deletions

File tree

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

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -144,7 +144,7 @@ private Job startJob(
144144
new Job(
145145
jobId,
146146
"",
147-
jobManager.getRunnerType().toString(),
147+
jobManager.getRunnerType().name(),
148148
Source.fromProto(source),
149149
Store.fromProto(sinkSpec),
150150
featureSets,
@@ -155,7 +155,7 @@ private Job startJob(
155155
jobId,
156156
Action.SUBMIT,
157157
"Building graph and submitting to %s",
158-
jobManager.getRunnerType().getName());
158+
jobManager.getRunnerType().toString());
159159

160160
job = jobManager.startJob(job);
161161
if (job.getExtId().isEmpty()) {
@@ -168,7 +168,7 @@ private Job startJob(
168168
jobId,
169169
Action.STATUS_CHANGE,
170170
"Job submitted to runner %s with ext id %s.",
171-
jobManager.getRunnerType().getName(),
171+
jobManager.getRunnerType().toString(),
172172
job.getExtId());
173173

174174
return job;
@@ -179,7 +179,7 @@ private Job startJob(
179179
jobId,
180180
Action.STATUS_CHANGE,
181181
"Job failed to be submitted to runner %s. Job status changed to ERROR.",
182-
jobManager.getRunnerType().getName());
182+
jobManager.getRunnerType().toString());
183183

184184
job.setStatus(JobStatus.ERROR);
185185
return job;
@@ -206,7 +206,7 @@ private Job updateJob(
206206
Action.UPDATE,
207207
"Updating job %s for runner %s",
208208
job.getId(),
209-
jobManager.getRunnerType().getName());
209+
jobManager.getRunnerType().toString());
210210
return jobManager.updateJob(job);
211211
}
212212

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

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -27,13 +27,19 @@ public enum Runner {
2727
this.name = name;
2828
}
2929

30-
public String getName() {
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+
*/
34+
@Override
35+
public String toString() {
3136
return name;
3237
}
3338

39+
/** Parses a runner from its human readable name. */
3440
public static Runner fromString(String runner) {
3541
for (Runner r : Runner.values()) {
36-
if (r.getName().equals(runner)) {
42+
if (r.toString().equals(runner)) {
3743
return r;
3844
}
3945
}

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

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -160,7 +160,7 @@ public void abortJob(String dataflowJobId) {
160160
*/
161161
@Override
162162
public JobStatus getJobStatus(Job job) {
163-
if (!Runner.DATAFLOW.getName().equals(job.getRunner())) {
163+
if (!Runner.DATAFLOW.name().equals(job.getRunner())) {
164164
return job.getStatus();
165165
}
166166

@@ -191,7 +191,7 @@ private Job submitDataflowJob(
191191
.map(
192192
fsp -> {
193193
FeatureSet featureSet = new FeatureSet();
194-
featureSet.setName(fsp.getSpec().getName());
194+
featureSet.setName(fsp.getSpec().toString());
195195
featureSet.setVersion(fsp.getSpec().getVersion());
196196
featureSet.setProject(new Project(fsp.getSpec().getProject()));
197197
return featureSet;
@@ -201,7 +201,7 @@ private Job submitDataflowJob(
201201
return new Job(
202202
jobName,
203203
jobId,
204-
getRunnerType().getName(),
204+
getRunnerType().name(),
205205
Source.fromProto(source),
206206
Store.fromProto(sink),
207207
featureSets,

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@ public class Job extends AbstractTimestampEntity {
3939
private String extId;
4040

4141
// Runner type
42+
// Use Runner.name() when converting a Runner to string to assign to this property.
4243
@Column(name = "runner")
4344
private String runner;
4445

‎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.getName(),
105+
Runner.DATAFLOW.name(),
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.getName(),
122+
Runner.DATAFLOW.name(),
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.getName(),
132+
Runner.DATAFLOW.name(),
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.getName(),
166+
Runner.DATAFLOW.name(),
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.getName(),
176+
Runner.DATAFLOW.name(),
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.getName(),
205+
Runner.DATAFLOW.name(),
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.getName(),
219+
Runner.DATAFLOW.name(),
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.getName(),
251+
Runner.DATAFLOW.name(),
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.getName(),
261+
Runner.DATAFLOW.name(),
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/dataflow/DataflowJobManagerTest.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -145,7 +145,7 @@ public void shouldStartJobWithCorrectPipelineOptions() throws IOException {
145145
new Job(
146146
jobName,
147147
"",
148-
Runner.DATAFLOW.getName(),
148+
Runner.DATAFLOW.name(),
149149
Source.fromProto(source),
150150
Store.fromProto(store),
151151
Lists.newArrayList(FeatureSet.fromProto(featureSet)),
@@ -226,7 +226,7 @@ public void shouldThrowExceptionWhenJobStateTerminal() throws IOException {
226226
new Job(
227227
"job",
228228
"",
229-
Runner.DATAFLOW.getName(),
229+
Runner.DATAFLOW.name(),
230230
Source.fromProto(source),
231231
Store.fromProto(store),
232232
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.getName(),
147+
Runner.DIRECT.name(),
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
@@ -161,7 +161,7 @@ public void shouldGenerateAndSubmitJobsIfAny() throws InvalidProtocolBufferExcep
161161
new Job(
162162
"",
163163
"",
164-
Runner.DATAFLOW.getName(),
164+
Runner.DATAFLOW.name(),
165165
feast.core.model.Source.fromProto(source),
166166
feast.core.model.Store.fromProto(store),
167167
Arrays.asList(FeatureSet.fromProto(featureSet1), FeatureSet.fromProto(featureSet2)),
@@ -171,7 +171,7 @@ public void shouldGenerateAndSubmitJobsIfAny() throws InvalidProtocolBufferExcep
171171
new Job(
172172
"some_id",
173173
extId,
174-
Runner.DATAFLOW.getName(),
174+
Runner.DATAFLOW.name(),
175175
feast.core.model.Source.fromProto(source),
176176
feast.core.model.Store.fromProto(store),
177177
Arrays.asList(FeatureSet.fromProto(featureSet1), FeatureSet.fromProto(featureSet2)),
@@ -261,7 +261,7 @@ public void shouldGroupJobsBySource() throws InvalidProtocolBufferException {
261261
new Job(
262262
"name1",
263263
"",
264-
Runner.DATAFLOW.getName(),
264+
Runner.DATAFLOW.name(),
265265
feast.core.model.Source.fromProto(source1),
266266
feast.core.model.Store.fromProto(store),
267267
Arrays.asList(FeatureSet.fromProto(featureSet1)),
@@ -271,7 +271,7 @@ public void shouldGroupJobsBySource() throws InvalidProtocolBufferException {
271271
new Job(
272272
"name1",
273273
"extId1",
274-
Runner.DATAFLOW.getName(),
274+
Runner.DATAFLOW.name(),
275275
feast.core.model.Source.fromProto(source1),
276276
feast.core.model.Store.fromProto(store),
277277
Arrays.asList(FeatureSet.fromProto(featureSet1)),
@@ -281,7 +281,7 @@ public void shouldGroupJobsBySource() throws InvalidProtocolBufferException {
281281
new Job(
282282
"",
283283
"extId2",
284-
Runner.DATAFLOW.getName(),
284+
Runner.DATAFLOW.name(),
285285
feast.core.model.Source.fromProto(source2),
286286
feast.core.model.Store.fromProto(store),
287287
Arrays.asList(FeatureSet.fromProto(featureSet2)),
@@ -291,7 +291,7 @@ public void shouldGroupJobsBySource() throws InvalidProtocolBufferException {
291291
new Job(
292292
"name2",
293293
"extId2",
294-
Runner.DATAFLOW.getName(),
294+
Runner.DATAFLOW.name(),
295295
feast.core.model.Source.fromProto(source2),
296296
feast.core.model.Store.fromProto(store),
297297
Arrays.asList(FeatureSet.fromProto(featureSet2)),

0 commit comments

Comments
 (0)