Skip to content

Add feature set status JOB_STARTING to denote feature sets waiting for job to get to RUNNING state #714

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
feast-ci-bot merged 2 commits into from
May 18, 2020
Merged

Add feature set status JOB_STARTING to denote feature sets waiting for job to get to RUNNING state #714

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f32cbb1
Add additional feature set status to differentiate between pending jo…
May 18, 2020
df083c3
Fix broken test
May 18, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 11 additions & 1 deletion core/src/main/java/feast/core/job/JobUpdateTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import feast.core.model.JobStatus;
import feast.core.model.Source;
import feast.core.model.Store;
import feast.proto.core.FeatureSetProto.FeatureSetStatus;
import java.time.Instant;
import java.util.List;
import java.util.Optional;
Expand Down Expand Up @@ -102,7 +103,16 @@ public Job call() {

boolean requiresUpdate(Job job) {
// If set of feature sets has changed
return !Sets.newHashSet(featureSets).equals(Sets.newHashSet(job.getFeatureSets()));
if (!Sets.newHashSet(featureSets).equals(Sets.newHashSet(job.getFeatureSets()))) {
return true;
}
// If any of the incoming feature sets were updated
for (FeatureSet featureSet : featureSets) {
if (featureSet.getStatus() == FeatureSetStatus.STATUS_PENDING) {
return true;
}
}
return false;
}

private Job createJob() {
Expand Down
1 change: 1 addition & 0 deletions core/src/main/java/feast/core/job/dataflow/DataflowJobManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -162,6 +162,7 @@ public Job updateJob(Job job) {
true);

job.setExtId(extId);
job.setStatus(JobStatus.PENDING);
return job;
} catch (InvalidProtocolBufferException e) {
log.error(e.getMessage());
Expand Down
6 changes: 4 additions & 2 deletions core/src/main/java/feast/core/service/JobCoordinatorService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,17 +132,19 @@ void startOrUpdateJobs(List<JobUpdateTask> tasks) {
tasks.forEach(ecs::submit);

int completedTasks = 0;
List<Job> startedJobs = new ArrayList<>();
while (completedTasks < tasks.size()) {
try {
Job job = ecs.take().get();
if (job != null) {
jobRepository.saveAndFlush(job);
startedJobs.add(job);
}
} catch (ExecutionException | InterruptedException e) {
log.warn("Unable to start or update job: {}", e.getMessage());
}
completedTasks++;
}
jobRepository.saveAll(startedJobs);
executorService.shutdown();
}

Expand All @@ -169,7 +171,7 @@ private void updateFeatureSetStatuses(List<JobUpdateTask> jobUpdateTasks) {
});
pending.forEach(
fs -> {
fs.setStatus(FeatureSetStatus.STATUS_PENDING);
fs.setStatus(FeatureSetStatus.STATUS_JOB_STARTING);
featureSetRepository.save(fs);
});
featureSetRepository.flush();
Expand Down
15 changes: 8 additions & 7 deletions core/src/test/java/feast/core/service/JobCoordinatorServiceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@
import feast.proto.core.StoreProto.Store.StoreType;
import feast.proto.core.StoreProto.Store.Subscription;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import org.junit.Before;
import org.junit.Rule;
Expand Down Expand Up @@ -149,7 +150,7 @@ public void shouldGenerateAndSubmitJobsIfAny() throws InvalidProtocolBufferExcep
.build();
FeatureSet featureSet2 = FeatureSet.fromProto(featureSetProto2);
String extId = "ext";
ArgumentCaptor<Job> jobArgCaptor = ArgumentCaptor.forClass(Job.class);
ArgumentCaptor<List<Job>> jobArgCaptor = ArgumentCaptor.forClass(List.class);

Job expectedInput =
new Job(
Expand Down Expand Up @@ -183,9 +184,9 @@ public void shouldGenerateAndSubmitJobsIfAny() throws InvalidProtocolBufferExcep
new JobCoordinatorService(
jobRepository, featureSetRepository, specService, jobManager, feastProperties);
jcs.Poll();
verify(jobRepository, times(1)).saveAndFlush(jobArgCaptor.capture());
Job actual = jobArgCaptor.getValue();
assertThat(actual, equalTo(expected));
verify(jobRepository, times(1)).saveAll(jobArgCaptor.capture());
List<Job> actual = jobArgCaptor.getValue();
assertThat(actual, equalTo(Collections.singletonList(expected)));
}

@Test
Expand Down Expand Up @@ -277,7 +278,7 @@ public void shouldGroupJobsBySource() throws InvalidProtocolBufferException {
feast.core.model.Store.fromProto(store),
Arrays.asList(featureSet2),
JobStatus.RUNNING);
ArgumentCaptor<Job> jobArgCaptor = ArgumentCaptor.forClass(Job.class);
ArgumentCaptor<List<Job>> jobArgCaptor = ArgumentCaptor.forClass(List.class);

when(featureSetRepository.findAllByNameLikeAndProject_NameLikeOrderByNameAsc("%", "project1"))
.thenReturn(Lists.newArrayList(featureSet1, featureSet2));
Expand All @@ -294,8 +295,8 @@ public void shouldGroupJobsBySource() throws InvalidProtocolBufferException {
jobRepository, featureSetRepository, specService, jobManager, feastProperties);
jcs.Poll();

verify(jobRepository, times(2)).saveAndFlush(jobArgCaptor.capture());
List<Job> actual = jobArgCaptor.getAllValues();
verify(jobRepository, times(1)).saveAll(jobArgCaptor.capture());
List<Job> actual = jobArgCaptor.getValue();

assertThat(actual.get(0), equalTo(expected1));
assertThat(actual.get(1), equalTo(expected2));
Expand Down
1 change: 1 addition & 0 deletions protos/feast/core/FeatureSet.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -147,5 +147,6 @@ message FeatureSetMeta {
enum FeatureSetStatus {
STATUS_INVALID = 0;
STATUS_PENDING = 1;

@woop woop May 18, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I prefer NOT_READY and PENDING, so pending means the job is starting. It's confusing to have pending and job_starting.

@zhilingc zhilingc May 18, 2020 •
edited
Loading

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

imo NOT_READY could mean anything that's... not READY. in this context PENDING means pending job initialization and JOB_STARTING is explicit in meaning the job is in the initializing state, which is less confusing to me

@woop woop May 18, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I disagree, this is super confusing to me.

imo NOT_READY could mean anything that's... not READY

Exactly. It's more future proof.

in this context PENDING means pending job initialization and

Then why not call it PENDING_JOB_INITIALIZATION? PENDING is meaningless.

@woop woop May 18, 2020 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also, it's a bit of a leaky abstraction to have job statuses on a feature set. Ideally these statuses would be on the job, not the feature set. Although I am not sure if that kind of change is in scope given our timelines.

@zhilingc zhilingc May 18, 2020

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think it's future proof, it's nebulous and doesn't convey any meaning to the user (or developer). I'd rather be more explicit with PENDING_JOB_INITIALIZATION in that case.

@zhilingc zhilingc May 18, 2020

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I mean, while jobs are being processed a feature set is also in the REGISTERED BUT NOT READY state, which is why I'm saying NOT_READY is ambiguous here. :/

@zhilingc zhilingc May 18, 2020

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

To add on, as long as feature sets remain the main source of information for users, it should have informative (read: actionable) statuses for whoever is querying for them, rather than masking them behind catch-alls for the sake of masking the existence of jobs from users. As a user i'd definitely prefer Feast being debuggable over being magical.

@woop woop May 18, 2020 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As a user i'd definitely prefer Feast being debuggable over being magical.

Taken to an extreme, should we open up the Feast internals like our database to our users? If not, then you agree that a line should be drawn somewhere.

That line is called abstraction, not "magic".

Being conservative in revealing system internals through your API is the safer approach, unless you want to have frequent breaking changes. We've had enough of those of late. If you'd read what I had said you'd have seen that I don't believe that job statuses should be attached to feature sets, and that this is a leaky abstraction. This whole PR and the whole way these statuses function requires attention and a rework, so leaking more details is definitely an approach I would want to avoid.

Also, the whole debugging argument falls flat when you consider that a mapping from a feature set to a job, where the job has an appropriate status, is sufficient for debugging.

No magic involved, just common sense.

@zhilingc zhilingc May 18, 2020

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nothing hinges on the fact that STATUS_JOB_STARTING is named as such, so I don't really see any problem with it introducing any APIs that could break in the future. Rather, the point of the name is to be distinct from the existing status PENDING, which NOT_READY really isn't.

And sure, users are able to query for the jobs, but the feature set should hold the first hand information as to where to look. I don't think it's common sense to look up the job if the status returned is NOT_READY.

@woop woop May 18, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the point of the name is to be distinct from the existing status PENDING, which NOT_READY really isn't.

What I was looking for in NOT_READY was a state that represented the fact that a spec was registered but not started. If NOT_READY is ambiguous for you then we can find another name (NOT_INITIALIZED?).

To be clear, this is a separate problem from the one I listed when I started this thread, which was the overlap between PENDING and JOB_STARTED.

STATUS_JOB_STARTING = 3;
STATUS_READY = 2;
}