From dc6ba0ef243ed36221f18921e9d63b01c090e070 Mon Sep 17 00:00:00 2001 From: Zhu Zhanyan Date: Fri, 27 Mar 2020 10:17:12 +0800 Subject: [PATCH 1/3] 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. --- .../java/feast/core/job/dataflow/DataflowJobManager.java | 6 +++--- .../feast/core/job/dataflow/DataflowJobManagerTest.java | 4 ++-- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/core/src/main/java/feast/core/job/dataflow/DataflowJobManager.java b/core/src/main/java/feast/core/job/dataflow/DataflowJobManager.java index 323eb35983e..4cedbc9d0b2 100644 --- a/core/src/main/java/feast/core/job/dataflow/DataflowJobManager.java +++ b/core/src/main/java/feast/core/job/dataflow/DataflowJobManager.java @@ -160,7 +160,7 @@ public void abortJob(String dataflowJobId) { */ @Override public JobStatus getJobStatus(Job job) { - if (!Runner.DATAFLOW.getName().equals(job.getRunner())) { + if (!Runner.DATAFLOW.toString().equals(job.getRunner())) { return job.getStatus(); } @@ -191,7 +191,7 @@ private Job submitDataflowJob( .map( fsp -> { FeatureSet featureSet = new FeatureSet(); - featureSet.setName(fsp.getSpec().getName()); + featureSet.setName(fsp.getSpec().toString()); featureSet.setVersion(fsp.getSpec().getVersion()); featureSet.setProject(new Project(fsp.getSpec().getProject())); return featureSet; @@ -201,7 +201,7 @@ private Job submitDataflowJob( return new Job( jobName, jobId, - getRunnerType().getName(), + getRunnerType().toString(), Source.fromProto(source), Store.fromProto(sink), featureSets, diff --git a/core/src/test/java/feast/core/job/dataflow/DataflowJobManagerTest.java b/core/src/test/java/feast/core/job/dataflow/DataflowJobManagerTest.java index 9f26c6919e4..c232f50c050 100644 --- a/core/src/test/java/feast/core/job/dataflow/DataflowJobManagerTest.java +++ b/core/src/test/java/feast/core/job/dataflow/DataflowJobManagerTest.java @@ -145,7 +145,7 @@ public void shouldStartJobWithCorrectPipelineOptions() throws IOException { new Job( jobName, "", - Runner.DATAFLOW.getName(), + Runner.DATAFLOW.toString(), Source.fromProto(source), Store.fromProto(store), Lists.newArrayList(FeatureSet.fromProto(featureSet)), @@ -226,7 +226,7 @@ public void shouldThrowExceptionWhenJobStateTerminal() throws IOException { new Job( "job", "", - Runner.DATAFLOW.getName(), + Runner.DATAFLOW.toString(), Source.fromProto(source), Store.fromProto(store), Lists.newArrayList(FeatureSet.fromProto(featureSet)), From 040780a68788db6e8b986b28ed714ce824e13a08 Mon Sep 17 00:00:00 2001 From: Zhu Zhanyan Date: Fri, 27 Mar 2020 10:39:23 +0800 Subject: [PATCH 2/3] Document how & when Runner.toString() or Runner.getName() should be used --- core/src/main/java/feast/core/job/Runner.java | 5 +++++ core/src/main/java/feast/core/model/Job.java | 1 + 2 files changed, 6 insertions(+) diff --git a/core/src/main/java/feast/core/job/Runner.java b/core/src/main/java/feast/core/job/Runner.java index 637621be359..2cbf9e3f783 100644 --- a/core/src/main/java/feast/core/job/Runner.java +++ b/core/src/main/java/feast/core/job/Runner.java @@ -27,10 +27,15 @@ public enum Runner { this.name = name; } + /** + * Get the human readable name of this runner. Returns a human readable name of the runner that + * can be used for logging/config files. + */ public String getName() { return name; } + /** Parses a runner from its human readable name. */ public static Runner fromString(String runner) { for (Runner r : Runner.values()) { if (r.getName().equals(runner)) { diff --git a/core/src/main/java/feast/core/model/Job.java b/core/src/main/java/feast/core/model/Job.java index 2fd17a58d74..6d9d5b1c464 100644 --- a/core/src/main/java/feast/core/model/Job.java +++ b/core/src/main/java/feast/core/model/Job.java @@ -39,6 +39,7 @@ public class Job extends AbstractTimestampEntity { private String extId; // Runner type + // Use Runner.toString() when converting a Runner to string to assign to this property. @Column(name = "runner") private String runner; From bb8a3cea9da65e82d02e9037d14f6a024b054329 Mon Sep 17 00:00:00 2001 From: Zhu Zhanyan Date: Fri, 27 Mar 2020 12:13:55 +0800 Subject: [PATCH 3/3] 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. --- .../java/feast/core/job/JobUpdateTask.java | 10 +++++----- core/src/main/java/feast/core/job/Runner.java | 7 ++++--- .../core/job/dataflow/DataflowJobManager.java | 4 ++-- core/src/main/java/feast/core/model/Job.java | 2 +- .../java/feast/core/job/JobUpdateTaskTest.java | 18 +++++++++--------- .../job/dataflow/DataflowJobManagerTest.java | 4 ++-- .../job/direct/DirectRunnerJobManagerTest.java | 2 +- .../service/JobCoordinatorServiceTest.java | 12 ++++++------ 8 files changed, 30 insertions(+), 29 deletions(-) diff --git a/core/src/main/java/feast/core/job/JobUpdateTask.java b/core/src/main/java/feast/core/job/JobUpdateTask.java index 87578cce25a..f3afe84df77 100644 --- a/core/src/main/java/feast/core/job/JobUpdateTask.java +++ b/core/src/main/java/feast/core/job/JobUpdateTask.java @@ -144,7 +144,7 @@ private Job startJob( new Job( jobId, "", - jobManager.getRunnerType().toString(), + jobManager.getRunnerType().name(), Source.fromProto(source), Store.fromProto(sinkSpec), featureSets, @@ -155,7 +155,7 @@ private Job startJob( jobId, Action.SUBMIT, "Building graph and submitting to %s", - jobManager.getRunnerType().getName()); + jobManager.getRunnerType().toString()); job = jobManager.startJob(job); if (job.getExtId().isEmpty()) { @@ -168,7 +168,7 @@ private Job startJob( jobId, Action.STATUS_CHANGE, "Job submitted to runner %s with ext id %s.", - jobManager.getRunnerType().getName(), + jobManager.getRunnerType().toString(), job.getExtId()); return job; @@ -179,7 +179,7 @@ private Job startJob( jobId, Action.STATUS_CHANGE, "Job failed to be submitted to runner %s. Job status changed to ERROR.", - jobManager.getRunnerType().getName()); + jobManager.getRunnerType().toString()); job.setStatus(JobStatus.ERROR); return job; @@ -206,7 +206,7 @@ private Job updateJob( Action.UPDATE, "Updating job %s for runner %s", job.getId(), - jobManager.getRunnerType().getName()); + jobManager.getRunnerType().toString()); return jobManager.updateJob(job); } diff --git a/core/src/main/java/feast/core/job/Runner.java b/core/src/main/java/feast/core/job/Runner.java index 2cbf9e3f783..4e2033fed69 100644 --- a/core/src/main/java/feast/core/job/Runner.java +++ b/core/src/main/java/feast/core/job/Runner.java @@ -29,16 +29,17 @@ public enum Runner { /** * Get the human readable name of this runner. Returns a human readable name of the runner that - * can be used for logging/config files. + * can be used for logging/config files/etc. */ - public String getName() { + @Override + public String toString() { return name; } /** Parses a runner from its human readable name. */ public static Runner fromString(String runner) { for (Runner r : Runner.values()) { - if (r.getName().equals(runner)) { + if (r.toString().equals(runner)) { return r; } } diff --git a/core/src/main/java/feast/core/job/dataflow/DataflowJobManager.java b/core/src/main/java/feast/core/job/dataflow/DataflowJobManager.java index 4cedbc9d0b2..e76568dfb48 100644 --- a/core/src/main/java/feast/core/job/dataflow/DataflowJobManager.java +++ b/core/src/main/java/feast/core/job/dataflow/DataflowJobManager.java @@ -160,7 +160,7 @@ public void abortJob(String dataflowJobId) { */ @Override public JobStatus getJobStatus(Job job) { - if (!Runner.DATAFLOW.toString().equals(job.getRunner())) { + if (!Runner.DATAFLOW.name().equals(job.getRunner())) { return job.getStatus(); } @@ -201,7 +201,7 @@ private Job submitDataflowJob( return new Job( jobName, jobId, - getRunnerType().toString(), + getRunnerType().name(), Source.fromProto(source), Store.fromProto(sink), featureSets, diff --git a/core/src/main/java/feast/core/model/Job.java b/core/src/main/java/feast/core/model/Job.java index 6d9d5b1c464..377f5f70956 100644 --- a/core/src/main/java/feast/core/model/Job.java +++ b/core/src/main/java/feast/core/model/Job.java @@ -39,7 +39,7 @@ public class Job extends AbstractTimestampEntity { private String extId; // Runner type - // Use Runner.toString() when converting a Runner to string to assign to this property. + // Use Runner.name() when converting a Runner to string to assign to this property. @Column(name = "runner") private String runner; diff --git a/core/src/test/java/feast/core/job/JobUpdateTaskTest.java b/core/src/test/java/feast/core/job/JobUpdateTaskTest.java index 19ce0858b20..2a1e80994ae 100644 --- a/core/src/test/java/feast/core/job/JobUpdateTaskTest.java +++ b/core/src/test/java/feast/core/job/JobUpdateTaskTest.java @@ -102,7 +102,7 @@ public void shouldUpdateJobIfPresent() { new Job( "job", "old_ext", - Runner.DATAFLOW.getName(), + Runner.DATAFLOW.name(), feast.core.model.Source.fromProto(source), feast.core.model.Store.fromProto(store), Arrays.asList(FeatureSet.fromProto(featureSet1)), @@ -119,7 +119,7 @@ public void shouldUpdateJobIfPresent() { new Job( "job", "old_ext", - Runner.DATAFLOW.getName(), + Runner.DATAFLOW.name(), feast.core.model.Source.fromProto(source), feast.core.model.Store.fromProto(store), Arrays.asList(FeatureSet.fromProto(featureSet1), FeatureSet.fromProto(featureSet2)), @@ -129,7 +129,7 @@ public void shouldUpdateJobIfPresent() { new Job( "job", "new_ext", - Runner.DATAFLOW.getName(), + Runner.DATAFLOW.name(), Source.fromProto(source), Store.fromProto(store), Arrays.asList(FeatureSet.fromProto(featureSet1), FeatureSet.fromProto(featureSet2)), @@ -163,7 +163,7 @@ public void shouldCreateJobIfNotPresent() { new Job( "job", "", - Runner.DATAFLOW.getName(), + Runner.DATAFLOW.name(), feast.core.model.Source.fromProto(source), feast.core.model.Store.fromProto(store), Arrays.asList(FeatureSet.fromProto(featureSet1)), @@ -173,7 +173,7 @@ public void shouldCreateJobIfNotPresent() { new Job( "job", "ext", - Runner.DATAFLOW.getName(), + Runner.DATAFLOW.name(), feast.core.model.Source.fromProto(source), feast.core.model.Store.fromProto(store), Arrays.asList(FeatureSet.fromProto(featureSet1)), @@ -202,7 +202,7 @@ public void shouldUpdateJobStatusIfNotCreateOrUpdate() { new Job( "job", "ext", - Runner.DATAFLOW.getName(), + Runner.DATAFLOW.name(), feast.core.model.Source.fromProto(source), feast.core.model.Store.fromProto(store), Arrays.asList(FeatureSet.fromProto(featureSet1)), @@ -216,7 +216,7 @@ public void shouldUpdateJobStatusIfNotCreateOrUpdate() { new Job( "job", "ext", - Runner.DATAFLOW.getName(), + Runner.DATAFLOW.name(), Source.fromProto(source), Store.fromProto(store), Arrays.asList(FeatureSet.fromProto(featureSet1)), @@ -248,7 +248,7 @@ public void shouldReturnJobWithErrorStatusIfFailedToSubmit() { new Job( "job", "", - Runner.DATAFLOW.getName(), + Runner.DATAFLOW.name(), feast.core.model.Source.fromProto(source), feast.core.model.Store.fromProto(store), Arrays.asList(FeatureSet.fromProto(featureSet1)), @@ -258,7 +258,7 @@ public void shouldReturnJobWithErrorStatusIfFailedToSubmit() { new Job( "job", "", - Runner.DATAFLOW.getName(), + Runner.DATAFLOW.name(), feast.core.model.Source.fromProto(source), feast.core.model.Store.fromProto(store), Arrays.asList(FeatureSet.fromProto(featureSet1)), diff --git a/core/src/test/java/feast/core/job/dataflow/DataflowJobManagerTest.java b/core/src/test/java/feast/core/job/dataflow/DataflowJobManagerTest.java index c232f50c050..2d562d38df2 100644 --- a/core/src/test/java/feast/core/job/dataflow/DataflowJobManagerTest.java +++ b/core/src/test/java/feast/core/job/dataflow/DataflowJobManagerTest.java @@ -145,7 +145,7 @@ public void shouldStartJobWithCorrectPipelineOptions() throws IOException { new Job( jobName, "", - Runner.DATAFLOW.toString(), + Runner.DATAFLOW.name(), Source.fromProto(source), Store.fromProto(store), Lists.newArrayList(FeatureSet.fromProto(featureSet)), @@ -226,7 +226,7 @@ public void shouldThrowExceptionWhenJobStateTerminal() throws IOException { new Job( "job", "", - Runner.DATAFLOW.toString(), + Runner.DATAFLOW.name(), Source.fromProto(source), Store.fromProto(store), Lists.newArrayList(FeatureSet.fromProto(featureSet)), diff --git a/core/src/test/java/feast/core/job/direct/DirectRunnerJobManagerTest.java b/core/src/test/java/feast/core/job/direct/DirectRunnerJobManagerTest.java index 64412f4391e..76530d9f404 100644 --- a/core/src/test/java/feast/core/job/direct/DirectRunnerJobManagerTest.java +++ b/core/src/test/java/feast/core/job/direct/DirectRunnerJobManagerTest.java @@ -144,7 +144,7 @@ public void shouldStartDirectJobAndRegisterPipelineResult() throws IOException { new Job( expectedJobId, "", - Runner.DIRECT.getName(), + Runner.DIRECT.name(), Source.fromProto(source), Store.fromProto(store), Lists.newArrayList(FeatureSet.fromProto(featureSet)), diff --git a/core/src/test/java/feast/core/service/JobCoordinatorServiceTest.java b/core/src/test/java/feast/core/service/JobCoordinatorServiceTest.java index 67a87e93167..aa71f201dde 100644 --- a/core/src/test/java/feast/core/service/JobCoordinatorServiceTest.java +++ b/core/src/test/java/feast/core/service/JobCoordinatorServiceTest.java @@ -161,7 +161,7 @@ public void shouldGenerateAndSubmitJobsIfAny() throws InvalidProtocolBufferExcep new Job( "", "", - Runner.DATAFLOW.getName(), + Runner.DATAFLOW.name(), feast.core.model.Source.fromProto(source), feast.core.model.Store.fromProto(store), Arrays.asList(FeatureSet.fromProto(featureSet1), FeatureSet.fromProto(featureSet2)), @@ -171,7 +171,7 @@ public void shouldGenerateAndSubmitJobsIfAny() throws InvalidProtocolBufferExcep new Job( "some_id", extId, - Runner.DATAFLOW.getName(), + Runner.DATAFLOW.name(), feast.core.model.Source.fromProto(source), feast.core.model.Store.fromProto(store), Arrays.asList(FeatureSet.fromProto(featureSet1), FeatureSet.fromProto(featureSet2)), @@ -261,7 +261,7 @@ public void shouldGroupJobsBySource() throws InvalidProtocolBufferException { new Job( "name1", "", - Runner.DATAFLOW.getName(), + Runner.DATAFLOW.name(), feast.core.model.Source.fromProto(source1), feast.core.model.Store.fromProto(store), Arrays.asList(FeatureSet.fromProto(featureSet1)), @@ -271,7 +271,7 @@ public void shouldGroupJobsBySource() throws InvalidProtocolBufferException { new Job( "name1", "extId1", - Runner.DATAFLOW.getName(), + Runner.DATAFLOW.name(), feast.core.model.Source.fromProto(source1), feast.core.model.Store.fromProto(store), Arrays.asList(FeatureSet.fromProto(featureSet1)), @@ -281,7 +281,7 @@ public void shouldGroupJobsBySource() throws InvalidProtocolBufferException { new Job( "", "extId2", - Runner.DATAFLOW.getName(), + Runner.DATAFLOW.name(), feast.core.model.Source.fromProto(source2), feast.core.model.Store.fromProto(store), Arrays.asList(FeatureSet.fromProto(featureSet2)), @@ -291,7 +291,7 @@ public void shouldGroupJobsBySource() throws InvalidProtocolBufferException { new Job( "name2", "extId2", - Runner.DATAFLOW.getName(), + Runner.DATAFLOW.name(), feast.core.model.Source.fromProto(source2), feast.core.model.Store.fromProto(store), Arrays.asList(FeatureSet.fromProto(featureSet2)),