[GOBBLIN-1863] Multi-Active Launch Job Related Issues #3727

umustafi · 2023-07-28T21:22:48Z

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 issues and references them in the PR title. For example, "[GOBBLIN-XXX] My Gobblin PR"
- https://issues.apache.org/jira/browse/GOBBLIN-1863

Description

Here are some details about my PR, including screenshots (if applicable):

Bugs

DagManager check leader status before addDag bc calling this method from non-leader hosts throws a NPE which may caused failed dag event to be emitted. Then those job executions are skipped since they are marked as failed.
also handle LAUNCH type events upon leader change and setting a new participant DagManager to be active. Failing to handle these events may be causing missed flow launches on any leader change or restart.
Refactoring Done
Create a static function in Orchestrator to emit a flow failed event if spec compilation fails. This function is shared by the between the DagManager and Orchestrator. It takes in the eventSubmitter as an argument to emit the event.
I abstracted all the checks in Orchestrator.orchestrate done to compile the flow and ensure an execution is allowed to be re-used by the DagActionStoreChangeMonitor when submitting the flow to the DagManager by calling submitFlowToDagManager.
Note that we skip spec compilation in multi-active mode to avoid two compilations since it's a memory intensive process and we can wait to do it before submitting to the DagManager.

Tests

My PR adds the following unit tests OR does not need testing for this extremely good reason:
NA

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":
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"

gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java

Will-Lo · 2023-07-31T19:12:26Z

...lin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java

  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);


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:

Checking for another execution

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.

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

Will-Lo · 2023-07-31T19:13:54Z

gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java

