Skip to content

Commit 7903dc7

Browse files
chesches
committed
core: Simplify JobUpdateTask internal helper calls
The helper methods like startJob and updateJob had parameters that are instance state, so there seemed to be no reason to pass around arguments for them.
1 parent 2ca3308 commit 7903dc7

1 file changed

Lines changed: 42 additions & 33 deletions

File tree

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

Lines changed: 42 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -80,61 +80,59 @@ public JobUpdateTask(
8080
@Override
8181
public Job call() {
8282
ExecutorService executorService = Executors.newSingleThreadExecutor();
83-
Source source = Source.fromProto(sourceSpec);
8483
Future<Job> submittedJob;
8584

86-
if (currentJob.isPresent()) {
85+
if (currentJob.isEmpty()) {
86+
submittedJob = executorService.submit(this::createJob);
87+
} else {
8788
Job job = currentJob.get();
88-
Set<String> existingFeatureSetsPopulatedByJob =
89-
job.getFeatureSets().stream().map(FeatureSet::getId).collect(Collectors.toSet());
90-
Set<String> newFeatureSetsPopulatedByJob =
91-
featureSets.stream()
92-
.map(fs -> FeatureSet.fromProto(fs).getId())
93-
.collect(Collectors.toSet());
94-
95-
if (newFeatureSetsPopulatedByJob.equals(existingFeatureSetsPopulatedByJob)) {
96-
JobStatus currentStatus = job.getStatus();
97-
JobStatus newStatus = jobManager.getJobStatus(job);
98-
if (newStatus != currentStatus) {
99-
var auditMessage = "Job status updated: changed from %s to %s";
100-
logAudit(Action.STATUS_CHANGE, job, auditMessage, currentStatus, newStatus);
101-
}
102-
103-
job.setStatus(newStatus);
104-
return job;
89+
90+
if (featureSetsChangedFor(job)) {
91+
submittedJob = executorService.submit(() -> updateJob(job));
10592
} else {
106-
submittedJob = executorService.submit(() -> updateJob(job, featureSets, store));
93+
return updateStatus(job);
10794
}
108-
} else {
109-
String jobId = createJobId(source.getId(), store.getName());
110-
submittedJob = executorService.submit(() -> startJob(jobId, featureSets, sourceSpec, store));
11195
}
11296

11397
try {
11498
return submittedJob.get(getJobUpdateTimeoutSeconds(), TimeUnit.SECONDS);
11599
} catch (InterruptedException | ExecutionException | TimeoutException e) {
100+
Source source = Source.fromProto(sourceSpec);
116101
log.warn("Unable to start job for source {} and sink {}: {}", source, store, e.getMessage());
117102
return null;
118103
} finally {
119104
executorService.shutdownNow();
120105
}
121106
}
122107

108+
boolean featureSetsChangedFor(Job job) {
109+
Set<String> existingFeatureSetsPopulatedByJob =
110+
job.getFeatureSets().stream().map(FeatureSet::getId).collect(Collectors.toSet());
111+
Set<String> newFeatureSetsPopulatedByJob =
112+
featureSets.stream()
113+
.map(fs -> FeatureSet.fromProto(fs).getId())
114+
.collect(Collectors.toSet());
115+
116+
return !newFeatureSetsPopulatedByJob.equals(existingFeatureSetsPopulatedByJob);
117+
}
118+
119+
private Job createJob() {
120+
Source source = Source.fromProto(sourceSpec);
121+
String jobId = createJobId(source.getId(), store.getName());
122+
return startJob(jobId);
123+
}
124+
123125
/** Start or update the job to ingest data to the sink. */
124-
private Job startJob(
125-
String jobId,
126-
List<FeatureSetProto.FeatureSet> featureSetProtos,
127-
SourceProto.Source source,
128-
StoreProto.Store sinkSpec) {
126+
private Job startJob(String jobId) {
129127

130128
Job job =
131129
new Job(
132130
jobId,
133131
"",
134132
jobManager.getRunnerType().name(),
135-
Source.fromProto(source),
136-
Store.fromProto(sinkSpec),
137-
featureSetsFromProto(featureSetProtos),
133+
Source.fromProto(sourceSpec),
134+
Store.fromProto(store),
135+
featureSetsFromProto(featureSets),
138136
JobStatus.PENDING);
139137
try {
140138
logAudit(Action.SUBMIT, job, "Building graph and submitting to %s", runnerType);
@@ -161,14 +159,25 @@ private Job startJob(
161159
}
162160

163161
/** Update the given job */
164-
private Job updateJob(
165-
Job job, List<FeatureSetProto.FeatureSet> featureSets, StoreProto.Store store) {
162+
private Job updateJob(Job job) {
166163
job.setFeatureSets(featureSetsFromProto(featureSets));
167164
job.setStore(Store.fromProto(store));
168165
logAudit(Action.UPDATE, job, "Updating job %s for runner %s", job.getId(), runnerType);
169166
return jobManager.updateJob(job);
170167
}
171168

169+
private Job updateStatus(Job job) {
170+
JobStatus currentStatus = job.getStatus();
171+
JobStatus newStatus = jobManager.getJobStatus(job);
172+
if (newStatus != currentStatus) {
173+
var auditMessage = "Job status updated: changed from %s to %s";
174+
logAudit(Action.STATUS_CHANGE, job, auditMessage, currentStatus, newStatus);
175+
}
176+
177+
job.setStatus(newStatus);
178+
return job;
179+
}
180+
172181
String createJobId(String sourceId, String storeName) {
173182
String dateSuffix = String.valueOf(Instant.now().toEpochMilli());
174183
String sourceIdTrunc = sourceId.split("/")[0].toLowerCase();

0 commit comments

Comments
 (0)