This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new acbe341c9c74 fix(test): cover SparkContext shutdown path in
HoodieMetadataTableValidator (#19349)
acbe341c9c74 is described below
commit acbe341c9c741eff55bd779d219f3b14fa868335
Author: Lokesh Jain <[email protected]>
AuthorDate: Fri Aug 14 14:20:37 2026 +0530
fix(test): cover SparkContext shutdown path in HoodieMetadataTableValidator
(#19349)
---
.../TestHoodieMetadataTableValidator.java | 102 +++++++++++++++++++++
1 file changed, 102 insertions(+)
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieMetadataTableValidator.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieMetadataTableValidator.java
index f3eaa254eeb5..af8e3b4d49c6 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieMetadataTableValidator.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieMetadataTableValidator.java
@@ -60,6 +60,7 @@ import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieCompactionConfig;
import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.exception.HoodieValidationException;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
@@ -81,6 +82,7 @@ import lombok.Setter;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import org.apache.spark.SparkException;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
@@ -1677,6 +1679,106 @@ public class TestHoodieMetadataTableValidator extends
HoodieSparkClientTestBase
fsFileSlices.size(), mdtFileSlices.size())));
}
+ @Test
+ void
testDoMetadataTableValidationThrowsHoodieExceptionOnSparkContextShutdown()
throws Exception {
+ Map<String, String> writeOptions = new HashMap<>();
+ writeOptions.put(DataSourceWriteOptions.TABLE_NAME().key(), "test_table");
+ writeOptions.put("hoodie.table.name", "test_table");
+ writeOptions.put(DataSourceWriteOptions.TABLE_TYPE().key(),
"MERGE_ON_READ");
+ writeOptions.put(DataSourceWriteOptions.RECORDKEY_FIELD().key(),
"_row_key");
+ writeOptions.put(DataSourceWriteOptions.PRECOMBINE_FIELD().key(),
"timestamp");
+ writeOptions.put(DataSourceWriteOptions.PARTITIONPATH_FIELD().key(),
"partition_path");
+
+ // Write with RLI enabled so checkMetadataTableIsAvailable() returns true
and
+ // doMetadataTableValidation() proceeds to call validateRecordIndex.
+ // File-slice validation flags are intentionally NOT set so
validateFilesInPartition
+ // is a no-op and cannot mask the SparkContext-shutdown exception we are
testing.
+ makeInsertDf("000", 5).write().format("hudi").options(writeOptions)
+ .option(DataSourceWriteOptions.OPERATION().key(),
WriteOperationType.BULK_INSERT.value())
+
.option(HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_ENABLE_PROP.key(),
"true")
+
.option(HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_MIN_FILE_GROUP_COUNT_PROP.key(),
"1")
+
.option(HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_MAX_FILE_GROUP_COUNT_PROP.key(),
"1")
+ .mode(SaveMode.Overwrite)
+ .save(basePath);
+
+ HoodieMetadataTableValidator.Config config = new
HoodieMetadataTableValidator.Config();
+ config.basePath = "file:" + basePath;
+ // Do NOT enable validateLatestFileSlices / validateAllFileGroups: those
call
+ // validateFilesInPartition inside a Spark map, and
HoodieValidationException from
+ // that path would be re-thrown before validateRecordIndex is ever reached.
+
+ // NOTE: static nested class, not anonymous, so it does NOT capture
+ // TestHoodieMetadataTableValidator.this (which is not Serializable). An
anonymous class
+ // would cause Spark's parallelize().map() to throw "Task not
serializable", which the
+ // outer catch converts to HoodieValidationException instead of
HoodieException.
+ HoodieMetadataTableValidator validator = new
SparkContextShutdownValidator(jsc, config);
+ HoodieException ex = assertThrows(HoodieException.class,
validator::doMetadataTableValidation);
+ assertFalse(ex instanceof HoodieValidationException,
+ "Expected HoodieException wrapping SparkContext shutdown, not
HoodieValidationException.");
+ }
+
+ @Test
+ void
testDoMetadataTableValidationThrowsHoodieValidationExceptionOnUnexpectedSparkFailure()
throws Exception {
+ Map<String, String> writeOptions = new HashMap<>();
+ writeOptions.put(DataSourceWriteOptions.TABLE_NAME().key(), "test_table");
+ writeOptions.put("hoodie.table.name", "test_table");
+ writeOptions.put(DataSourceWriteOptions.TABLE_TYPE().key(),
"MERGE_ON_READ");
+ writeOptions.put(DataSourceWriteOptions.RECORDKEY_FIELD().key(),
"_row_key");
+ writeOptions.put(DataSourceWriteOptions.PRECOMBINE_FIELD().key(),
"timestamp");
+ writeOptions.put(DataSourceWriteOptions.PARTITIONPATH_FIELD().key(),
"partition_path");
+
+ makeInsertDf("000", 5).write().format("hudi").options(writeOptions)
+ .option(DataSourceWriteOptions.OPERATION().key(),
WriteOperationType.BULK_INSERT.value())
+
.option(HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_ENABLE_PROP.key(),
"true")
+
.option(HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_MIN_FILE_GROUP_COUNT_PROP.key(),
"1")
+
.option(HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_MAX_FILE_GROUP_COUNT_PROP.key(),
"1")
+ .mode(SaveMode.Overwrite)
+ .save(basePath);
+
+ HoodieMetadataTableValidator.Config config = new
HoodieMetadataTableValidator.Config();
+ config.basePath = "file:" + basePath;
+
+ // Covers the false branch of else if
(ExceptionUtil.validateErrorMsg(...)):
+ // a SparkException with a non-cancellation message should still produce
+ // HoodieValidationException("Unexpected spark failure").
+ HoodieMetadataTableValidator validator = new
UnexpectedSparkFailureValidator(jsc, config);
+ HoodieValidationException ex =
assertThrows(HoodieValidationException.class,
validator::doMetadataTableValidation);
+ assertTrue(ex.getMessage().contains("Unexpected spark failure"));
+ }
+
+ /** Static nested class; does NOT capture the enclosing test instance (not
Serializable). */
+ private static final class SparkContextShutdownValidator extends
HoodieMetadataTableValidator {
+ private static final long serialVersionUID = 1L;
+
+ SparkContextShutdownValidator(JavaSparkContext jsc, Config cfg) {
+ super(jsc, cfg);
+ }
+
+ @Override
+ void validateRecordIndex(HoodieSparkEngineContext sparkEngineContext,
HoodieTableMetaClient metaClient) {
+ sneakyThrow(new SparkException("cancelled because SparkContext was shut
down"));
+ }
+ }
+
+ /** Covers the false branch of the cancellation-message check. */
+ private static final class UnexpectedSparkFailureValidator extends
HoodieMetadataTableValidator {
+ private static final long serialVersionUID = 1L;
+
+ UnexpectedSparkFailureValidator(JavaSparkContext jsc, Config cfg) {
+ super(jsc, cfg);
+ }
+
+ @Override
+ void validateRecordIndex(HoodieSparkEngineContext sparkEngineContext,
HoodieTableMetaClient metaClient) {
+ sneakyThrow(new SparkException("some unexpected spark error"));
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T extends Throwable> void sneakyThrow(Throwable e) throws T {
+ throw (T) e;
+ }
+
private void mockPartitionWithFiles(List<String> partition1, HoodieStorage
storage) throws IOException {
for (String partition : partition1) {
StoragePathInfo storagePathInfo = mock(StoragePathInfo.class);
