Modified handling the job events for better coordination #430
Conversation
ajoymajumdar
commented
Nov 3, 2016
ajoymajumdar
commented
- Modified to trigger the job launch when handling the JobScheduledEvent. Earlier there was a possibility of a race condition.
- Use one concurrent map of JobInfo to track the jobs on a local instance. Synchronize on one job instead of all the jobs.
- On job kill/interruption, stop running the rest of the tasks in the LocalJobRunner.
| * @throws Exception If there is any problem. | ||
| */ | ||
| @Test | ||
| public void submitAndKillJob() throws Exception { |
There was a problem hiding this comment.
Don't we have a test for killing the job in the JobRestControllerIntegrationTests?
| @@ -32,25 +36,37 @@ | |||
| @Getter | |||
| public class JobScheduledEvent extends BaseJobEvent { | |||
There was a problem hiding this comment.
So we talked about renaming this cause not it's more of a "schedule a job event" and that's something that might be better as a direct method call to a service than an event? The events are intended to be listened and acted on by multiple listeners that are idempotent. This seems more like a direct action that needs to be taken.
There was a problem hiding this comment.
Yes, you are right. I too wanted to have this directly called. Before this is called an init method should be called. So init, scheduled, started and finished methods. Should I make the change?
| } | ||
| // Tell the system a new job has been scheduled so any actions can be taken | ||
| log.info("Publishing job scheduled event for job {}", jobId); | ||
| this.eventPublisher.publishEvent(new JobScheduledEvent(jobId, jobRequest, |
There was a problem hiding this comment.
What happens now if the scheduling causes an exception?
There was a problem hiding this comment.
We update the job status to Failed but we do not raise JobFinishedEvent event.
|
|
||
| // Automatically track the number of jobs running on this node | ||
| this.registry.mapSize("genie.jobs.running.gauge", this.jobMonitors); | ||
| this.registry.mapSize("genie.jobs.scheduled.gauge", this.scheduledJobs); | ||
| this.registry.mapSize("genie.jobs.running.gauge", this.jobs); | ||
| this.registry.methodValue("genie.jobs.active.gauge", this, "getNumActiveJobs"); |
There was a problem hiding this comment.
maybe it's better to tag them with the status so that we can break them up that way?
| public void onJobScheduled(final JobScheduledEvent event) { | ||
| final String jobId = event.getId(); | ||
| jobs.put(jobId, new JobInfo()); | ||
| handle(event.getId(), () -> { |
There was a problem hiding this comment.
oh fancy functional programming haha
| } | ||
| } | ||
| } catch (GenieException e) { | ||
| jobStateService.done(jobId); |
There was a problem hiding this comment.
Is there a reason we don't have the persistence steps done within the state machine? It seems to me they're tightly coupled no?
There was a problem hiding this comment.
It does makes sense but wanted to keep it separate to being with. First draft to be in-memory. This should actually have all the job related persistence steps done from within here.
| @@ -781,7 +781,9 @@ private void copyRequestHeaders(final HttpServletRequest request, final ClientHt | |||
| private void copyResponseHeaders(final HttpServletResponse response, final ClientHttpResponse forwardResponse) { | |||
| final HttpHeaders headers = forwardResponse.getHeaders(); | |||
| for (final Map.Entry<String, String> header : headers.toSingleValueMap().entrySet()) { | |||
| response.setHeader(header.getKey(), header.getValue()); | |||
| if (!TRANSFER_ENCODING_HEADER.equalsIgnoreCase(header.getKey())) { | |||
There was a problem hiding this comment.
I would add a comment here why you're doing this so that no one takes it out by accident going forward
…vent. Earlier there was a possibility of a race condition. 2. Use one concurrent map of JobInfo to track the jobs on a local instance. Synchronize on one job instead of all the jobs. 3.On job kill/interruption, stop running the rest of the tasks in the LocalJobRunner.
…vent. Earlier there was a possibility of a race condition. 2. Use one concurrent map of JobInfo to track the jobs on a local instance. Synchronize on one job instead of all the jobs. 3.On job kill/interruption, stop running the rest of the tasks in the LocalJobRunner.
…vent. Earlier there was a possibility of a race condition. 2. Use one concurrent map of JobInfo to track the jobs on a local instance. Synchronize on one job instead of all the jobs. 3.On job kill/interruption, stop running the rest of the tasks in the LocalJobRunner.
…vent. Earlier there was a possibility of a race condition. 2. Use one concurrent map of JobInfo to track the jobs on a local instance. Synchronize on one job instead of all the jobs. 3.On job kill/interruption, stop running the rest of the tasks in the LocalJobRunner.
