phet commented on code in PR #3853:
URL: https://github.com/apache/gobblin/pull/3853#discussion_r1447971860
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java:
##########
@@ -655,15 +655,15 @@ private void beginResumingDag(DagId dagIdToResume) throws
IOException {
*/
private void finishResumingDags() throws IOException {
for (Map.Entry<String, Dag<JobExecutionPlan>> dag :
this.resumingDags.entrySet()) {
- JobStatus flowStatus = DagManagerUtils.pollFlowStatus(dag.getValue(),
this.jobStatusRetriever, this.jobStatusPolledTimer);
- if (flowStatus == null ||
!flowStatus.getEventName().equals(PENDING_RESUME.name())) {
+ Optional<JobStatus> flowStatus =
DagManagerUtils.pollFlowStatus(dag.getValue(), this.jobStatusRetriever,
this.jobStatusPolledTimer);
+ if (!flowStatus.isPresent() ||
!flowStatus.get().getEventName().equals(PENDING_RESUME.name())) {
Review Comment:
how about
```
if (!flowStatus.filter(fs ->
fs.getEventName().equals(PENDING_RESUME.name())))
```
?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagerUtils.java:
##########
@@ -433,17 +431,17 @@ public static JobStatus
pollFlowStatus(Dag<JobExecutionPlan> dag, JobStatusRetr
/**
Review Comment:
nit: document that `Optional.absent()` indicates job not found (e.g. rather
than throwing)
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java:
##########
@@ -655,15 +655,15 @@ private void beginResumingDag(DagId dagIdToResume) throws
IOException {
*/
private void finishResumingDags() throws IOException {
for (Map.Entry<String, Dag<JobExecutionPlan>> dag :
this.resumingDags.entrySet()) {
- JobStatus flowStatus = DagManagerUtils.pollFlowStatus(dag.getValue(),
this.jobStatusRetriever, this.jobStatusPolledTimer);
- if (flowStatus == null ||
!flowStatus.getEventName().equals(PENDING_RESUME.name())) {
+ Optional<JobStatus> flowStatus =
DagManagerUtils.pollFlowStatus(dag.getValue(), this.jobStatusRetriever,
this.jobStatusPolledTimer);
+ if (!flowStatus.isPresent() ||
!flowStatus.get().getEventName().equals(PENDING_RESUME.name())) {
continue;
}
boolean dagReady = true;
for (DagNode<JobExecutionPlan> node : dag.getValue().getNodes()) {
- JobStatus jobStatus = DagManagerUtils.pollJobStatus(node,
this.jobStatusRetriever, this.jobStatusPolledTimer);
- if (jobStatus == null ||
jobStatus.getEventName().equals(FAILED.name()) ||
jobStatus.getEventName().equals(CANCELLED.name())) {
+ Optional<JobStatus> jobStatus = DagManagerUtils.pollJobStatus(node,
this.jobStatusRetriever, this.jobStatusPolledTimer);
+ if (!jobStatus.isPresent() ||
jobStatus.get().getEventName().equals(FAILED.name()) ||
jobStatus.get().getEventName().equals(CANCELLED.name())) {
Review Comment:
(similar here... and below)
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
