You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@gobblin.apache.org by "arjun4084346 (via GitHub)" <gi...@apache.org> on 2024/08/02 00:59:59 UTC
[PR] keep the job status as FAILED even if job should be retried, when dag… [gobblin]
arjun4084346 opened a new pull request, #4013:
URL: https://github.com/apache/gobblin/pull/4013
…ProcEngine is enabled.
Dear Gobblin maintainers,
Please accept this PR. I understand that it will not be reviewed until I have checked off all the steps below!
### JIRA
- [ ] My PR addresses the following [Gobblin JIRA](https://issues.apache.org/jira/browse/GOBBLIN/) issues and references them in the PR title. For example, "[GOBBLIN-XXX] My Gobblin PR"
- https://issues.apache.org/jira/browse/GOBBLIN-XXX
### Description
- [ ] Here are some details about my PR, including screenshots (if applicable):
### Tests
- [ ] My PR adds the following unit tests __OR__ does not need testing for this extremely good reason:
### Commits
- [ ] My commits all reference JIRA issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)":
1. Subject is separated from body by a blank line
2. Subject is limited to 50 characters
3. Subject does not end with a period
4. Subject uses the imperative mood ("add", not "adding")
5. Body wraps at 72 characters
6. Body explains "what" and "why", not "how"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2123] create reevaluate dag action for jobs in state PENDING_RETRY [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #4013:
URL: https://github.com/apache/gobblin/pull/4013#discussion_r1702454087
##########
gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/FlowStatusGenerator.java:
##########
@@ -41,7 +41,7 @@
@Slf4j
public class FlowStatusGenerator {
public static final List<String> FINISHED_STATUSES = Lists.newArrayList(ExecutionStatus.FAILED.name(),
- ExecutionStatus.COMPLETE.name(), ExecutionStatus.CANCELLED.name());
+ ExecutionStatus.COMPLETE.name(), ExecutionStatus.CANCELLED.name(), ExecutionStatus.PENDING_RETRY.name());
Review Comment:
Yea, we can actually make it work without PENDING_RETRY. That was the approach in the 1st commit (sorry, it lost because I mistakenly force pushed another squashed commit)
We can completely remove PENDING_RETRY. But in that approach, job's status will be shown 'FAILED' between both the attempts,, Kip preferred 'PENDING_RETRY'...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] keep the job status as FAILED even if job should be retried, when dag… [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #4013:
URL: https://github.com/apache/gobblin/pull/4013#discussion_r1702053662
##########
gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/FlowStatusGenerator.java:
##########
@@ -41,7 +41,7 @@
@Slf4j
public class FlowStatusGenerator {
public static final List<String> FINISHED_STATUSES = Lists.newArrayList(ExecutionStatus.FAILED.name(),
- ExecutionStatus.COMPLETE.name(), ExecutionStatus.CANCELLED.name());
+ ExecutionStatus.COMPLETE.name(), ExecutionStatus.CANCELLED.name(), ExecutionStatus.PENDING_RETRY.name());
Review Comment:
The code is like
```
if (jobStatus.isShouldRetry()) {
XXX
} else if (!dagManagementStateStore.hasRunningJobs(getDagId())) {
YYY
}
```
so both code parts do not collide. Also it is kinda ok for `hasRunningJobs` to return `true` when status is PENDING_RETRY, isn't it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] keep the job status as FAILED even if job should be retried, when dag… [gobblin]
Posted by "phet (via GitHub)" <gi...@apache.org>.
phet commented on code in PR #4013:
URL: https://github.com/apache/gobblin/pull/4013#discussion_r1701616940
##########
gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/FlowStatusGenerator.java:
##########
@@ -41,7 +41,7 @@
@Slf4j
public class FlowStatusGenerator {
public static final List<String> FINISHED_STATUSES = Lists.newArrayList(ExecutionStatus.FAILED.name(),
- ExecutionStatus.COMPLETE.name(), ExecutionStatus.CANCELLED.name());
+ ExecutionStatus.COMPLETE.name(), ExecutionStatus.CANCELLED.name(), ExecutionStatus.PENDING_RETRY.name());
Review Comment:
this is used in a number of places. is it acceptable to change the semantics for all of them? e.g. is it OK here in `MysqlDMStateStore`:
```
public boolean hasRunningJobs(DagManager.DagId dagId) throws IOException {
return getDagNodes(dagId).stream()
.anyMatch(node -> !FlowStatusGenerator.FINISHED_STATUSES.contains(node.getValue().getExecutionStatus().name()));
}
```
?
couldn't it lead to this in `DagManager`:
```
default:
log.warn("Unexpected flow event {} for dag {}", dag.getFlowEvent(), dagId);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] keep the job status as FAILED even if job should be retried, when dag… [gobblin]
Posted by "phet (via GitHub)" <gi...@apache.org>.
phet commented on code in PR #4013:
URL: https://github.com/apache/gobblin/pull/4013#discussion_r1701618512
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/spec/JobExecutionPlanListDeserializer.java:
##########
@@ -110,6 +110,8 @@ public List<JobExecutionPlan> deserialize(JsonElement json, Type typeOfT, JsonDe
JobExecutionPlan jobExecutionPlan = new JobExecutionPlan(jobSpec, specExecutor);
jobExecutionPlan.setExecutionStatus(executionStatus);
+ jobExecutionPlan.setCurrentGeneration(serializedJobExecutionPlan.get(SerializationConstants.CURRENT_GENERATION_KEY).getAsInt());
+ jobExecutionPlan.setCurrentAttempts(serializedJobExecutionPlan.get(SerializationConstants.CURRENT_ATTEMPTS_KEY).getAsInt());
Review Comment:
on a meta level, this PR would be more cohesive as two, one for PENDING_RETRY and the other for (de)serialization
##########
gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/FlowStatusGenerator.java:
##########
@@ -41,7 +41,7 @@
@Slf4j
public class FlowStatusGenerator {
public static final List<String> FINISHED_STATUSES = Lists.newArrayList(ExecutionStatus.FAILED.name(),
- ExecutionStatus.COMPLETE.name(), ExecutionStatus.CANCELLED.name());
+ ExecutionStatus.COMPLETE.name(), ExecutionStatus.CANCELLED.name(), ExecutionStatus.PENDING_RETRY.name());
Review Comment:
this is used in a number of places. is it acceptable to change the semantics for all of them? e.g. is it OK here in `MysqlDMStateStore`:
```
public boolean hasRunningJobs(DagManager.DagId dagId) throws IOException {
return getDagNodes(dagId).stream()
.anyMatch(node -> !FlowStatusGenerator.FINISHED_STATUSES.contains(node.getValue().getExecutionStatus().name()));
}
```
?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] keep the job status as FAILED even if job should be retried, when dag… [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #4013:
URL: https://github.com/apache/gobblin/pull/4013#discussion_r1702090447
##########
gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/FlowStatusGenerator.java:
##########
@@ -41,7 +41,7 @@
@Slf4j
public class FlowStatusGenerator {
public static final List<String> FINISHED_STATUSES = Lists.newArrayList(ExecutionStatus.FAILED.name(),
- ExecutionStatus.COMPLETE.name(), ExecutionStatus.CANCELLED.name());
+ ExecutionStatus.COMPLETE.name(), ExecutionStatus.CANCELLED.name(), ExecutionStatus.PENDING_RETRY.name());
Review Comment:
hmm, actually it should not go to that warn message, because `flowEvent` field is never set to PENDING_RETRY. in fact PENDING_RETRY does not correspond to any "flow event", this is a case when GTE name does not have any execution status
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2123] create reevaluate dag action for jobs in state PENDING_RETRY [gobblin]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #4013:
URL: https://github.com/apache/gobblin/pull/4013#issuecomment-2266253772
## [Codecov](https://app.codecov.io/gh/apache/gobblin/pull/4013?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 55.34%. Comparing base [(`2ea9622`)](https://app.codecov.io/gh/apache/gobblin/commit/2ea96226cedbc8286952029101d8acb82cecadc7?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`c262498`)](https://app.codecov.io/gh/apache/gobblin/commit/c2624982b07029aa0446972badebe36e9506cee4?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 4 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #4013 +/- ##
============================================
+ Coverage 55.20% 55.34% +0.13%
- Complexity 1650 1657 +7
============================================
Files 319 319
Lines 11808 11830 +22
Branches 1205 1207 +2
============================================
+ Hits 6519 6547 +28
+ Misses 4710 4704 -6
Partials 579 579
```
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/gobblin/pull/4013?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2123] create reevaluate dag action for jobs in state PENDING_RETRY [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #4013:
URL: https://github.com/apache/gobblin/pull/4013#discussion_r1702456107
##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/proc/ReevaluateDagProcTest.java:
##########
@@ -263,6 +262,47 @@ public void testMultipleNextJobToRun() throws Exception {
specProducers.forEach(sp -> Mockito.verify(sp, Mockito.never()).addSpec(any()));
}
+ @Test
+ public void testRetryCurrentFailedJob() throws Exception {
+ String flowName = "fn5";
+ Dag<JobExecutionPlan> dag = DagManagerTest.buildDag("1", flowExecutionId, DagManager.FailureOption.FINISH_ALL_POSSIBLE.name(),
+ 2, "user5", ConfigFactory.empty()
+ .withValue(ConfigurationKeys.FLOW_GROUP_KEY, ConfigValueFactory.fromAnyRef(flowGroup))
+ .withValue(ConfigurationKeys.FLOW_NAME_KEY, ConfigValueFactory.fromAnyRef(flowName))
+ .withValue(ConfigurationKeys.JOB_GROUP_KEY, ConfigValueFactory.fromAnyRef(flowGroup))
+ .withValue(ConfigurationKeys.SPECEXECUTOR_INSTANCE_URI_KEY, ConfigValueFactory.fromAnyRef(
+ MySqlDagManagementStateStoreTest.TEST_SPEC_EXECUTOR_URI))
+ );
+ List<SpecProducer<Spec>> specProducers = getDagSpecProducers(dag);
+ dagManagementStateStore.addDag(dag);
+ // a job status with shouldRetry=true
+ JobStatus jobStatus = JobStatus.builder().flowName(flowName).flowGroup(flowGroup).jobGroup(flowGroup)
+ .jobName("job0").flowExecutionId(flowExecutionId).message("Test message").eventName(ExecutionStatus.FAILED.name())
+ .startTime(flowExecutionId).shouldRetry(true).orchestratedTime(flowExecutionId).build();
+ doReturn(new ImmutablePair<>(Optional.of(dag.getNodes().get(0)), Optional.of(jobStatus)))
+ .when(dagManagementStateStore).getDagNodeWithJobStatus(any());
+
+ ReevaluateDagProc
+ reEvaluateDagProc = new ReevaluateDagProc(new ReevaluateDagTask(new DagActionStore.DagAction(flowGroup, flowName,
+ flowExecutionId, "job0", DagActionStore.DagActionType.REEVALUATE), null,
+ dagManagementStateStore, mockedDagProcEngineMetrics));
+ reEvaluateDagProc.process(dagManagementStateStore, mockedDagProcEngineMetrics);
+
+ int numOfLaunchedJobs = 1; // only the current job
+ // only the current job should have run
+ Mockito.verify(specProducers.get(0), Mockito.times(1)).addSpec(any());
+
+ specProducers.stream().skip(numOfLaunchedJobs) // separately verified `specProducers.get(0)`
Review Comment:
yes, ok, updated the comment
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] keep the job status as FAILED even if job should be retried, when dag… [gobblin]
Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on code in PR #4013:
URL: https://github.com/apache/gobblin/pull/4013#discussion_r1702192349
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/spec/JobExecutionPlanListSerializer.java:
##########
@@ -82,6 +82,8 @@ public JsonElement serialize(List<JobExecutionPlan> src, Type typeOfSrc, JsonSer
String executionStatus = jobExecutionPlan.getExecutionStatus().name();
jobExecutionPlanJson.addProperty(SerializationConstants.EXECUTION_STATUS_KEY, executionStatus);
+ jobExecutionPlanJson.addProperty(SerializationConstants.CURRENT_GENERATION_KEY, jobExecutionPlan.getCurrentGeneration());
Review Comment:
are these two fields always present in jobExecutionPlan or do they need a default value?
##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/proc/ReevaluateDagProcTest.java:
##########
@@ -263,6 +262,47 @@ public void testMultipleNextJobToRun() throws Exception {
specProducers.forEach(sp -> Mockito.verify(sp, Mockito.never()).addSpec(any()));
}
+ @Test
+ public void testRetryCurrentFailedJob() throws Exception {
+ String flowName = "fn5";
+ Dag<JobExecutionPlan> dag = DagManagerTest.buildDag("1", flowExecutionId, DagManager.FailureOption.FINISH_ALL_POSSIBLE.name(),
+ 2, "user5", ConfigFactory.empty()
+ .withValue(ConfigurationKeys.FLOW_GROUP_KEY, ConfigValueFactory.fromAnyRef(flowGroup))
+ .withValue(ConfigurationKeys.FLOW_NAME_KEY, ConfigValueFactory.fromAnyRef(flowName))
+ .withValue(ConfigurationKeys.JOB_GROUP_KEY, ConfigValueFactory.fromAnyRef(flowGroup))
+ .withValue(ConfigurationKeys.SPECEXECUTOR_INSTANCE_URI_KEY, ConfigValueFactory.fromAnyRef(
+ MySqlDagManagementStateStoreTest.TEST_SPEC_EXECUTOR_URI))
+ );
+ List<SpecProducer<Spec>> specProducers = getDagSpecProducers(dag);
+ dagManagementStateStore.addDag(dag);
+ // a job status with shouldRetry=true
+ JobStatus jobStatus = JobStatus.builder().flowName(flowName).flowGroup(flowGroup).jobGroup(flowGroup)
+ .jobName("job0").flowExecutionId(flowExecutionId).message("Test message").eventName(ExecutionStatus.FAILED.name())
+ .startTime(flowExecutionId).shouldRetry(true).orchestratedTime(flowExecutionId).build();
+ doReturn(new ImmutablePair<>(Optional.of(dag.getNodes().get(0)), Optional.of(jobStatus)))
+ .when(dagManagementStateStore).getDagNodeWithJobStatus(any());
+
+ ReevaluateDagProc
+ reEvaluateDagProc = new ReevaluateDagProc(new ReevaluateDagTask(new DagActionStore.DagAction(flowGroup, flowName,
+ flowExecutionId, "job0", DagActionStore.DagActionType.REEVALUATE), null,
+ dagManagementStateStore, mockedDagProcEngineMetrics));
+ reEvaluateDagProc.process(dagManagementStateStore, mockedDagProcEngineMetrics);
+
+ int numOfLaunchedJobs = 1; // only the current job
+ // only the current job should have run
+ Mockito.verify(specProducers.get(0), Mockito.times(1)).addSpec(any());
+
+ specProducers.stream().skip(numOfLaunchedJobs) // separately verified `specProducers.get(0)`
Review Comment:
is this due to reevaluate dagProc relaunching the job? can u make that more clear with a comment this test is hard to follow bc of the detailed logic being tested here
##########
gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/FlowStatusGenerator.java:
##########
@@ -41,7 +41,7 @@
@Slf4j
public class FlowStatusGenerator {
public static final List<String> FINISHED_STATUSES = Lists.newArrayList(ExecutionStatus.FAILED.name(),
- ExecutionStatus.COMPLETE.name(), ExecutionStatus.CANCELLED.name());
+ ExecutionStatus.COMPLETE.name(), ExecutionStatus.CANCELLED.name(), ExecutionStatus.PENDING_RETRY.name());
Review Comment:
Why use this PENDING_RETRY status at all if Reevaluate is going to use the # of retry attempts to determine if anything to retry for the job? can you summarize how this particular status is used in the PR desc?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2123] create reevaluate dag action for jobs in state PENDING_RETRY [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 merged PR #4013:
URL: https://github.com/apache/gobblin/pull/4013
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org