Skip to content

[MINOR] fix(test): cover SparkContext shutdown path in HoodieMetadataTableValidator #19349

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
lokeshj1703 wants to merge 1 commit into
base: master
Choose a base branch
from
+102 −0
Open

[MINOR] fix(test): cover SparkContext shutdown path in HoodieMetadataTableValidator #19349

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
102 changes: 102 additions & 0 deletions hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieMetadataTableValidator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,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.exception.HoodieException;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.exception.HoodieValidationException;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
Expand All @@ -74,6 +75,7 @@
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;
Expand Down Expand Up @@ -1670,6 +1672,106 @@ void testValidateFileSlicesTruncation() {
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 {

@hudi-agent hudi-agent Jul 22, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 nit: could you add a brief Javadoc on sneakyThrow explaining the constraint that drives it? Something like /** SparkException is checked; overrides cannot widen the throws clause, so this unchecked-cast rethrow is needed. */ would save the next reader a "why not just throw it?" moment.

⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.

@voonhous voonhous Jul 24, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1

throw (T) e;
}

private void mockPartitionWithFiles(List<String> partition1, HoodieStorage storage) throws IOException {
for (String partition : partition1) {
StoragePathInfo storagePathInfo = mock(StoragePathInfo.class);
Expand Down
Loading