[HUDI-6967] Add clearJobStatus api in HoodieEngineContext by wecharyu · Pull Request #9899 · apache/hudi

wecharyu · 2023-10-22T15:21:47Z

Change Logs

Add clearJobStatus api in HoodieEngineContext
Add call of clearJobStatus after setJobStatus.

Impact

Fix the incorrect job group and descriptions.

Risk level (write none, low medium or high below)

None.

Documentation Update

Describe any necessary documentation update if there is any new feature, config, or user-facing change

The config description must be updated if new configs are added or the default value of the configs are changed
Any new feature or user-facing change requires updating the Hudi website. Please create a Jira ticket, attach the
ticket number here and follow the instruction to make
changes to the website.

Contributor's checklist

Read through contributor's guide
Change Logs and Impact were stated clearly
Adequate tests were added if applicable
CI passed

yihua · 2023-10-23T18:56:38Z

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndex.java

    String keyField = hoodieTable.getMetaClient().getTableConfig().getRecordKeyFieldProp();

    List<Pair<String, HoodieBaseFile>> baseFilesForAllPartitions = HoodieIndexUtils.getLatestBaseFilesForAllPartitions(partitions, context, hoodieTable);
+    context.clearJobStatus();


This shouldn't be added. Key range loading has not finished here.

The code after here will not create new job, it seems OK to clear job status here. WDYT?

yihua · 2023-10-23T19:00:15Z

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java

        }

        // Now delete partially written files
-        context.setJobStatus(this.getClass().getSimpleName(), "Delete all partially written files: " + config.getTableName());


Why deleting this?

This status will be overwritten in HoodieTable#deleteInvalidFilesByPartitions, so just delete it.

yihua · 2023-10-23T19:06:00Z

...nt/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BaseWriteHelper.java

        // perform index loop up to get existing location of records
        context.setJobStatus(this.getClass().getSimpleName(), "Tagging: " + table.getConfig().getTableName());
        taggedRecords = tag(dedupedRecords, context, table);
+        context.clearJobStatus();


If lazy execution happens afterwards, the job status may not be properly populated. Have you verified all places that this won't happen?

Let me check all lazy execution. For this one, "Tagging xxx" status will also populate todeduplicateRecords, but clear here will not affect other jobs, so we retain this line.

yihua · 2023-10-23T19:09:50Z

...src/main/java/org/apache/hudi/table/action/bootstrap/SparkBootstrapCommitActionExecutor.java

-            partitionFsPair.getRight().getLeft(), keyGenerator));
+              partitionFsPair.getRight().getLeft(), keyGenerator));
+    } finally {
+      context.clearJobStatus();


This method composes a DAG and is triggered by lazy execution.

Will remove clearJobStatus of lazy execution in CommitActionExecutor, because it will clear job status finally:

hudi/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala

Lines 499 to 505 in 051eb0e

} finally {

// close the write client in all cases

val asyncCompactionEnabled = isAsyncCompactionEnabled(writeClient, tableConfig, parameters, jsc.hadoopConfiguration())

val asyncClusteringEnabled = isAsyncClusteringEnabled(writeClient, parameters)

if (!asyncCompactionEnabled && !asyncClusteringEnabled) {

log.info("Closing write client")

writeClient.close()

yihua · 2023-10-23T19:11:06Z

...-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java

+      }
      return recordsAndPendingClusteringFileGroups.getLeft();
+    } finally {
+      context.clearJobStatus();


Could you check here for lazy execution too?

yihua · 2023-10-23T19:15:17Z

hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java

+        };
+      });
+    } finally {
+      engineContext.clearJobStatus();


This may be subject to lazy execution.

Remove this clearJobStatus.

yihua · 2023-10-23T19:15:52Z

hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java

-    });
+      });
+    } finally {
+      engineContext.clearJobStatus();


Check this one too for lazy execution.

yihua · 2023-10-23T19:16:27Z

...-spark-datasource/hudi-spark/src/main/java/org/apache/hudi/cli/HDFSParquetImporterUtils.java

+                new HoodieJsonPayload(genericRecord.toString()));
+          });
+    } finally {
+      context.clearJobStatus();


This one too.

yihua · 2023-10-23T19:16:45Z

hudi-utilities/src/main/java/org/apache/hudi/utilities/HDFSParquetImporter.java

+                new HoodieJsonPayload(genericRecord.toString()));
+          });
+    } finally {
+      context.clearJobStatus();


This one too.

yihua · 2023-10-23T19:18:48Z

hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieSnapshotExporter.java

+            executorOutputFs.getConf());
+      }, parallelism);
+    } finally {
+      context.clearJobStatus();


This should be at the end of the method, correct? since context.foreach also triggers Spark stages?

wecharyu · 2023-10-24T09:34:49Z

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndex.java

    String keyField = hoodieTable.getMetaClient().getTableConfig().getRecordKeyFieldProp();

    List<Pair<String, HoodieBaseFile>> baseFilesForAllPartitions = HoodieIndexUtils.getLatestBaseFilesForAllPartitions(partitions, context, hoodieTable);
+    context.clearJobStatus();


The code after here will not create new job, it seems OK to clear job status here. WDYT?

wecharyu · 2023-10-24T09:41:20Z

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java

        }

        // Now delete partially written files