+  public void handleLaunchFlowEvent(DagActionStore.DagAction action) {
+    if (this.specCompiler.isPresent()) {
+      FlowId flowId = new FlowId().setFlowGroup(action.getFlowGroup()).setFlowName(action.getFlowName());


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?

Re-used helper function created in Orchestrator

...lin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java

phet · 2023-07-31T17:14:47Z

...lin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java

+    Optional<TimingEvent> flowCompileFailedTimer = eventSubmitter.transform(submitter ->
+        new TimingEvent(submitter, TimingEvent.FlowTimings.FLOW_COMPILE_FAILED));
+
+    if (flowCompileFailedTimer.isPresent()) {
+      flowCompileFailedTimer.get().stop(flowMetadata);
+    }


if this works (w/o being challenged by checked exceptions), it would be more idiomatic:

eventSubmitter.transform(...).ifPresent(timer -> timer.stop(fmd));

See previous response about Java Optional vs. Google Optional

phet · 2023-07-31T17:15:18Z

...lin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java

+      return;
+    }
+    submitFlowToDagManager(flowSpec, jobExecutionPlanDag);


I find else MUCH preferable to the early return here

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.

phet · 2023-07-31T17:17:12Z

...lin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java

+   * @param spec
+   * @param flowMetadata
+   */
+  public void emitFlowCompilationFailedEvent(Spec spec, Map<String, String> flowMetadata) {


no biggie, but naming-wise, I agree the timing is emitted... but what may be most noteworthy is that it inserts the METADATA_MESSAGE

I renamed to populateFlowCompilationFailedEventMessage, what do you think?

phet · 2023-07-31T18:19:09Z

gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java

+   * the process.
+   */
+  public void handleLaunchFlowEvent(DagActionStore.DagAction action) {
+    if (this.specCompiler.isPresent()) {


this.specCompiler.ifPresent(theSpecCompiler -> { ... });

?

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.

gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java

...lin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java

Will-Lo · 2023-08-01T19:44:53Z

gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java

+    try {
+      URI flowUri = FlowSpec.Utils.createFlowSpecUri(flowId);
+      spec = (FlowSpec) flowCatalog.getSpecs(flowUri);
+      Optional<Dag<JobExecutionPlan>> optionalJobExecutionPlanDag = orchestrator.handleChecksBeforeExecution(spec);


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?

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

codecov-commenter · 2023-08-01T22:34:06Z

Codecov Report

Merging #3727 (8b9be6e) into master (8198404) will decrease coverage by 0.38%.
Report is 1 commits behind head on master.
The diff coverage is 26.82%.

@@             Coverage Diff              @@
##             master    #3727      +/-   ##
============================================
- Coverage     47.45%   47.08%   -0.38%     
- Complexity     8687    10856    +2169     
============================================
  Files          1731     2144     +413     
  Lines         66736    84746   +18010     
  Branches       7219     9409    +2190     
============================================
+ Hits          31671    39902    +8231     
- Misses        32306    41222    +8916     
- Partials       2759     3622     +863

Files Changed	Coverage Δ
...org/apache/gobblin/runtime/api/DagActionStore.java	`84.61% <0.00%> (-7.06%)`	⬇️
...ervice/monitoring/DagActionStoreChangeMonitor.java	`0.00% <0.00%> (ø)`
...blin/service/modules/orchestration/DagManager.java	`79.87% <12.00%> (-2.65%)`	⬇️
...in/service/modules/orchestration/Orchestrator.java	`47.36% <32.58%> (-4.23%)`	⬇️
...blin/runtime/api/MysqlMultiActiveLeaseArbiter.java	`78.84% <100.00%> (ø)`

... and 426 files with indirect coverage changes

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

* DagManager checks leader status before adding dag to avoid NPE * handle launch events after leader change in DagManager * Refactor Orchestrator and DagManager * remove unecessary specCompiler changes * add docstring --------- Co-authored-by: Urmi Mustafi <[email protected]>

DagManager checks leader status before adding dag to avoid NPE

e9f1ae1

umustafi changed the title ~~[GOBBLIN-1863] DagManager checks leader status before adding dag to avoid NPE~~ [GOBBLIN-1863] Multi-Active Launch Job Related Issues Jul 29, 2023

handle launch events after leader change in DagManager

af24809

Will-Lo requested changes Jul 31, 2023

View reviewed changes

phet reviewed Jul 31, 2023

View reviewed changes

Urmi Mustafi added 2 commits July 31, 2023 17:32

Refactor Orchestrator and DagManager

d5ccbef

remove unecessary specCompiler changes

3ff2330

Will-Lo reviewed Aug 1, 2023

View reviewed changes

add docstring

8b9be6e

Will-Lo approved these changes Aug 1, 2023

View reviewed changes

Will-Lo merged commit d13c327 into apache:master Aug 1, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[GOBBLIN-1863] Multi-Active Launch Job Related Issues #3727

[GOBBLIN-1863] Multi-Active Launch Job Related Issues #3727

umustafi commented Jul 28, 2023 •

edited

Loading

Will-Lo Jul 31, 2023 •

edited

Loading

umustafi Aug 1, 2023

Will-Lo Jul 31, 2023 •

edited

Loading

umustafi Aug 1, 2023

phet Jul 31, 2023 •

edited

Loading

umustafi Jul 31, 2023

phet Jul 31, 2023

umustafi Jul 31, 2023

phet Jul 31, 2023 •

edited

Loading

umustafi Aug 1, 2023

phet Jul 31, 2023

umustafi Jul 31, 2023

Will-Lo Aug 1, 2023

umustafi Aug 1, 2023

codecov-commenter commented Aug 1, 2023 •

edited

Loading

[GOBBLIN-1863] Multi-Active Launch Job Related Issues #3727

[GOBBLIN-1863] Multi-Active Launch Job Related Issues #3727

Conversation

umustafi commented Jul 28, 2023 • edited Loading

JIRA

Description

Tests

Commits

Will-Lo Jul 31, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Will-Lo Jul 31, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

phet Jul 31, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

phet Jul 31, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov-commenter commented Aug 1, 2023 • edited Loading

Codecov Report

umustafi commented Jul 28, 2023 •

edited

Loading

Will-Lo Jul 31, 2023 •

edited

Loading

Will-Lo Jul 31, 2023 •

edited

Loading

phet Jul 31, 2023 •

edited

Loading

phet Jul 31, 2023 •

edited

Loading

codecov-commenter commented Aug 1, 2023 •

edited

Loading