[GOBBLIN-1919] Simplify a few elements of MR-related job exec before reusing code in Temporal-based execution #3784

phet · 2023-09-21T17:01:09Z

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

Description

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

Modest changes before (re-)using it, to simplify impl and add JSON pretty-printing.

Tests

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

continues use of existing tests

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 · 2023-09-22T16:59:29Z

gobblin-utility/src/main/java/org/apache/gobblin/util/JobLauncherUtils.java

+
+    if (workUnit instanceof MultiWorkUnit) {
+      return JobLauncherUtils.flattenWorkUnits(((MultiWorkUnit) workUnit).getWorkUnits());
+    } else {


Small nit, we don't need else return after a return statement

I agree it's not necessary, but I find the parallel indentation of if/else to be clearer, hence preferable. TBH, I almost went w/ the ternary operator... but it seemed harder to read.

would you prefer that?

I'd prefer the current over ternary, I guess because there's quite a bit of casting involved here so it's not as easy to follow what's going on.

Will-Lo

LGTM for the most part just one small comment

… Temporal-based execution

…ns, plus encapsulated convience method `JobState.getJobIdFromProps`

Will-Lo · 2023-10-26T19:37:18Z

gobblin-cluster/src/main/java/org/apache/gobblin/cluster/GobblinHelixJobLauncher.java

@@ -539,7 +537,7 @@ private void deleteWorkUnitFromStateStore(String workUnitId, ParallelRunner stat
    Path workUnitFile = new Path(workUnitFilePath);
    final String fileName = workUnitFile.getName();
    final String storeName = workUnitFile.getParent().getName();
-    if (fileName.endsWith(MULTI_WORK_UNIT_FILE_EXTENSION)) {
+    if (fileName.endsWith(JobLauncherUtils.MULTI_WORK_UNIT_FILE_EXTENSION)) {


I see this sort of check a lot in this PR, does it make sense to make it a reusable function of JobLauncherUtils where we check if a given path/string is a multiworkunit file?

good call! I added

codecov-commenter · 2023-10-26T20:19:48Z

Codecov Report

Merging #3784 (8f5f632) into master (918716f) will decrease coverage by 0.09%.
The diff coverage is 26.24%.

@@             Coverage Diff              @@
##             master    #3784      +/-   ##
============================================
- Coverage     47.68%   47.60%   -0.09%     
+ Complexity    11052    11041      -11     
============================================
  Files          2155     2155              
  Lines         85260    85303      +43     
  Branches       9483     9484       +1     
============================================
- Hits          40656    40606      -50     
- Misses        40919    41011      +92     
- Partials       3685     3686       +1

Files	Coverage Δ
...pache/gobblin/configuration/ConfigurationKeys.java	`0.00% <ø> (ø)`
...rg/apache/gobblin/runtime/AbstractJobLauncher.java	`59.03% <ø> (ø)`
...in/java/org/apache/gobblin/runtime/JobContext.java	`77.12% <100.00%> (-0.13%)`	⬇️
.../apache/gobblin/source/workunit/MultiWorkUnit.java	`0.00% <0.00%> (ø)`
...in/java/org/apache/gobblin/cluster/SingleTask.java	`76.62% <0.00%> (ø)`
...main/java/org/apache/gobblin/runtime/JobState.java	`79.46% <90.90%> (-0.43%)`	⬇️
...pache/gobblin/cluster/GobblinHelixJobLauncher.java	`66.03% <14.28%> (-1.51%)`	⬇️
...blin/runtime/mapreduce/GobblinOutputCommitter.java	`52.50% <12.50%> (+8.05%)`	⬆️
...pache/gobblin/runtime/mapreduce/MRJobLauncher.java	`51.42% <65.00%> (-0.79%)`	⬇️
...ain/java/org/apache/gobblin/runtime/TaskState.java	`78.77% <52.94%> (-4.84%)`	⬇️
... and 3 more

... and 6 files with indirect coverage changes

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

…k-unit

* [GOBBLIN-1921] Properly handle reminder events (apache#3790) * Add millisecond level precision to timestamp cols & proper timezone conversion - existing tests pass with minor modifications * Handle reminder events properly * Fix compilation errors & add isReminder flag * Add unit tests * Address review comments * Add newline to address comment * Include reminder/original tag in logging * Clarify timezone issues in comment --------- Co-authored-by: Urmi Mustafi <umustafi@linkedin.com> * [GOBBLIN-1924] Reminder event flag true (apache#3795) * Set reminder event flag to true for reminders * Update unit tests * remove unused variable --------- Co-authored-by: Urmi Mustafi <umustafi@linkedin.com> * [GOBBLIN-1923] Add retention for lease arbiter table (apache#3792) * Add retention for lease arbiter table * Replace blocking thread with scheduled thread pool executor * Make Calendar instance thread-safe * Rename variables, make values more clear * Update timestamp related cols --------- Co-authored-by: Urmi Mustafi <umustafi@linkedin.com> * Change debug statements to info temporarily to debug (apache#3796) Co-authored-by: Urmi Mustafi <umustafi@linkedin.com> * [GOBBLIN-1926] Fix Reminder Event Epsilon Comparison (apache#3797) * Fix Reminder Event Epsilon Comparison * Add TODO comment --------- Co-authored-by: Urmi Mustafi <umustafi@linkedin.com> * [GOBBLIN-1930] Improve Multi-active related logs and metrics (apache#3800) * Improve Multi-active related logs and metrics * Add more metrics and logs around forwarding dag action to DagManager * Improve logs in response to review comments * Replace flow execution id with trigger timestamp from multi-active * Update flow action execution id within lease arbiter * Fix test & make Lease Statuses more lean * Update javadoc --------- Co-authored-by: Urmi Mustafi <umustafi@linkedin.com> * add dataset root in some common type of datasets * [GOBBLIN-1927] Add topic validation support in KafkaSource, and add TopicNameValidator (apache#3793) * * Add generic topic validation support * Add the first validator TopicNameValidator into the validator chain, as a refactor of existing codes * Refine to address comments * Refine --------- Co-authored-by: Tao Qin <tqin@linkedin.com> * [GOBBLIN-1931] Refactor dag action updating method & add clarifying comment (apache#3801) * Refactor dag action updating method & add clarifying comment * Log filtered out duplicate messages * logs and metrics for missing messages from change monitor * Only add gobblin.service prefix for dagActionStoreChangeMonitor --------- Co-authored-by: Urmi Mustafi <umustafi@linkedin.com> * [GOBBLIN-1934] Monitor High Level Consumer queue size (apache#3805) * Emit metrics to monitor high level consumer queue size * Empty commit to trigger tests * Use BlockingQueue.size() func instead of atomic integer array * Remove unused import & add DagActionChangeMonitor prefix to metric * Refactor to avoid repeating code * Make protected variables private where possible * Fix white space --------- Co-authored-by: Urmi Mustafi <umustafi@linkedin.com> * [GOBBLIN-1935] Skip null dag action types unable to be processed (apache#3807) * Skip over null dag actions from malformed messages * Add new metric for skipped messages --------- Co-authored-by: Urmi Mustafi <umustafi@linkedin.com> * [GOBBLIN-1922]Add function in Kafka Source to recompute workUnits for filtered partitions (apache#3798) * add function in Kafka Source to recompute workUnits for filtered partitions * address comments * set default min container value to 1 * add condition when create empty wu * update the condition * Expose functions to fetch record partitionColumn value (apache#3810) * [GOBBLIN-1938] preserve x bit in manifest file based copy (apache#3804) * preserve x bit in manifest file based copy * fix project structure preventing running unit tests from intellij * fix unit test * [GOBBLIN-1919] Simplify a few elements of MR-related job exec before reusing code in Temporal-based execution (apache#3784) * Simplify a few elements of MR-related job exec before reusing code in Temporal-based execution * Add JSON-ification to several foundational config-state representations, plus encapsulated convience method `JobState.getJobIdFromProps` * Update javadoc comments * Encapsulate check for whether a path has the extension of a multi-work-unit * [GOBBLIN-1939] Bump AWS version to use a compatible version of Jackson with Gobblin (apache#3809) * Bump AWS version to use a compatible version of jackson with Gobblin * use shared aws version * [GOBBLIN-1937] Quantify Missed Work Completed by Reminders (apache#3808) * Quantify Missed Work Completed by Reminders Also fix bug to filter out heartbeat events before extracting field * Refactor changeMonitorUtils & add delimiter to metrics prefix * Re-order params to group similar ones --------- Co-authored-by: Urmi Mustafi <umustafi@linkedin.com> * GOBBLIN-1933]Change the logic in completeness verifier to support multi reference tier (apache#3806) * address comments * use connectionmanager when httpclient is not cloesable * [GOBBLIN-1933] Change the logic in completeness verifier to support multi reference tier * add uite test * fix typo * change the javadoc * change the javadoc --------- Co-authored-by: Zihan Li <zihli@zihli-mn2.linkedin.biz> * [GOBBLIN-1943] Use AWS version 1.12.261 to fix a security vulnerability in the previous version (apache#3813) * [GOBBLIN-1941] Develop Temporal abstractions, including `Workload` for workflows of unbounded size through sub-workflow nesting (apache#3811) * Define `Workload` abstraction for Temporal workflows of unbounded size through sub-workflow nesting * Adjust Gobblin-Temporal configurability for consistency and abstraction * Define `WorkerConfig`, to pass the `TemporalWorker`'s configuration to the workflows and activities it hosts * Improve javadoc * Javadoc fixup * Minor changes * Update per review suggestions * Insert pause, to spread the load on the temporal server, before launch of each child workflow that may have direct leaves of its own * Appease findbugs by having `SeqSliceBackedWorkSpan::next` throw `NoSuchElementException` * Add comment * [GOBBLIN-1944] Add gobblin-temporal load generator for a single subsuming super-workflow with a configurable number of activities nested beneath (apache#3815) * Add gobblin-temporal load generator for a single subsuming super-workflow with a configurable number of activities nested beneath * Update per findbugs advice * Improve processing of int props * [GOBBLIN-1945] Implement Distributed Data Movement (DDM) Gobblin-on-Temporal `WorkUnit` evaluation (apache#3816) * Implement Distributed Data Movement (DDM) Gobblin-on-Temporal `WorkUnit` evaluation * Adjust work unit processing tuning for start-to-close timeout and nested execution branching * Rework `ProcessWorkUnitImpl` and fix `FileSystem` misuse; plus convenience abstractions to load `FileSystem`, `JobState`, and `StateStore<TaskState>` * Fix `FileSystem` resource lifecycle, uniquely name each workflow, and drastically reduce worker concurrent task execution * Heed findbugs advice * prep before commit * Improve processing of required props * Update comment in response to PR feedback * [GOBBLIN-1942] Create MySQL util class for re-usable methods and setup MysqlDagActio… (apache#3812) * Create MySQL util class for re-usable methods and setup MysqlDagActionStore retention * Add a java doc * Address review comments * Close scheduled executors on shutdown & clarify naming and comments * Remove extra period making config key invalid * implement Closeable * Use try with resources --------- Co-authored-by: Urmi Mustafi <umustafi@linkedin.com> * [Hotfix][GOBBLIN-1949] add option to detect malformed orc during commit (apache#3818) * add option to detect malformed ORC during commit phase * better logging * address comment * catch more generic exception * validate ORC file after close * move validate in between close and commit * syntax * whitespace * update log * [GOBBLIN-1948] Use same flowExecutionId across participants (apache#3819) * Use same flowExecutionId across participants * Set config field as well in new FlowSpec * Use gobblin util to create config * Rename function and move to util --------- Co-authored-by: Urmi Mustafi <umustafi@linkedin.com> * Allow extension of functions in GobblinMCEPublisher and customization of fileList file metrics are calculated for (apache#3820) * [GOBBLIN-1951] Emit GTE when deleting corrupted ORC files (apache#3821) * [GOBBLIN-1951] Emit GTE when deleting corrupted ORC files This commit adds ORC file validation during the commit phase and deletes corrupted files. It also includes a test for ORC file validation. * Linter fixes * Add framework and unit tests for DagActionStoreChangeMonitor (apache#3817) * Add framework and unit tests for DagActionStoreChangeMonitor * Add more test cases and validation * Add header for new file * Move FlowSpec static function to Utils class * Remove unused import * Fix compile error * Fix unit tests --------- Co-authored-by: Urmi Mustafi <umustafi@linkedin.com> * [GOBBLIN-1952] Make jobname shortening in GaaS more aggressive (apache#3822) * Make jobname shortening in GaaS more aggressive * Change long name prefix to flowgroup * Make KafkaTopicGroupingWorkUnitPacker pack with desired num of container (apache#3814) * Make KafkaTopicGroupingWorkUnitPacker pack with desired num of container * update comment * [GOBBLIN-1953] Add an exception message to orc writer validation GTE (apache#3826) * Fix FlowSpec Updating Function (apache#3823) * Fix FlowSpec Updating Function * makes Config object with FlowSpec mutable * adds unit test to ensure flow compiles after updating FlowSpec * ensure DagManager resilient to exceptions on leadership change * Only update Properties obj not Config to avoid GC overhead * Address findbugs error * Avoid updating or creating new FlowSpec objects by passing flowExecutionId directly to metadata * Remove changes that are not needed anymore * Add TODO to handle failed DagManager leadership change * Overload function and add more documentation --------- Co-authored-by: Urmi Mustafi <umustafi@linkedin.com> * Emit metric to tune LeaseArbiter Linger metric (apache#3824) * Monitor number of failed persisting leases to tune linger * Increase default linger and epsilon values * Add metric for lease persisting success * Rename metrics --------- Co-authored-by: Urmi Mustafi <umustafi@linkedin.com> * [GOBBLIN-1956]Make Kafka streaming pipeline be able to config the max poll records during runtime (apache#3827) * address comments * use connectionmanager when httpclient is not cloesable * add uite test * fix typo * [GOBBLIN-1956] Make Kafka streaming pipeline be able to config the max poll records during runtime * small refractor --------- Co-authored-by: Zihan Li <zihli@zihli-mn2.linkedin.biz> * Add semantics for failure on partial success (apache#3831) * Consistly handle Rest.li /flowexecutions KILL and RESUME actions (apache#3830) * [GOBBLIN-1957] GobblinOrcwriter improvements for large records (apache#3828) * WIP * Optimization to limit batchsize based on large record sizes * Address review * Use DB-qualified table ID as `IcebergTable` dataset descriptor (apache#3834) * [GOBBLIN-1961] Allow `IcebergDatasetFinder` to use separate names for source vs. destination-side DB and table (apache#3835) * Allow `IcebergDatasetFinder` to use separate names for source vs. destination-side DB and table * Adjust Mockito.verify to pass test * Prevent NPE in `FlowCompilationValidationHelper.validateAndHandleConcurrentExecution` (apache#3836) * Prevent NPE in `FlowCompilationValidationHelper.validateAndHandleConcurrentExecution` * improved `MultiHopFlowCompiler` javadoc * Delete Launch Action Events After Processing (apache#3837) * Delete launch action event after persisting * Fix default value for flowExecutionId retrieval from metadata map * Address review comments and add unit test * Code clean up --------- Co-authored-by: Urmi Mustafi <umustafi@linkedin.com> * [GOBBLIN-1960] Emit audit count after commit in IcebergMetadataWriter (apache#3833) * Emit audit count after commit in IcebergMetadataWriter * Unit tests by extracting to a post commit * Emit audit count first * find bugs complaint * [GOBBLIN-1967] Add external data node for generic ingress/egress on GaaS (apache#3838) * Add external data node for generic ingress/egress on GaaS * Address reviews and cleanup * Use URI representation for external dataset descriptor node * Fix error message in containing check * Address review * [GOBBLIN-1971] Allow `IcebergCatalog` to specify the `DatasetDescriptor` name for the `IcebergTable`s it creates (apache#3842) * Allow `IcebergCatalog` to specify the `DatasetDescriptor` name for the `IcebergTable`s it creates * small method javadoc * [GOBBLIN-1970] Consolidate processing dag actions to one code path (apache#3841) * Consolidate processing dag actions to one code path * Delete dag action in failure cases too * Distinguish metrics for startup * Refactor to avoid duplicated code and create static metrics proxy class * Remove DagManager checks that don't apply on startup * Add test to check kill/resume dag action removal after processing * Remove unused import * Initialize metrics proxy with Null Pattern --------- Co-authored-by: Urmi Mustafi <umustafi@linkedin.com> * [GOBBLIN-1972] Fix `CopyDataPublisher` to avoid committing post-publish WUs before they've actually run (apache#3844) * Fix `CopyDataPublisher` to avoid committing post-publish WUs before they've actually run * fixup findbugsMain * [GOBBLIN-1975] Keep job.name configuration immutable if specified on GaaS (apache#3847) * Revert "[GOBBLIN-1952] Make jobname shortening in GaaS more aggressive (apache#3822)" This reverts commit 5619a0a. * use configuration to keep specified jobname if enabled * Cleanup * [GOBBLIN-1974] Ensure Adhoc Flows can be Executed in Multi-active Scheduler state (apache#3846) * Ensure Adhoc Flows can be Executed in Multi-active Scheduler state * Only delete spec for adhoc flows & always after orchestration * Delete adhoc flows when dagManager is not present as well * Fix flaky test for scheduler * Add clarifying comment about failure recovery * Re-ordered private method * Move private methods again * Enforce sequential ordering of unit tests to make more reliable --------- Co-authored-by: Urmi Mustafi <umustafi@linkedin.com> * [GOBBLIN-1973] Change Manifest distcp logic to compare permissions of source and dest files even when source is older (apache#3845) * change should copy logic * Add tests, address review * Fix checkstyle * Remove unused imports * [GOBBLIN-1976] Allow an `IcebergCatalog` to override the `DatasetDescriptor` platform name for the `IcebergTable`s it creates (apache#3848) * Allow an `IcebergCatalog` to override the `DatasetDescriptor` platform for the `IcebergTable`s it creates * fixup javadoc * Log when `PasswordManager` fails to load any master password (apache#3849) * [GOBBLIN-1968] Temporal commit step integration (apache#3829) Add commit step to Gobblin temporal workflow for job publish * Add codeql analysis * Make gradle specific * Add codeql as part of build script * Initialize codeql * Use separate workflow for codeql instead with custom build function as autobuild seems to not work * Add jdk jar for global dependencies script --------- Co-authored-by: umustafi <umust77@gmail.com> Co-authored-by: Urmi Mustafi <umustafi@linkedin.com> Co-authored-by: Arjun <abora@linkedin.com> Co-authored-by: Tao Qin <35046097+wsarecv@users.noreply.github.com> Co-authored-by: Tao Qin <tqin@linkedin.com> Co-authored-by: Hanghang Nate Liu <nate.hanghang.liu@gmail.com> Co-authored-by: Andy Jiang <andy.jiang99@outlook.com> Co-authored-by: Kip Kohn <ckohn@linkedin.com> Co-authored-by: Zihan Li <zihli@linkedin.com> Co-authored-by: Zihan Li <zihli@zihli-mn2.linkedin.biz> Co-authored-by: Matthew Ho <mho@linkedin.com>

Will-Lo reviewed Sep 22, 2023

View reviewed changes

phet changed the title ~~Simplify a few elements of MR-related job exec before reusing code in Temporal-based execution~~ [GOBBLIN-1919] Simplify a few elements of MR-related job exec before reusing code in Temporal-based execution Sep 22, 2023

phet force-pushed the cleanup-mr-exec branch from d39ce8f to c09873c Compare October 26, 2023 06:18

phet added 2 commits October 26, 2023 10:24

Simplify a few elements of MR-related job exec before reusing code in…

8e6d27a

… Temporal-based execution

Add JSON-ification to several foundational config-state representatio…

ed0326c

…ns, plus encapsulated convience method `JobState.getJobIdFromProps`

phet force-pushed the cleanup-mr-exec branch from c09873c to ed0326c Compare October 26, 2023 17:24

Update javadoc comments

f591228

Will-Lo reviewed Oct 26, 2023

View reviewed changes

Encapsulate check for whether a path has the extension of a multi-wor…

8f5f632

…k-unit

Will-Lo approved these changes Oct 26, 2023

View reviewed changes

Will-Lo merged commit dce9e25 into apache:master Oct 27, 2023
6 checks passed

phet mentioned this pull request Feb 15, 2024

[GOBBLIN-1919] Rework a few more elements of MR-related job exec for reuse in Temporal-based execution #3879

Merged

4 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[GOBBLIN-1919] Simplify a few elements of MR-related job exec before reusing code in Temporal-based execution #3784

[GOBBLIN-1919] Simplify a few elements of MR-related job exec before reusing code in Temporal-based execution #3784

phet commented Sep 21, 2023 •

edited

Loading

Will-Lo Sep 22, 2023

phet Oct 26, 2023 •

edited

Loading

Will-Lo Oct 26, 2023

Will-Lo left a comment

Will-Lo Oct 26, 2023

phet Oct 26, 2023

codecov-commenter commented Oct 26, 2023 •

edited

Loading

[GOBBLIN-1919] Simplify a few elements of MR-related job exec before reusing code in Temporal-based execution #3784

[GOBBLIN-1919] Simplify a few elements of MR-related job exec before reusing code in Temporal-based execution #3784

Conversation

phet commented Sep 21, 2023 • edited Loading

JIRA

Description

Tests

Commits

Will-Lo Sep 22, 2023

Choose a reason for hiding this comment

phet Oct 26, 2023 • edited Loading

Choose a reason for hiding this comment

Will-Lo Oct 26, 2023

Choose a reason for hiding this comment

Will-Lo left a comment

Choose a reason for hiding this comment

Will-Lo Oct 26, 2023

Choose a reason for hiding this comment

phet Oct 26, 2023

Choose a reason for hiding this comment

codecov-commenter commented Oct 26, 2023 • edited Loading

Codecov Report

phet commented Sep 21, 2023 •

edited

Loading

phet Oct 26, 2023 •

edited

Loading

codecov-commenter commented Oct 26, 2023 •

edited

Loading