-        context.setJobStatus(this.getClass().getSimpleName(), "Delete all partially written files: " + config.getTableName());


This status will be overwritten in HoodieTable#deleteInvalidFilesByPartitions, so just delete it.

wecharyu · 2023-10-24T10:25:51Z

...nt/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BaseWriteHelper.java

        // perform index loop up to get existing location of records
        context.setJobStatus(this.getClass().getSimpleName(), "Tagging: " + table.getConfig().getTableName());
        taggedRecords = tag(dedupedRecords, context, table);
+        context.clearJobStatus();


Let me check all lazy execution. For this one, "Tagging xxx" status will also populate todeduplicateRecords, but clear here will not affect other jobs, so we retain this line.

wecharyu · 2023-10-25T03:57:24Z

...src/main/java/org/apache/hudi/table/action/bootstrap/SparkBootstrapCommitActionExecutor.java

-            partitionFsPair.getRight().getLeft(), keyGenerator));
+              partitionFsPair.getRight().getLeft(), keyGenerator));
+    } finally {
+      context.clearJobStatus();


Will remove clearJobStatus of lazy execution in CommitActionExecutor, because it will clear job status finally:

hudi/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala

Lines 499 to 505 in 051eb0e

} finally {

// close the write client in all cases

val asyncCompactionEnabled = isAsyncCompactionEnabled(writeClient, tableConfig, parameters, jsc.hadoopConfiguration())

val asyncClusteringEnabled = isAsyncClusteringEnabled(writeClient, parameters)

if (!asyncCompactionEnabled && !asyncClusteringEnabled) {

log.info("Closing write client")

writeClient.close()

wecharyu · 2023-10-25T04:01:49Z

...-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java

+      }
      return recordsAndPendingClusteringFileGroups.getLeft();
+    } finally {
+      context.clearJobStatus();


wecharyu · 2023-10-25T06:27:51Z

hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java

+        };
+      });
+    } finally {
+      engineContext.clearJobStatus();


Remove this clearJobStatus.

wecharyu · 2023-10-25T06:29:09Z

...-spark-datasource/hudi-spark/src/main/java/org/apache/hudi/cli/HDFSParquetImporterUtils.java

+                new HoodieJsonPayload(genericRecord.toString()));
+          });
+    } finally {
+      context.clearJobStatus();


wecharyu · 2023-10-25T06:33:00Z

hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieSnapshotExporter.java

+            executorOutputFs.getConf());
+      }, parallelism);
+    } finally {
+      context.clearJobStatus();


wecharyu · 2023-10-25T06:48:16Z

hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java

-    return Pair.of(schemaProvider, Pair.of(checkpointStr, records));
+      return Pair.of(schemaProvider, Pair.of(checkpointStr, records));
+    } finally {
+      hoodieSparkContext.clearJobStatus();


This one seems also lazy execution.

wecharyu · 2023-10-25T06:49:36Z

hudi-utilities/src/main/java/org/apache/hudi/utilities/HDFSParquetImporter.java

+                new HoodieJsonPayload(genericRecord.toString()));
+          });
+    } finally {
+      context.clearJobStatus();


hudi-bot · 2023-10-25T09:36:59Z

CI report:

c72b4cf Azure: FAILURE

Bot commands

@hudi-bot supports the following commands:

@hudi-bot run azure re-run the last Azure build

nsivabalan · 2023-11-16T13:25:56Z

hey @yihua and @wecharyu : can we target this for 0.14.1 ? Is this patch landable in a weeks time?

yihua reviewed Oct 23, 2023

View reviewed changes

wecharyu commented Oct 25, 2023

View reviewed changes

wecharyu added 2 commits October 25, 2023 15:05

[HUDI-6967] Add clearJobStatus api in HoodieEngineContext

973ef0f

remove clearJobStatus for lazy execution

c72b4cf

wecharyu force-pushed the HUDI-6967 branch from 6827d7a to c72b4cf Compare October 25, 2023 07:05

nsivabalan assigned yihua Nov 15, 2023

nsivabalan added priority:critical Production degraded; pipelines stalled release-0.14.1 labels Nov 15, 2023

github-actions bot added the size:XL PR with lines of changes > 1000 label Feb 26, 2024

hudi-bot mentioned this pull request Dec 9, 2025

Add clearJobStatus api in HoodieEngineContext #16271

Open

yihua mentioned this pull request Jan 8, 2026

fix(spark): Add clearJobStatus() calls after setJobStatus() operations #17451

Merged

3 tasks

	} finally {
	// close the write client in all cases
	val asyncCompactionEnabled = isAsyncCompactionEnabled(writeClient, tableConfig, parameters, jsc.hadoopConfiguration())
	val asyncClusteringEnabled = isAsyncClusteringEnabled(writeClient, parameters)
	if (!asyncCompactionEnabled && !asyncClusteringEnabled) {
	log.info("Closing write client")
	writeClient.close()

Conversation

wecharyu commented Oct 22, 2023

Change Logs

Impact

Risk level (write none, low medium or high below)

Documentation Update

Contributor's checklist

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hudi-bot commented Oct 25, 2023

CI report:

Uh oh!

nsivabalan commented Nov 16, 2023

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!