You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@gobblin.apache.org by "umustafi (via GitHub)" <gi...@apache.org> on 2023/07/28 21:22:50 UTC
[GitHub] [gobblin] umustafi opened a new pull request, #3727: [GOBBLIN-1863] DagManager checks leader status before adding dag to avoid NPE
umustafi opened a new pull request, #3727:
URL: https://github.com/apache/gobblin/pull/3727
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
- [X] 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-1863
### Description
- [X] Here are some details about my PR, including screenshots (if applicable):
Calling this method from non-leader hosts in Multi-Active state throws a NPE which causes a FLOW FAILURE event to be emitted
### Tests
- [X] My PR adds the following unit tests __OR__ does not need testing for this extremely good reason:
NA
### Commits
- [X] 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
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3727: [GOBBLIN-1863] Multi-Active Launch Job Related Issues
Posted by "Will-Lo (via GitHub)" <gi...@apache.org>.
Will-Lo commented on code in PR #3727:
URL: https://github.com/apache/gobblin/pull/3727#discussion_r1281064022
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -374,9 +336,114 @@ public void orchestrate(Spec spec, Properties jobProps, long triggerTimestampMil
Instrumented.updateTimer(this.flowOrchestrationTimer, System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
}
- public void submitFlowToDagManager(FlowSpec flowSpec)
+ /**
+ * If it is a scheduled flow (and hence, does not have flowExecutionId in the FlowSpec) and the flow compilation is
+ * successful, retrieve the flowExecutionId from the JobSpec.
+ */
+ public void addFlowExecutionIdIfAbsent(Map<String,String> flowMetadata, Dag<JobExecutionPlan> jobExecutionPlanDag) {
+ flowMetadata.putIfAbsent(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD,
+ jobExecutionPlanDag.getNodes().get(0).getValue().getJobSpec().getConfigAsProperties().getProperty(
+ ConfigurationKeys.FLOW_EXECUTION_ID_KEY));
+ }
+
+ /**
+ * Checks if flowSpec disallows concurrent executions, and if so then checks if another instance of the flow is
+ * already running and emits a FLOW FAILED event. Otherwise, this check passes.
+ * @return true if caller can proceed to execute flow, false otherwise
+ * @throws IOException
+ */
+ public boolean isExecutionPermittedHandler(Config flowConfig, Spec spec, String flowName, String flowGroup)
throws IOException {
- submitFlowToDagManager(flowSpec, specCompiler.compileFlow(flowSpec));
+ boolean allowConcurrentExecution = ConfigUtils.getBoolean(flowConfig, ConfigurationKeys.FLOW_ALLOW_CONCURRENT_EXECUTION, this.flowConcurrencyFlag);
+
+ Dag<JobExecutionPlan> jobExecutionPlanDag = specCompiler.compileFlow(spec);
+
+ if (!isExecutionPermitted(flowName, flowGroup, allowConcurrentExecution)) {
+ _log.warn("Another instance of flowGroup: {}, flowName: {} running; Skipping flow execution since "
+ + "concurrent executions are disabled for this flow.", flowGroup, flowName);
+ conditionallyUpdateFlowGaugeSpecState(spec, CompiledState.SKIPPED);
+ Instrumented.markMeter(this.skippedFlowsMeter);
+ if (!((FlowSpec) spec).getConfigAsProperties().containsKey(ConfigurationKeys.JOB_SCHEDULE_KEY)) {
+ // For ad-hoc flow, we might already increase quota, we need to decrease here
+ for (Dag.DagNode dagNode : jobExecutionPlanDag.getStartNodes()) {
+ quotaManager.releaseQuota(dagNode);
+ }
+ }
+
+ // Send FLOW_FAILED event
+ Map<String, String> flowMetadata = TimingEventUtils.getFlowMetadata((FlowSpec) spec);
+ flowMetadata.put(TimingEvent.METADATA_MESSAGE, "Flow failed because another instance is running and concurrent "
+ + "executions are disabled. Set flow.allowConcurrentExecution to true in the flow spec to change this behaviour.");
+ if (this.eventSubmitter.isPresent()) {
+ new TimingEvent(this.eventSubmitter.get(), TimingEvent.FlowTimings.FLOW_FAILED).stop(flowMetadata);
+ }
+ return false;
+ }
+ return true;
+ }
+
+
+ /**
+ * Abstraction used to populate the message of and emit a FlowCompileFailed event for the Orchestrator.
+ * @param spec
+ * @param flowMetadata
+ */
+ public void populateFlowCompilationFailedEventMessage(Spec spec,
+ Map<String, String> flowMetadata) {
+ // For scheduled flows, we do not insert the flowExecutionId into the FlowSpec. As a result, if the flow
+ // compilation fails (i.e. we are unable to find a path), the metadata will not have flowExecutionId.
+ // In this case, the current time is used as the flow executionId.
+ flowMetadata.putIfAbsent(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD,
+ Long.toString(System.currentTimeMillis()));
+
+ String message = "Flow was not compiled successfully.";
+ if (!((FlowSpec) spec).getCompilationErrors().isEmpty()) {
+ message = message + " Compilation errors encountered: " + ((FlowSpec) spec).getCompilationErrors();
+ }
+ flowMetadata.put(TimingEvent.METADATA_MESSAGE, message);
+
+ Optional<TimingEvent> flowCompileFailedTimer = eventSubmitter.transform(submitter ->
+ new TimingEvent(submitter, TimingEvent.FlowTimings.FLOW_COMPILE_FAILED));
+
+ if (flowCompileFailedTimer.isPresent()) {
+ flowCompileFailedTimer.get().stop(flowMetadata);
+ }
+ }
+
+ public Optional<Dag<JobExecutionPlan>> handleChecksBeforeExecution(FlowSpec flowSpec)
Review Comment:
Add javadoc explaining what this does
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java:
##########
@@ -455,6 +471,33 @@ public synchronized void setActive(boolean active) {
}
}
+ /**
+ * Used by the DagManager to launch a new execution for a flow action event loaded from the DagActionStore upon
+ * setting this instance of the DagManager to active. Because it may be a completely new DAG not contained in the
+ * dagStore, we compile the flow to generate the dag before calling addDag(), handling any errors that may result in
+ * the process.
+ */
+ public void handleLaunchFlowEvent(DagActionStore.DagAction action) {
+ FlowId flowId = action.getFlowId();
+ FlowSpec spec;
+ try {
+ URI flowUri = FlowSpec.Utils.createFlowSpecUri(flowId);
+ spec = (FlowSpec) flowCatalog.getSpecs(flowUri);
+ Optional<Dag<JobExecutionPlan>> optionalJobExecutionPlanDag = orchestrator.handleChecksBeforeExecution(spec);
Review Comment:
I'm thinking since we need to pass the orchestrator to the dagmanager here which seems like an antipattern, it would be easier to have the orchestrator handle the dagmanager launch event on startup. What would the LOE be in that case?
--
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
[GitHub] [gobblin] umustafi commented on a diff in pull request #3727: [GOBBLIN-1863] Multi-Active Launch Job Related Issues
Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on code in PR #3727:
URL: https://github.com/apache/gobblin/pull/3727#discussion_r1279979986
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -374,9 +358,41 @@ public void orchestrate(Spec spec, Properties jobProps, long triggerTimestampMil
Instrumented.updateTimer(this.flowOrchestrationTimer, System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
}
+ /**
+ * Abstraction used to populate the message of and emit a FlowCompileFailed event for the Orchestrator.
+ * @param spec
+ * @param flowMetadata
+ */
+ public void emitFlowCompilationFailedEvent(Spec spec, Map<String, String> flowMetadata) {
+ // For scheduled flows, we do not insert the flowExecutionId into the FlowSpec. As a result, if the flow
+ // compilation fails (i.e. we are unable to find a path), the metadata will not have flowExecutionId.
+ // In this case, the current time is used as the flow executionId.
+ flowMetadata.putIfAbsent(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD,
+ Long.toString(System.currentTimeMillis()));
+
+ String message = "Flow was not compiled successfully.";
+ if (!((FlowSpec) spec).getCompilationErrors().isEmpty()) {
+ message = message + " Compilation errors encountered: " + ((FlowSpec) spec).getCompilationErrors();
+ }
+ _log.warn(message);
+ flowMetadata.put(TimingEvent.METADATA_MESSAGE, message);
+
+ Optional<TimingEvent> flowCompileFailedTimer = eventSubmitter.transform(submitter ->
+ new TimingEvent(submitter, TimingEvent.FlowTimings.FLOW_COMPILE_FAILED));
+
+ if (flowCompileFailedTimer.isPresent()) {
+ flowCompileFailedTimer.get().stop(flowMetadata);
+ }
+ }
+
public void submitFlowToDagManager(FlowSpec flowSpec)
throws IOException {
- submitFlowToDagManager(flowSpec, specCompiler.compileFlow(flowSpec));
+ Dag<JobExecutionPlan> jobExecutionPlanDag = specCompiler.compileFlow(flowSpec);
+ if (jobExecutionPlanDag == null || jobExecutionPlanDag.isEmpty()) {
+ emitFlowCompilationFailedEvent(flowSpec, TimingEventUtils.getFlowMetadata(flowSpec));
+ return;
+ }
+ submitFlowToDagManager(flowSpec, jobExecutionPlanDag);
Review Comment:
With lots of nesting it becomes hard to tell at which points we may have exited the code block, I'd prefer early return since I end up adding a few more checks that must pass before we submit to the dagManager.
--
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
[GitHub] [gobblin] phet commented on a diff in pull request #3727: [GOBBLIN-1863] Multi-Active Launch Job Related Issues
Posted by "phet (via GitHub)" <gi...@apache.org>.
phet commented on code in PR #3727:
URL: https://github.com/apache/gobblin/pull/3727#discussion_r1279644630
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -374,9 +358,41 @@ public void orchestrate(Spec spec, Properties jobProps, long triggerTimestampMil
Instrumented.updateTimer(this.flowOrchestrationTimer, System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
}
+ /**
+ * Abstraction used to populate the message of and emit a FlowCompileFailed event for the Orchestrator.
+ * @param spec
+ * @param flowMetadata
+ */
+ public void emitFlowCompilationFailedEvent(Spec spec, Map<String, String> flowMetadata) {
+ // For scheduled flows, we do not insert the flowExecutionId into the FlowSpec. As a result, if the flow
+ // compilation fails (i.e. we are unable to find a path), the metadata will not have flowExecutionId.
+ // In this case, the current time is used as the flow executionId.
+ flowMetadata.putIfAbsent(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD,
+ Long.toString(System.currentTimeMillis()));
+
+ String message = "Flow was not compiled successfully.";
+ if (!((FlowSpec) spec).getCompilationErrors().isEmpty()) {
+ message = message + " Compilation errors encountered: " + ((FlowSpec) spec).getCompilationErrors();
+ }
+ _log.warn(message);
+ flowMetadata.put(TimingEvent.METADATA_MESSAGE, message);
+
+ Optional<TimingEvent> flowCompileFailedTimer = eventSubmitter.transform(submitter ->
+ new TimingEvent(submitter, TimingEvent.FlowTimings.FLOW_COMPILE_FAILED));
+
+ if (flowCompileFailedTimer.isPresent()) {
+ flowCompileFailedTimer.get().stop(flowMetadata);
+ }
Review Comment:
if this works (w/o being challenged by checked exceptions), it would be more idiomatic:
```
eventSubmitter.transform(...).ifPresent(timer -> timer.stop(fmd));
```
--
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
[GitHub] [gobblin] phet commented on a diff in pull request #3727: [GOBBLIN-1863] Multi-Active Launch Job Related Issues
Posted by "phet (via GitHub)" <gi...@apache.org>.
phet commented on code in PR #3727:
URL: https://github.com/apache/gobblin/pull/3727#discussion_r1279643104
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -374,9 +358,41 @@ public void orchestrate(Spec spec, Properties jobProps, long triggerTimestampMil
Instrumented.updateTimer(this.flowOrchestrationTimer, System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
}
+ /**
+ * Abstraction used to populate the message of and emit a FlowCompileFailed event for the Orchestrator.
+ * @param spec
+ * @param flowMetadata
+ */
+ public void emitFlowCompilationFailedEvent(Spec spec, Map<String, String> flowMetadata) {
+ // For scheduled flows, we do not insert the flowExecutionId into the FlowSpec. As a result, if the flow
+ // compilation fails (i.e. we are unable to find a path), the metadata will not have flowExecutionId.
+ // In this case, the current time is used as the flow executionId.
+ flowMetadata.putIfAbsent(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD,
+ Long.toString(System.currentTimeMillis()));
+
+ String message = "Flow was not compiled successfully.";
+ if (!((FlowSpec) spec).getCompilationErrors().isEmpty()) {
+ message = message + " Compilation errors encountered: " + ((FlowSpec) spec).getCompilationErrors();
+ }
+ _log.warn(message);
Review Comment:
this wasn't logged before. is it not enough to return the outcome to the user? should we also log it internally? (there may be a lot...)
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java:
##########
@@ -455,6 +470,54 @@ public synchronized void setActive(boolean active) {
}
}
+ /**
+ * Used by the DagManager to launch a new execution for a flow action event loaded from the DagActionStore upon
+ * setting this instance of the DagManager to active. Because it may be a completely new DAG not contained in the
+ * dagStore, we compile the flow to generate the dag before calling addDag(), handling any errors that may result in
+ * the process.
+ */
+ public void handleLaunchFlowEvent(DagActionStore.DagAction action) {
+ if (this.specCompiler.isPresent()) {
+ FlowId flowId = new FlowId().setFlowGroup(action.getFlowGroup()).setFlowName(action.getFlowName());
Review Comment:
perhaps `DagAction` deserves a `getFlowId()` method?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java:
##########
@@ -455,6 +470,54 @@ public synchronized void setActive(boolean active) {
}
}
+ /**
+ * Used by the DagManager to launch a new execution for a flow action event loaded from the DagActionStore upon
+ * setting this instance of the DagManager to active. Because it may be a completely new DAG not contained in the
+ * dagStore, we compile the flow to generate the dag before calling addDag(), handling any errors that may result in
+ * the process.
+ */
+ public void handleLaunchFlowEvent(DagActionStore.DagAction action) {
+ if (this.specCompiler.isPresent()) {
Review Comment:
```
this.specCompiler.ifPresent(theSpecCompiler -> {
...
});
```
?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -374,9 +358,41 @@ public void orchestrate(Spec spec, Properties jobProps, long triggerTimestampMil
Instrumented.updateTimer(this.flowOrchestrationTimer, System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
}
+ /**
+ * Abstraction used to populate the message of and emit a FlowCompileFailed event for the Orchestrator.
+ * @param spec
+ * @param flowMetadata
+ */
+ public void emitFlowCompilationFailedEvent(Spec spec, Map<String, String> flowMetadata) {
+ // For scheduled flows, we do not insert the flowExecutionId into the FlowSpec. As a result, if the flow
+ // compilation fails (i.e. we are unable to find a path), the metadata will not have flowExecutionId.
+ // In this case, the current time is used as the flow executionId.
+ flowMetadata.putIfAbsent(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD,
+ Long.toString(System.currentTimeMillis()));
+
+ String message = "Flow was not compiled successfully.";
+ if (!((FlowSpec) spec).getCompilationErrors().isEmpty()) {
+ message = message + " Compilation errors encountered: " + ((FlowSpec) spec).getCompilationErrors();
+ }
+ _log.warn(message);
+ flowMetadata.put(TimingEvent.METADATA_MESSAGE, message);
+
+ Optional<TimingEvent> flowCompileFailedTimer = eventSubmitter.transform(submitter ->
+ new TimingEvent(submitter, TimingEvent.FlowTimings.FLOW_COMPILE_FAILED));
+
+ if (flowCompileFailedTimer.isPresent()) {
+ flowCompileFailedTimer.get().stop(flowMetadata);
+ }
Review Comment:
if this works (w/o being challenged by non-RuntimeExceptions), it would be more idiomatic:
```
eventSubmitter.transform(...).ifPresent(timer -> timer.stop(fmd));
```
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java:
##########
@@ -455,6 +470,54 @@ public synchronized void setActive(boolean active) {
}
}
+ /**
+ * Used by the DagManager to launch a new execution for a flow action event loaded from the DagActionStore upon
+ * setting this instance of the DagManager to active. Because it may be a completely new DAG not contained in the
+ * dagStore, we compile the flow to generate the dag before calling addDag(), handling any errors that may result in
+ * the process.
+ */
+ public void handleLaunchFlowEvent(DagActionStore.DagAction action) {
+ if (this.specCompiler.isPresent()) {
+ FlowId flowId = new FlowId().setFlowGroup(action.getFlowGroup()).setFlowName(action.getFlowName());
+ FlowSpec spec;
+ try {
+ URI flowUri = FlowSpec.Utils.createFlowSpecUri(flowId);
+ spec = (FlowSpec) flowCatalog.getSpecs(flowUri);
+ Dag<JobExecutionPlan> jobExecutionPlanDag = specCompiler.get().compileFlow(spec);
+ if (jobExecutionPlanDag == null || jobExecutionPlanDag.isEmpty()) {
+ Map<String, String> flowMetadata = TimingEventUtils.getFlowMetadata(spec);
+ // For scheduled flows, we do not insert the flowExecutionId into the FlowSpec. As a result, if the flow
+ // compilation fails (i.e. we are unable to find a path), the metadata will not have flowExecutionId.
+ // In this case, the current time is used as the flow executionId.
+ flowMetadata.putIfAbsent(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD,
+ Long.toString(System.currentTimeMillis()));
+
+ String message = "Flow was not compiled successfully.";
+ if (!( spec).getCompilationErrors().isEmpty()) {
+ message = message + " Compilation errors encountered: " + ( spec).getCompilationErrors();
+ }
+ flowMetadata.put(TimingEvent.METADATA_MESSAGE, message);
+
+ Optional<TimingEvent> flowCompileFailedTimer = eventSubmitter.transform(submitter ->
+ new TimingEvent(submitter, TimingEvent.FlowTimings.FLOW_COMPILE_FAILED));
+
+ if (flowCompileFailedTimer.isPresent()) {
+ flowCompileFailedTimer.get().stop(flowMetadata);
+ }
+ }
+ addDag(jobExecutionPlanDag, true, true);
+ } catch (URISyntaxException e) {
+ log.warn("Could not create URI object for flowId {} due to error {}", flowId, e.getMessage());
+ } catch (SpecNotFoundException e) {
+ log.warn("Spec not found for flow group: {} name: {} Exception: {}", action.getFlowGroup(),
+ action.getFlowName(), e);
+ } catch (IOException e) {
+ log.warn("Failed to add Job Execution Plan for flow group: {} name: {} due to error {}",
+ action.getFlowGroup(), action.getFlowName(), e);
Review Comment:
for consistency, let's have ALL log `flowId`
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java:
##########
@@ -455,6 +470,54 @@ public synchronized void setActive(boolean active) {
}
}
+ /**
+ * Used by the DagManager to launch a new execution for a flow action event loaded from the DagActionStore upon
+ * setting this instance of the DagManager to active. Because it may be a completely new DAG not contained in the
+ * dagStore, we compile the flow to generate the dag before calling addDag(), handling any errors that may result in
+ * the process.
+ */
+ public void handleLaunchFlowEvent(DagActionStore.DagAction action) {
+ if (this.specCompiler.isPresent()) {
+ FlowId flowId = new FlowId().setFlowGroup(action.getFlowGroup()).setFlowName(action.getFlowName());
+ FlowSpec spec;
+ try {
+ URI flowUri = FlowSpec.Utils.createFlowSpecUri(flowId);
+ spec = (FlowSpec) flowCatalog.getSpecs(flowUri);
+ Dag<JobExecutionPlan> jobExecutionPlanDag = specCompiler.get().compileFlow(spec);
+ if (jobExecutionPlanDag == null || jobExecutionPlanDag.isEmpty()) {
+ Map<String, String> flowMetadata = TimingEventUtils.getFlowMetadata(spec);
+ // For scheduled flows, we do not insert the flowExecutionId into the FlowSpec. As a result, if the flow
+ // compilation fails (i.e. we are unable to find a path), the metadata will not have flowExecutionId.
+ // In this case, the current time is used as the flow executionId.
+ flowMetadata.putIfAbsent(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD,
+ Long.toString(System.currentTimeMillis()));
+
+ String message = "Flow was not compiled successfully.";
+ if (!( spec).getCompilationErrors().isEmpty()) {
+ message = message + " Compilation errors encountered: " + ( spec).getCompilationErrors();
+ }
+ flowMetadata.put(TimingEvent.METADATA_MESSAGE, message);
+
+ Optional<TimingEvent> flowCompileFailedTimer = eventSubmitter.transform(submitter ->
+ new TimingEvent(submitter, TimingEvent.FlowTimings.FLOW_COMPILE_FAILED));
+
+ if (flowCompileFailedTimer.isPresent()) {
+ flowCompileFailedTimer.get().stop(flowMetadata);
+ }
Review Comment:
rather than copy-paste what you just defined as `emitFlowCompilationFailureEvent`, can't we call to that? (e.g. make the impl `static` and then maybe have an `Orchestrator` method that delegates to it).
that said, when we observe the need to add two new members to `DagMgr` plus to "borrow" a chunk of code from `Orchestrator`, it may be worth reviewing whether the legacy abstractions and separations of responsibility are still right for the present and future...
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -374,9 +358,41 @@ public void orchestrate(Spec spec, Properties jobProps, long triggerTimestampMil
Instrumented.updateTimer(this.flowOrchestrationTimer, System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
}
+ /**
+ * Abstraction used to populate the message of and emit a FlowCompileFailed event for the Orchestrator.
+ * @param spec
+ * @param flowMetadata
+ */
+ public void emitFlowCompilationFailedEvent(Spec spec, Map<String, String> flowMetadata) {
+ // For scheduled flows, we do not insert the flowExecutionId into the FlowSpec. As a result, if the flow
+ // compilation fails (i.e. we are unable to find a path), the metadata will not have flowExecutionId.
+ // In this case, the current time is used as the flow executionId.
+ flowMetadata.putIfAbsent(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD,
+ Long.toString(System.currentTimeMillis()));
+
+ String message = "Flow was not compiled successfully.";
+ if (!((FlowSpec) spec).getCompilationErrors().isEmpty()) {
+ message = message + " Compilation errors encountered: " + ((FlowSpec) spec).getCompilationErrors();
+ }
+ _log.warn(message);
+ flowMetadata.put(TimingEvent.METADATA_MESSAGE, message);
+
+ Optional<TimingEvent> flowCompileFailedTimer = eventSubmitter.transform(submitter ->
+ new TimingEvent(submitter, TimingEvent.FlowTimings.FLOW_COMPILE_FAILED));
+
+ if (flowCompileFailedTimer.isPresent()) {
+ flowCompileFailedTimer.get().stop(flowMetadata);
+ }
+ }
+
public void submitFlowToDagManager(FlowSpec flowSpec)
throws IOException {
- submitFlowToDagManager(flowSpec, specCompiler.compileFlow(flowSpec));
+ Dag<JobExecutionPlan> jobExecutionPlanDag = specCompiler.compileFlow(flowSpec);
+ if (jobExecutionPlanDag == null || jobExecutionPlanDag.isEmpty()) {
+ emitFlowCompilationFailedEvent(flowSpec, TimingEventUtils.getFlowMetadata(flowSpec));
+ return;
+ }
+ submitFlowToDagManager(flowSpec, jobExecutionPlanDag);
Review Comment:
I find `else` MUCH preferable to the early return here
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -374,9 +358,41 @@ public void orchestrate(Spec spec, Properties jobProps, long triggerTimestampMil
Instrumented.updateTimer(this.flowOrchestrationTimer, System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
}
+ /**
+ * Abstraction used to populate the message of and emit a FlowCompileFailed event for the Orchestrator.
+ * @param spec
+ * @param flowMetadata
+ */
+ public void emitFlowCompilationFailedEvent(Spec spec, Map<String, String> flowMetadata) {
Review Comment:
no biggie, but naming-wise, I agree the timing is emitted... but what may be most noteworthy is that it adds the `METADATA_MESSAGE`
--
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
[GitHub] [gobblin] umustafi commented on a diff in pull request #3727: [GOBBLIN-1863] Multi-Active Launch Job Related Issues
Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on code in PR #3727:
URL: https://github.com/apache/gobblin/pull/3727#discussion_r1279990300
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -374,9 +358,41 @@ public void orchestrate(Spec spec, Properties jobProps, long triggerTimestampMil
Instrumented.updateTimer(this.flowOrchestrationTimer, System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
}
+ /**
+ * Abstraction used to populate the message of and emit a FlowCompileFailed event for the Orchestrator.
+ * @param spec
+ * @param flowMetadata
+ */
+ public void emitFlowCompilationFailedEvent(Spec spec, Map<String, String> flowMetadata) {
+ // For scheduled flows, we do not insert the flowExecutionId into the FlowSpec. As a result, if the flow
+ // compilation fails (i.e. we are unable to find a path), the metadata will not have flowExecutionId.
+ // In this case, the current time is used as the flow executionId.
+ flowMetadata.putIfAbsent(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD,
+ Long.toString(System.currentTimeMillis()));
+
+ String message = "Flow was not compiled successfully.";
+ if (!((FlowSpec) spec).getCompilationErrors().isEmpty()) {
+ message = message + " Compilation errors encountered: " + ((FlowSpec) spec).getCompilationErrors();
+ }
+ _log.warn(message);
+ flowMetadata.put(TimingEvent.METADATA_MESSAGE, message);
+
+ Optional<TimingEvent> flowCompileFailedTimer = eventSubmitter.transform(submitter ->
+ new TimingEvent(submitter, TimingEvent.FlowTimings.FLOW_COMPILE_FAILED));
+
+ if (flowCompileFailedTimer.isPresent()) {
+ flowCompileFailedTimer.get().stop(flowMetadata);
+ }
+ }
+
public void submitFlowToDagManager(FlowSpec flowSpec)
throws IOException {
- submitFlowToDagManager(flowSpec, specCompiler.compileFlow(flowSpec));
+ Dag<JobExecutionPlan> jobExecutionPlanDag = specCompiler.compileFlow(flowSpec);
+ if (jobExecutionPlanDag == null || jobExecutionPlanDag.isEmpty()) {
+ emitFlowCompilationFailedEvent(flowSpec, TimingEventUtils.getFlowMetadata(flowSpec));
+ return;
+ }
+ submitFlowToDagManager(flowSpec, jobExecutionPlanDag);
Review Comment:
We don't call orchestrate function directly from the `DagActionStoreMonitor` because the chain of events is `Scheduler -> Orchestrator -> FlowTriggerHandler (Multi-Active algorithm) -> MySQL -> Brooklin change stream -> DagActionStoreMonitor -> DagManager` so we don't want the MA algorithm triggered again. However, I did end up abstracting all the checks done to compile the flow and ensure an execution is allowed to re-use when submitting the flow to the `DagManager` from the `DagActionStoreMonitor`
--
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
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3727: [GOBBLIN-1863] Multi-Active Launch Job Related Issues
Posted by "Will-Lo (via GitHub)" <gi...@apache.org>.
Will-Lo commented on code in PR #3727:
URL: https://github.com/apache/gobblin/pull/3727#discussion_r1279761914
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java:
##########
@@ -455,6 +470,54 @@ public synchronized void setActive(boolean active) {
}
}
+ /**
+ * Used by the DagManager to launch a new execution for a flow action event loaded from the DagActionStore upon
+ * setting this instance of the DagManager to active. Because it may be a completely new DAG not contained in the
+ * dagStore, we compile the flow to generate the dag before calling addDag(), handling any errors that may result in
+ * the process.
+ */
+ public void handleLaunchFlowEvent(DagActionStore.DagAction action) {
+ if (this.specCompiler.isPresent()) {
+ FlowId flowId = new FlowId().setFlowGroup(action.getFlowGroup()).setFlowName(action.getFlowName());
Review Comment:
Same comment here as the one below, wanted to double check that all the checks are being done before submitting these flows which the orchestrator is doing, otherwise there will be some bugs introduced that are guarded against. Is it possible to use a helper or a static function?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java:
##########
@@ -280,20 +290,22 @@ protected void startUp() {
* Note this should only be called from the {@link Orchestrator} or {@link org.apache.gobblin.service.monitoring.DagActionStoreChangeMonitor}
*/
public synchronized void addDag(Dag<JobExecutionPlan> dag, boolean persist, boolean setStatus) throws IOException {
- if (persist) {
- //Persist the dag
- this.dagStateStore.writeCheckpoint(dag);
- }
- int queueId = DagManagerUtils.getDagQueueId(dag, this.numThreads);
- // Add the dag to the specific queue determined by flowExecutionId
- // Flow cancellation request has to be forwarded to the same DagManagerThread where the
- // flow create request was forwarded. This is because Azkaban Exec Id is stored in the DagNode of the
- // specific DagManagerThread queue
- if (!this.runQueue[queueId].offer(dag)) {
- throw new IOException("Could not add dag" + DagManagerUtils.generateDagId(dag) + "to queue");
- }
- if (setStatus) {
- submitEventsAndSetStatus(dag);
+ if (isActive) {
Review Comment:
Stylistic nit but sometimes I like doing the reverse and exit early. Exactly the same thing but sometimes more readable than a lot of nested if statements
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -374,9 +358,41 @@ public void orchestrate(Spec spec, Properties jobProps, long triggerTimestampMil
Instrumented.updateTimer(this.flowOrchestrationTimer, System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
}
+ /**
+ * Abstraction used to populate the message of and emit a FlowCompileFailed event for the Orchestrator.
+ * @param spec
+ * @param flowMetadata
+ */
+ public void emitFlowCompilationFailedEvent(Spec spec, Map<String, String> flowMetadata) {
+ // For scheduled flows, we do not insert the flowExecutionId into the FlowSpec. As a result, if the flow
+ // compilation fails (i.e. we are unable to find a path), the metadata will not have flowExecutionId.
+ // In this case, the current time is used as the flow executionId.
+ flowMetadata.putIfAbsent(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD,
+ Long.toString(System.currentTimeMillis()));
+
+ String message = "Flow was not compiled successfully.";
+ if (!((FlowSpec) spec).getCompilationErrors().isEmpty()) {
+ message = message + " Compilation errors encountered: " + ((FlowSpec) spec).getCompilationErrors();
+ }
+ _log.warn(message);
+ flowMetadata.put(TimingEvent.METADATA_MESSAGE, message);
+
+ Optional<TimingEvent> flowCompileFailedTimer = eventSubmitter.transform(submitter ->
+ new TimingEvent(submitter, TimingEvent.FlowTimings.FLOW_COMPILE_FAILED));
+
+ if (flowCompileFailedTimer.isPresent()) {
+ flowCompileFailedTimer.get().stop(flowMetadata);
+ }
+ }
+
public void submitFlowToDagManager(FlowSpec flowSpec)
throws IOException {
- submitFlowToDagManager(flowSpec, specCompiler.compileFlow(flowSpec));
+ Dag<JobExecutionPlan> jobExecutionPlanDag = specCompiler.compileFlow(flowSpec);
+ if (jobExecutionPlanDag == null || jobExecutionPlanDag.isEmpty()) {
+ emitFlowCompilationFailedEvent(flowSpec, TimingEventUtils.getFlowMetadata(flowSpec));
+ return;
+ }
+ submitFlowToDagManager(flowSpec, jobExecutionPlanDag);
Review Comment:
So I'm wondering why we don't directly call the `orchestrate()` function from the SpecMonitor instead of `submitFlowToDagManager`, because there are a few more checks done in that function, namely:
1. Checking for another execution
2. Ensuring that the specCompiler is ready before compilation (happens less these days with a file based compiler but still possible that there will be a race condition introduced where you try to compile a flow when the service starts up but the flowgraph is not completed).
If this is not possible, we need to abstract these checks to another function and ensure that they're done before submitting to the dagmanager in any configuration.
--
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
[GitHub] [gobblin] phet commented on a diff in pull request #3727: [GOBBLIN-1863] Multi-Active Launch Job Related Issues
Posted by "phet (via GitHub)" <gi...@apache.org>.
phet commented on code in PR #3727:
URL: https://github.com/apache/gobblin/pull/3727#discussion_r1279648102
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -374,9 +358,41 @@ public void orchestrate(Spec spec, Properties jobProps, long triggerTimestampMil
Instrumented.updateTimer(this.flowOrchestrationTimer, System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
}
+ /**
+ * Abstraction used to populate the message of and emit a FlowCompileFailed event for the Orchestrator.
+ * @param spec
+ * @param flowMetadata
+ */
+ public void emitFlowCompilationFailedEvent(Spec spec, Map<String, String> flowMetadata) {
Review Comment:
no biggie, but naming-wise, I agree the timing is emitted... but what may be most noteworthy is that it inserts the `METADATA_MESSAGE`
--
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
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3727: [GOBBLIN-1863] Multi-Active Launch Job Related Issues
Posted by "Will-Lo (via GitHub)" <gi...@apache.org>.
Will-Lo commented on code in PR #3727:
URL: https://github.com/apache/gobblin/pull/3727#discussion_r1279761914
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java:
##########
@@ -455,6 +470,54 @@ public synchronized void setActive(boolean active) {
}
}
+ /**
+ * Used by the DagManager to launch a new execution for a flow action event loaded from the DagActionStore upon
+ * setting this instance of the DagManager to active. Because it may be a completely new DAG not contained in the
+ * dagStore, we compile the flow to generate the dag before calling addDag(), handling any errors that may result in
+ * the process.
+ */
+ public void handleLaunchFlowEvent(DagActionStore.DagAction action) {
+ if (this.specCompiler.isPresent()) {
+ FlowId flowId = new FlowId().setFlowGroup(action.getFlowGroup()).setFlowName(action.getFlowName());
Review Comment:
Same comment here as the one above, wanted to double check that all the checks are being done before submitting these flows which the orchestrator is doing, otherwise there will be some bugs introduced that are guarded against. Is it possible to use a helper or a static function?
--
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
[GitHub] [gobblin] umustafi commented on a diff in pull request #3727: [GOBBLIN-1863] Multi-Active Launch Job Related Issues
Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on code in PR #3727:
URL: https://github.com/apache/gobblin/pull/3727#discussion_r1279996442
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java:
##########
@@ -455,6 +470,54 @@ public synchronized void setActive(boolean active) {
}
}
+ /**
+ * Used by the DagManager to launch a new execution for a flow action event loaded from the DagActionStore upon
+ * setting this instance of the DagManager to active. Because it may be a completely new DAG not contained in the
+ * dagStore, we compile the flow to generate the dag before calling addDag(), handling any errors that may result in
+ * the process.
+ */
+ public void handleLaunchFlowEvent(DagActionStore.DagAction action) {
+ if (this.specCompiler.isPresent()) {
+ FlowId flowId = new FlowId().setFlowGroup(action.getFlowGroup()).setFlowName(action.getFlowName());
Review Comment:
Re-used helper function created in Orchestrator
--
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
[GitHub] [gobblin] umustafi commented on a diff in pull request #3727: [GOBBLIN-1863] Multi-Active Launch Job Related Issues
Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on code in PR #3727:
URL: https://github.com/apache/gobblin/pull/3727#discussion_r1281090842
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java:
##########
@@ -455,6 +471,33 @@ public synchronized void setActive(boolean active) {
}
}
+ /**
+ * Used by the DagManager to launch a new execution for a flow action event loaded from the DagActionStore upon
+ * setting this instance of the DagManager to active. Because it may be a completely new DAG not contained in the
+ * dagStore, we compile the flow to generate the dag before calling addDag(), handling any errors that may result in
+ * the process.
+ */
+ public void handleLaunchFlowEvent(DagActionStore.DagAction action) {
+ FlowId flowId = action.getFlowId();
+ FlowSpec spec;
+ try {
+ URI flowUri = FlowSpec.Utils.createFlowSpecUri(flowId);
+ spec = (FlowSpec) flowCatalog.getSpecs(flowUri);
+ Optional<Dag<JobExecutionPlan>> optionalJobExecutionPlanDag = orchestrator.handleChecksBeforeExecution(spec);
Review Comment:
Moving this function to the `Orchestrator`, does not actually change the code pattern here since there's only a bit of wrapper code (mainly to get the spec corresponding to the `dagAction`) around the `Orchestrator` helper funcs that it relies on. It's more clear to keep it here since the use case is `DagManager` specific
--
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
[GitHub] [gobblin] umustafi commented on a diff in pull request #3727: [GOBBLIN-1863] Multi-Active Launch Job Related Issues
Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on code in PR #3727:
URL: https://github.com/apache/gobblin/pull/3727#discussion_r1279951090
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -374,9 +358,41 @@ public void orchestrate(Spec spec, Properties jobProps, long triggerTimestampMil
Instrumented.updateTimer(this.flowOrchestrationTimer, System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
}
+ /**
+ * Abstraction used to populate the message of and emit a FlowCompileFailed event for the Orchestrator.
+ * @param spec
+ * @param flowMetadata
+ */
+ public void emitFlowCompilationFailedEvent(Spec spec, Map<String, String> flowMetadata) {
+ // For scheduled flows, we do not insert the flowExecutionId into the FlowSpec. As a result, if the flow
+ // compilation fails (i.e. we are unable to find a path), the metadata will not have flowExecutionId.
+ // In this case, the current time is used as the flow executionId.
+ flowMetadata.putIfAbsent(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD,
+ Long.toString(System.currentTimeMillis()));
+
+ String message = "Flow was not compiled successfully.";
+ if (!((FlowSpec) spec).getCompilationErrors().isEmpty()) {
+ message = message + " Compilation errors encountered: " + ((FlowSpec) spec).getCompilationErrors();
+ }
+ _log.warn(message);
+ flowMetadata.put(TimingEvent.METADATA_MESSAGE, message);
+
+ Optional<TimingEvent> flowCompileFailedTimer = eventSubmitter.transform(submitter ->
+ new TimingEvent(submitter, TimingEvent.FlowTimings.FLOW_COMPILE_FAILED));
+
+ if (flowCompileFailedTimer.isPresent()) {
+ flowCompileFailedTimer.get().stop(flowMetadata);
+ }
Review Comment:
See previous response about Java Optional vs. Google Optional
--
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
[GitHub] [gobblin] Will-Lo merged pull request #3727: [GOBBLIN-1863] Multi-Active Launch Job Related Issues
Posted by "Will-Lo (via GitHub)" <gi...@apache.org>.
Will-Lo merged PR #3727:
URL: https://github.com/apache/gobblin/pull/3727
--
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
[GitHub] [gobblin] umustafi commented on a diff in pull request #3727: [GOBBLIN-1863] Multi-Active Launch Job Related Issues
Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on code in PR #3727:
URL: https://github.com/apache/gobblin/pull/3727#discussion_r1279948789
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java:
##########
@@ -455,6 +470,54 @@ public synchronized void setActive(boolean active) {
}
}
+ /**
+ * Used by the DagManager to launch a new execution for a flow action event loaded from the DagActionStore upon
+ * setting this instance of the DagManager to active. Because it may be a completely new DAG not contained in the
+ * dagStore, we compile the flow to generate the dag before calling addDag(), handling any errors that may result in
+ * the process.
+ */
+ public void handleLaunchFlowEvent(DagActionStore.DagAction action) {
+ if (this.specCompiler.isPresent()) {
Review Comment:
this is using google library's Optional which does not have this method. I've generally seen us using this Optional version so will keep as is.
--
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
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3727: [GOBBLIN-1863] Multi-Active Launch Job Related Issues
Posted by "Will-Lo (via GitHub)" <gi...@apache.org>.
Will-Lo commented on code in PR #3727:
URL: https://github.com/apache/gobblin/pull/3727#discussion_r1279760164
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -374,9 +358,41 @@ public void orchestrate(Spec spec, Properties jobProps, long triggerTimestampMil
Instrumented.updateTimer(this.flowOrchestrationTimer, System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
}
+ /**
+ * Abstraction used to populate the message of and emit a FlowCompileFailed event for the Orchestrator.
+ * @param spec
+ * @param flowMetadata
+ */
+ public void emitFlowCompilationFailedEvent(Spec spec, Map<String, String> flowMetadata) {
+ // For scheduled flows, we do not insert the flowExecutionId into the FlowSpec. As a result, if the flow
+ // compilation fails (i.e. we are unable to find a path), the metadata will not have flowExecutionId.
+ // In this case, the current time is used as the flow executionId.
+ flowMetadata.putIfAbsent(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD,
+ Long.toString(System.currentTimeMillis()));
+
+ String message = "Flow was not compiled successfully.";
+ if (!((FlowSpec) spec).getCompilationErrors().isEmpty()) {
+ message = message + " Compilation errors encountered: " + ((FlowSpec) spec).getCompilationErrors();
+ }
+ _log.warn(message);
+ flowMetadata.put(TimingEvent.METADATA_MESSAGE, message);
+
+ Optional<TimingEvent> flowCompileFailedTimer = eventSubmitter.transform(submitter ->
+ new TimingEvent(submitter, TimingEvent.FlowTimings.FLOW_COMPILE_FAILED));
+
+ if (flowCompileFailedTimer.isPresent()) {
+ flowCompileFailedTimer.get().stop(flowMetadata);
+ }
+ }
+
public void submitFlowToDagManager(FlowSpec flowSpec)
throws IOException {
- submitFlowToDagManager(flowSpec, specCompiler.compileFlow(flowSpec));
+ Dag<JobExecutionPlan> jobExecutionPlanDag = specCompiler.compileFlow(flowSpec);
+ if (jobExecutionPlanDag == null || jobExecutionPlanDag.isEmpty()) {
+ emitFlowCompilationFailedEvent(flowSpec, TimingEventUtils.getFlowMetadata(flowSpec));
+ return;
+ }
+ submitFlowToDagManager(flowSpec, jobExecutionPlanDag);
Review Comment:
So I'm wondering why we don't directly call the `orchestrate()` function from the SpecMonitor instead of `submitFlowToDagManager`, because there are a few more checks done in that function, namely:
1. Checking for another execution
2. Ensuring that the specCompiler is ready before compilation (happens less these days with a file based compiler but still possible that there will be a race condition introduced where you try to compile a flow when the service starts up but the flowgraph is not finished loading into memory).
If this is not possible, we need to abstract these checks to another function and ensure that they're done before submitting to the dagmanager in any configuration.
--
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
[GitHub] [gobblin] umustafi commented on a diff in pull request #3727: [GOBBLIN-1863] Multi-Active Launch Job Related Issues
Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on code in PR #3727:
URL: https://github.com/apache/gobblin/pull/3727#discussion_r1280000488
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -374,9 +358,41 @@ public void orchestrate(Spec spec, Properties jobProps, long triggerTimestampMil
Instrumented.updateTimer(this.flowOrchestrationTimer, System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
}
+ /**
+ * Abstraction used to populate the message of and emit a FlowCompileFailed event for the Orchestrator.
+ * @param spec
+ * @param flowMetadata
+ */
+ public void emitFlowCompilationFailedEvent(Spec spec, Map<String, String> flowMetadata) {
Review Comment:
I renamed to `populateFlowCompilationFailedEventMessage`, what do you think?
--
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
[GitHub] [gobblin] phet commented on a diff in pull request #3727: [GOBBLIN-1863] Multi-Active Launch Job Related Issues
Posted by "phet (via GitHub)" <gi...@apache.org>.
phet commented on code in PR #3727:
URL: https://github.com/apache/gobblin/pull/3727#discussion_r1279876662
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java:
##########
@@ -455,6 +470,54 @@ public synchronized void setActive(boolean active) {
}
}
+ /**
+ * Used by the DagManager to launch a new execution for a flow action event loaded from the DagActionStore upon
+ * setting this instance of the DagManager to active. Because it may be a completely new DAG not contained in the
+ * dagStore, we compile the flow to generate the dag before calling addDag(), handling any errors that may result in
+ * the process.
+ */
+ public void handleLaunchFlowEvent(DagActionStore.DagAction action) {
+ if (this.specCompiler.isPresent()) {
+ FlowId flowId = new FlowId().setFlowGroup(action.getFlowGroup()).setFlowName(action.getFlowName());
+ FlowSpec spec;
+ try {
+ URI flowUri = FlowSpec.Utils.createFlowSpecUri(flowId);
+ spec = (FlowSpec) flowCatalog.getSpecs(flowUri);
+ Dag<JobExecutionPlan> jobExecutionPlanDag = specCompiler.get().compileFlow(spec);
+ if (jobExecutionPlanDag == null || jobExecutionPlanDag.isEmpty()) {
+ Map<String, String> flowMetadata = TimingEventUtils.getFlowMetadata(spec);
+ // For scheduled flows, we do not insert the flowExecutionId into the FlowSpec. As a result, if the flow
+ // compilation fails (i.e. we are unable to find a path), the metadata will not have flowExecutionId.
+ // In this case, the current time is used as the flow executionId.
+ flowMetadata.putIfAbsent(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD,
+ Long.toString(System.currentTimeMillis()));
+
+ String message = "Flow was not compiled successfully.";
+ if (!( spec).getCompilationErrors().isEmpty()) {
+ message = message + " Compilation errors encountered: " + ( spec).getCompilationErrors();
+ }
+ flowMetadata.put(TimingEvent.METADATA_MESSAGE, message);
+
+ Optional<TimingEvent> flowCompileFailedTimer = eventSubmitter.transform(submitter ->
+ new TimingEvent(submitter, TimingEvent.FlowTimings.FLOW_COMPILE_FAILED));
+
+ if (flowCompileFailedTimer.isPresent()) {
+ flowCompileFailedTimer.get().stop(flowMetadata);
+ }
Review Comment:
rather than copy-paste what you just defined as `emitFlowCompilationFailureEvent`, can't we call to that? (e.g. make the impl `static` in `Orchestrator`, who also maybe has an instance method that delegates to it).
that said, when we observe the need to add two new members to `DagMgr` plus to "borrow" a chunk of code from `Orchestrator`, it may be worth reviewing whether the legacy abstractions and separations of responsibility are still right for the present and future...
--
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
[GitHub] [gobblin] codecov-commenter commented on pull request #3727: [GOBBLIN-1863] Multi-Active Launch Job Related Issues
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #3727:
URL: https://github.com/apache/gobblin/pull/3727#issuecomment-1661189179
## [Codecov](https://app.codecov.io/gh/apache/gobblin/pull/3727?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
> Merging [#3727](https://app.codecov.io/gh/apache/gobblin/pull/3727?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (8b9be6e) into [master](https://app.codecov.io/gh/apache/gobblin/commit/8198404ecacc919eefabf22307ab8abf1b97e885?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (8198404) will **decrease** coverage by `2.12%`.
> Report is 1 commits behind head on master.
> The diff coverage is `n/a`.
```diff
@@ Coverage Diff @@
## master #3727 +/- ##
============================================
- Coverage 47.45% 45.33% -2.12%
+ Complexity 8687 2139 -6548
============================================
Files 1731 413 -1318
Lines 66736 17954 -48782
Branches 7219 2183 -5036
============================================
- Hits 31671 8140 -23531
+ Misses 32306 8948 -23358
+ Partials 2759 866 -1893
```
[see 2144 files with indirect coverage changes](https://app.codecov.io/gh/apache/gobblin/pull/3727/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?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