[GOBBLIN-1782] Fix Merge State for Flow Pending Resume statuses

[meethngala]

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-XXX
  • https://issues.apache.org/jira/browse/GOBBLIN-1782

Description

• Here are some details about my PR, including screenshots (if applicable):
• Flow statuses in pending_resume state were not getting executed since we incorrectly update the merge state for such flows
• Thus, we now detect such flows and update their merge state so that they can proceed to their next state in the DAG

Tests

• My PR adds the following unit tests OR does not need testing for this extremely good reason:
• Added unit test for the fix

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"

[codecov-commenter]

Codecov Report

Merging #3639 (6c258c5) into master (69f3f9c) will decrease coverage by 0.01%.
The diff coverage is 0.00%.

@@             Coverage Diff              @@
##             master    #3639      +/-   ##
============================================
- Coverage     46.58%   46.58%   -0.01%     
- Complexity    10672    10676       +4     
============================================
  Files          2133     2133              
  Lines         83557    83560       +3     
  Branches       9290     9292       +2     
============================================
+ Hits          38928    38929       +1     
- Misses        41068    41069       +1     
- Partials       3561     3562       +1     

Impacted Files	Coverage Δ
...blin/service/monitoring/KafkaJobStatusMonitor.java	45.25% <0.00%> (-1.02%)	⬇️
...g/apache/gobblin/hive/orc/HiveOrcSerDeManager.java	57.29% <0.00%> (-3.13%)	⬇️
.../org/apache/gobblin/cluster/GobblinTaskRunner.java	62.22% <0.00%> (-0.31%)	⬇️
...anagement/copy/replication/ConfigBasedDataset.java	68.87% <0.00%> (ø)
...odules/orchestration/InMemoryUserQuotaManager.java	70.86% <0.00%> (+0.78%)	⬆️
.../apache/gobblin/runtime/api/JobExecutionState.java	80.37% <0.00%> (+0.93%)	⬆️
...he/gobblin/source/PartitionAwareFileRetriever.java	55.55% <0.00%> (+7.40%)	⬆️
...a/org/apache/gobblin/util/limiter/NoopLimiter.java	60.00% <0.00%> (+20.00%)	⬆️

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

[Will-Lo]

can we abstract this check into a function similar to how JobStatusRetriever has this check?

[Will-Lo]

Also, since the side effect of this change is similar to the else case, would it be simpler for us to append to the previous if statement and make an initial check that the currentStatus != a flow status and pending resume?

[meethngala]

yeah for sure! I have abstracted the logic to check for flow statuses and in pending resume in my latest commit. Regarding the second comment, I have added the logic for currentStatus != pending resume in the abstracted method itself... so that in case we want to extend it to other status types in the future we can just do it in the method and not change the if condition often... wdyt?

[Will-Lo]

LGTM! Awesome work :)