[GOBBLIN-1970] Consolidate processing dag actions to one code path

[umustafi]

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-1970

Description

• Here are some details about my PR, including screenshots (if applicable):
  We have similar code in the DagActionStoreChangeMonitor and DagManager to process dag actions from the DagActionStore. There have been small discrepancies between the code in each area leading to unexpected bugs in how the actions are processed, so to fix forward and make this easier to maintain with code in one place we consolidate all logic relating to the dag action processing to the DagActionStoreChangeMonitor. The processing in the change monitor is working correctly, while the handling of launch events in the DagManager is failing compilation most likely due to an error loading the job templates when initializing the compiler but we're not able to identify the exact issue. 

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

[Will-Lo]

Can you digress what you mean by this comment? A bit confused

[umustafi]

I'll remove this for now unless we see other problems but basically the idea was I can do a join of dag actions with completed leases in the lease arbiter table to exclude actions that have already been completed but it's only an interim soln if this doesn't work and kind of hacky

[Will-Lo]

What happens when the non-leaders run this?

[umustafi]

They should be ignored because DagManager is not active - see sample check. We shouldn't have access to whether or not the DagManager is active at the monitor stage ie: later when DagManager is active for all hosts they will do lease arbitration over this.

[Will-Lo]

I missed this in a prior PR, but did we handle the non-happy launch path to ensure that we still delete the dag actions?

[umustafi]

No we don't. I guess that is worth doing since we can't compile/add them so no point repeatedly accumulating and trying them.

[codecov-commenter]

Codecov Report

Attention: 67 lines in your changes are missing coverage. Please review.

Comparison is base (694ed37) 47.58% compared to head (2afd806) 47.56%.
Report is 1 commits behind head on master.

Files	Patch %	Lines
...ervice/monitoring/DagActionStoreChangeMonitor.java	0.00%	63 Missing ⚠️
...monitoring/DagActionStoreChangeMonitorFactory.java	0.00%	4 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master    #3841      +/-   ##
============================================
- Coverage     47.58%   47.56%   -0.03%     
+ Complexity    11050    11045       -5     
============================================
  Files          2159     2159              
  Lines         85429    85431       +2     
  Branches       9493     9491       -2     
============================================
- Hits          40649    40632      -17     
- Misses        41080    41101      +21     
+ Partials       3700     3698       -2     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[phet]

feels like duplication... could this code and processMessage each call on a method in common to handle this action-specific dispatch?

[phet]

given the only thing isLoadedOnStartup determines is which meter to increment, I recommend to codify that:

@Data
private class SubmissionMetricProxy {
  private final ContextAwareMeter successMeter;
  private final ContextAwareMeter failureMeter;

  public static SubmissionMetricProxy ON_STARTUP = new SubmissionMetricProxy(
DagActionStoreChangeMonitor.this.successfulLaunchSubmissionsOnStartup,
DagActionStoreChangeMonitor.this.failedLaunchSubmissionsOnStartup);

  public static SubmissionMetricProxy POST_STARTUP = ...;

  public void markSuccess() {
    getSuccessMeter.mark();
  }
  public void markFailure() {
    getFailureMeter.mark();
  }
}

then create an overloaded version:

protected void submitFlow(DagAction dagAction, boolean isStartup) {
  submitFlow(dagAction, isStartup ? SubmissionMetricProxy.ON_STARTUP : SubmissionMetricProxy.POST_STARTUP);
}

protected void submitFlow(DagAction dagAction, SubmissionMetricProxy smp) {
  ...
}

[umustafi]

https://github.com/umustafi/gobblin/pull/13/checks?sha=8553a62e7965439e10f47ec5a91d91f686aff099 checks on my fork

[phet]

putting the initialization here means NPE during any invocation of submitFlowToDagManagerHelper that may come prior.

if you can't rework, I suggest at first to temporarily initialize these according to the Null Pattern:

LaunchSubmissionMetricProxy ON_STARTUP = new LaunchSubmissionMetricProxy {
  @Override public void markSuccess() { }
  @Override public void markFailure() { }
}

[umustafi]

Created a Null LaunchSubmissionMetricProxy class to initialize these proxies with until the metrics are setup.

[phet]

the other alternative is to use the Null/NoOp Pattern in the call site, e.g.:

protected void submitFlow(DagAction dagAction, boolean isStartup) {
  submitFlow(dagAction, (isStartup ? ON_STARTUP : POST_STARTUP).orElse(new LaunchSubmissionMetricProxy { /* as in above comment */}));
}

i.e. make these two volatile Optional<LSMP>

[phet]

looks good!

[arjun4084346]

On the left side, this piece of code was written with switch-case, now it is changed to if-else. Why?
IntelliJ complains when if-else is used.