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 8af07e1e1534 fix(utilities): validator Config.equals and
drop-partitions hive sync, with tests for the standalone tools (#19875)
8af07e1e1534 is described below
commit 8af07e1e1534c92e1ba97dbc1e18a50f73f6c65c
Author: voonhous <[email protected]>
AuthorDate: Thu Sep 10 16:59:31 2026 +0800
fix(utilities): validator Config.equals and drop-partitions hive sync, with
tests for the standalone tools (#19875)
Three bugs in the standalone hudi-utilities tools, found while adding
tests for tools that had no coverage at all: HoodieDropPartitionsTool,
HoodieDataTableValidator, TableSizeStats and HoodieTTLJob.
- HoodieDataTableValidator.Config.equals cast its argument to the
sibling HoodieMetadataTableValidator.Config, so comparing any two
instances threw ClassCastException. It now casts to its own Config.
All four tools compare basePath through Objects.equals (a default
instance NPEd) and drop help from hashCode, which equals never
compared, so every Config hashes over exactly the fields it
compares.
- HoodieDropPartitionsTool --sync-hive-meta could never work:
buildHiveSyncProps stored two ConfigProperty objects as Properties
keys, so HiveSyncConfig threw ClassCastException before any
metastore contact. It now uses the keys and string values for the
boolean flags.
- HoodieDropPartitionsTool validated the hive arguments only after
masking the partitions, so a typo in --hive-database cost the
partitions first. The DELETE arm now verifies them before anything
is dropped (dry runs unchanged), and --hive-partition-field is
required with --sync-hive-meta: its empty default was written into
the sync props verbatim, which blocked the inference from the table
config and made HiveSyncTool skip every partition.
Tests, one class per tool, each writing a small three-partition COW
table and running the tool in-process:
- TestHoodieDropPartitionsTool (10): dry-run reports the file ids of
exactly the named partitions and changes nothing; delete writes one
replacecommit masking exactly those partitions; props sourcing;
unsupported mode; hive arguments verified before the drop; a failing
metastore connection after the drop is committed; Config contracts.
- TestHoodieDataTableValidator (7): healthy table; dangling base file
with and without --ignore-failed, asserted through the validator's
log; extra file for a completed commit; continuous mode stopping on
the first failure, bounded by a timeout; Config contracts.
- TestTableSizeStats (15): table and partition stats; the date
interval filters including the half-open bounds and a partition
dated yesterday for --num-days; base paths from --props-path; the
error branches; Config contracts.
- TestHoodieTTLJob (2): both constructors; expired partitions are
replaced and the fresh one survives.
- Shared test utilities: CapturingLogAppender for tools whose only
output is their log, and ToolTestUtils for the helpers the classes
had in common.
Each fix has a test that fails without it. main() of each tool and
HudiHiveSyncJob (needs a metastore) stay uncovered on purpose.
---
.../hudi/utilities/HoodieDataTableValidator.java | 6 +-
.../hudi/utilities/HoodieDropPartitionsTool.java | 21 +-
.../utilities/HoodieMetadataTableValidator.java | 4 +-
.../org/apache/hudi/utilities/TableSizeStats.java | 4 +-
.../utilities/TestHoodieDataTableValidator.java | 229 +++++++++++++
.../utilities/TestHoodieDropPartitionsTool.java | 358 +++++++++++++++++++++
.../TestHoodieMetadataTableValidator.java | 18 ++
.../apache/hudi/utilities/TestHoodieTTLJob.java | 130 ++++++++
.../apache/hudi/utilities/TestTableSizeStats.java | 311 ++++++++++++++++++
.../utilities/testutils/CapturingLogAppender.java | 85 +++++
.../hudi/utilities/testutils/ToolTestUtils.java | 59 ++++
11 files changed, 1212 insertions(+), 13 deletions(-)
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieDataTableValidator.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieDataTableValidator.java
index ed689022da64..4c89816fbede 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieDataTableValidator.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieDataTableValidator.java
@@ -212,8 +212,8 @@ public class HoodieDataTableValidator implements
Serializable {
if (o == null || getClass() != o.getClass()) {
return false;
}
- HoodieMetadataTableValidator.Config config =
(HoodieMetadataTableValidator.Config) o;
- return basePath.equals(config.basePath)
+ Config config = (Config) o;
+ return Objects.equals(basePath, config.basePath)
&& Objects.equals(continuous, config.continuous)
&& Objects.equals(minValidateIntervalSeconds,
config.minValidateIntervalSeconds)
&& Objects.equals(parallelism, config.parallelism)
@@ -228,7 +228,7 @@ public class HoodieDataTableValidator implements
Serializable {
@Override
public int hashCode() {
return Objects.hash(basePath, continuous, minValidateIntervalSeconds,
parallelism, ignoreFailed, sparkMaster, sparkMemory,
- assumeDatePartitioning, propsFilePath, configs, help);
+ assumeDatePartitioning, propsFilePath, configs);
}
}
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieDropPartitionsTool.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieDropPartitionsTool.java
index 9b6a89c8dae9..0999f16ed438 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieDropPartitionsTool.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieDropPartitionsTool.java
@@ -244,7 +244,7 @@ public class HoodieDropPartitionsTool implements
Serializable {
return false;
}
Config config = (Config) o;
- return basePath.equals(config.basePath)
+ return Objects.equals(basePath, config.basePath)
&& Objects.equals(runningMode, config.runningMode)
&& Objects.equals(tableName, config.tableName)
&& Objects.equals(partitions, config.partitions)
@@ -270,7 +270,7 @@ public class HoodieDropPartitionsTool implements
Serializable {
return Objects.hash(basePath, runningMode, tableName, partitions,
syncToHive, hiveDataBase, hiveTableName, hiveUserName, hivePassWord,
hiveURL,
hivePartitionsField, hiveUseJdbc, hiveHMSUris,
partitionValueExtractorClass,
- sparkMaster, sparkMemory, propsFilePath, configs,
hiveSyncIgnoreException, help);
+ sparkMaster, sparkMemory, propsFilePath, configs,
hiveSyncIgnoreException);
}
}
@@ -304,6 +304,11 @@ public class HoodieDropPartitionsTool implements
Serializable {
switch (mode) {
case DELETE:
log.info(" ****** The Hoodie Drop Partitions Tool is in delete mode
****** ");
+ if (cfg.syncToHive) {
+ // Check the hive configs before anything is dropped: they are
otherwise only read once the partitions
+ // have already been masked, so a typo in --hive-database would
cost the partitions before it surfaces.
+ verifyHiveConfigs();
+ }
doDeleteTablePartitions();
syncToHiveIfNecessary();
break;
@@ -357,12 +362,12 @@ public class HoodieDropPartitionsTool implements
Serializable {
props.put(DataSourceWriteOptions.HIVE_PASS().key(), cfg.hivePassWord);
props.put(DataSourceWriteOptions.HIVE_URL().key(), cfg.hiveURL);
props.put(DataSourceWriteOptions.HIVE_PARTITION_FIELDS().key(),
cfg.hivePartitionsField);
- props.put(DataSourceWriteOptions.HIVE_USE_JDBC().key(), cfg.hiveUseJdbc);
+ props.put(DataSourceWriteOptions.HIVE_USE_JDBC().key(),
String.valueOf(cfg.hiveUseJdbc));
props.put(DataSourceWriteOptions.HIVE_SYNC_MODE().key(), cfg.hiveSyncMode);
- props.put(DataSourceWriteOptions.HIVE_IGNORE_EXCEPTIONS().key(),
cfg.hiveSyncIgnoreException);
+ props.put(DataSourceWriteOptions.HIVE_IGNORE_EXCEPTIONS().key(),
String.valueOf(cfg.hiveSyncIgnoreException));
props.put(DataSourceWriteOptions.HIVE_PASS().key(), cfg.hivePassWord);
- props.put(HiveSyncConfig.META_SYNC_BASE_PATH, cfg.basePath);
- props.put(HiveSyncConfig.META_SYNC_BASE_FILE_FORMAT, "PARQUET");
+ props.put(HiveSyncConfig.META_SYNC_BASE_PATH.key(), cfg.basePath);
+ props.put(HiveSyncConfig.META_SYNC_BASE_FILE_FORMAT.key(), "PARQUET");
props.put(DataSourceWriteOptions.PARTITIONS_TO_DELETE().key(),
cfg.partitions);
props.put(DataSourceWriteOptions.HIVE_PARTITION_EXTRACTOR_CLASS().key(),
cfg.partitionValueExtractorClass);
props.put(KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key(),
cfg.hivePartitionsField);
@@ -373,6 +378,10 @@ public class HoodieDropPartitionsTool implements
Serializable {
private void verifyHiveConfigs() {
ValidationUtils.checkArgument(!StringUtils.isNullOrEmpty(cfg.hiveDataBase),
"Hive database name couldn't be null or empty when enable sync meta, please set
--hive-database/-db.");
ValidationUtils.checkArgument(!StringUtils.isNullOrEmpty(cfg.hiveTableName),
"Hive table name couldn't be null or empty when enable sync meta, please set
--hive-table-name/-tn.");
+ // This is written into the sync props verbatim, which stops
HoodieSyncConfig inferring the fields from the
+ // table config; left empty, hive sync skips every partition and the drop
is never reflected in the metastore.
+
ValidationUtils.checkArgument(!StringUtils.isNullOrEmpty(cfg.hivePartitionsField),
+ "Hive partition fields couldn't be null or empty when enable sync
meta, otherwise no partition is synced, please set --hive-partition-field.");
}
private void syncHive(HiveSyncConfig hiveSyncConfig) {
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableValidator.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableValidator.java
index fa99795c22fc..95ac651eda56 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableValidator.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableValidator.java
@@ -457,7 +457,7 @@ public class HoodieMetadataTableValidator implements
Serializable {
return false;
}
Config config = (Config) o;
- return basePath.equals(config.basePath)
+ return Objects.equals(basePath, config.basePath)
&& Objects.equals(continuous, config.continuous)
&& Objects.equals(skipDataFilesForCleaning,
config.skipDataFilesForCleaning)
&& Objects.equals(validateLatestFileSlices,
config.validateLatestFileSlices)
@@ -491,7 +491,7 @@ public class HoodieMetadataTableValidator implements
Serializable {
validateSecondaryIndex, validateRecordIndexCount,
validateRecordIndexContent, numRecordIndexErrorSamples,
viewStorageTypeForFSListing, viewStorageTypeForMetadata,
minValidateIntervalSeconds, parallelism, recordIndexParallelism,
ignoreFailed,
- sparkMaster, sparkMemory, assumeDatePartitioning,
logDetailMaxLength, propsFilePath, configs, help);
+ sparkMaster, sparkMemory, assumeDatePartitioning,
logDetailMaxLength, propsFilePath, configs);
}
}
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/TableSizeStats.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/TableSizeStats.java
index 9a5bca95c318..fccc19cf1937 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/TableSizeStats.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/TableSizeStats.java
@@ -204,7 +204,7 @@ public class TableSizeStats implements Serializable {
return false;
}
Config config = (Config) o;
- return basePath.equals(config.basePath)
+ return Objects.equals(basePath, config.basePath)
&& Objects.equals(numDays, config.numDays)
&& Objects.equals(startDate, config.startDate)
&& Objects.equals(endDate, config.endDate)
@@ -219,7 +219,7 @@ public class TableSizeStats implements Serializable {
@Override
public int hashCode() {
- return Objects.hash(basePath, numDays, startDate, endDate, tableStats,
partitionStats, parallelism, sparkMaster, sparkMemory, propsFilePath, configs,
help);
+ return Objects.hash(basePath, numDays, startDate, endDate, tableStats,
partitionStats, parallelism, sparkMaster, sparkMemory, propsFilePath, configs);
}
}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieDataTableValidator.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieDataTableValidator.java
new file mode 100644
index 000000000000..81d12a02bf78
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieDataTableValidator.java
@@ -0,0 +1,229 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.utilities;
+
+import org.apache.hudi.client.SparkRDDWriteClient;
+import org.apache.hudi.client.WriteClientTestUtils;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+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.testutils.HoodieSparkClientTestBase;
+import org.apache.hudi.utilities.testutils.CapturingLogAppender;
+
+import org.apache.spark.api.java.JavaRDD;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH;
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_SECOND_PARTITION_PATH;
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_THIRD_PARTITION_PATH;
+import static org.apache.hudi.utilities.testutils.ToolTestUtils.stackMessages;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests {@link HoodieDataTableValidator} against a small three-partition COW
table, with and without
+ * data files that the timeline does not account for.
+ */
+public class TestHoodieDataTableValidator extends HoodieSparkClientTestBase {
+
+ private static final int RECORDS_PER_PARTITION = 4;
+
+ private HoodieDataTableValidator.Config validatorConfig(boolean
ignoreFailed) {
+ HoodieDataTableValidator.Config cfg = new
HoodieDataTableValidator.Config();
+ cfg.basePath = basePath;
+ cfg.parallelism = 2;
+ cfg.ignoreFailed = ignoreFailed;
+ return cfg;
+ }
+
+ private String writeOneCommit() {
+ HoodieWriteConfig writeConfig = getConfigBuilder().build();
+ try (SparkRDDWriteClient client = getHoodieWriteClient(writeConfig)) {
+ String instantTime = WriteClientTestUtils.createNewInstantTime();
+ List<HoodieRecord> records = new ArrayList<>();
+ for (String partition : Arrays.asList(
+ DEFAULT_FIRST_PARTITION_PATH, DEFAULT_SECOND_PARTITION_PATH,
DEFAULT_THIRD_PARTITION_PATH)) {
+ records.addAll(dataGen.generateInsertsForPartition(instantTime,
RECORDS_PER_PARTITION, partition));
+ }
+ WriteClientTestUtils.startCommitWithTime(client, instantTime);
+ JavaRDD<WriteStatus> writeStatuses =
client.insert(jsc.parallelize(records, 1), instantTime);
+ client.commit(instantTime, writeStatuses);
+ return instantTime;
+ }
+ }
+
+ /**
+ * Copies an existing base file of the first partition to a new base file
named after {@code instantTime} and a
+ * brand new file id, which is exactly the shape of a data file the timeline
does not account for.
+ */
+ private String addUnaccountedBaseFile(String instantTime) throws IOException
{
+ Path partitionDir = Paths.get(basePath, DEFAULT_FIRST_PARTITION_PATH);
+ Path source;
+ try (Stream<Path> files = Files.list(partitionDir)) {
+ source = files.filter(p -> p.toString().endsWith(".parquet")).findFirst()
+ .orElseThrow(() -> new IllegalStateException("no base file written
under " + partitionDir));
+ }
+ String danglingName =
+ FSUtils.makeBaseFileName(instantTime, "1-0-1",
UUID.randomUUID().toString(), ".parquet");
+ Files.copy(source, partitionDir.resolve(danglingName));
+ return danglingName;
+ }
+
+ @Test
+ public void testValidationPassesOnAHealthyTable() {
+ writeOneCommit();
+ HoodieDataTableValidator validator = new HoodieDataTableValidator(jsc,
validatorConfig(false));
+ // the validator reports through an exception only, so a clean table is
asserted by the absence of one
+ assertDoesNotThrow(validator::run);
+ }
+
+ @Test
+ public void testMissingPropsFileFails() {
+ HoodieDataTableValidator.Config cfg = validatorConfig(false);
+ cfg.propsFilePath =
tempDir.resolve("does-not-exist.properties").toAbsolutePath().toString();
+ assertThrows(HoodieIOException.class, () -> new
HoodieDataTableValidator(jsc, cfg));
+ }
+
+ /**
+ * A base file whose instant time precedes the first instant of the active
timeline is dangling; whether that
+ * fails the job depends on --ignore-failed.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testDanglingFileBeforeTheActiveTimeline(boolean ignoreFailed)
throws IOException {
+ writeOneCommit();
+ String danglingFile = addUnaccountedBaseFile("00000000000001");
+
+ HoodieDataTableValidator validator = new HoodieDataTableValidator(jsc,
validatorConfig(ignoreFailed));
+ if (ignoreFailed) {
+ // the run survives, but the finding still has to be reported
+ List<String> messages;
+ try (CapturingLogAppender logs =
CapturingLogAppender.attachTo(HoodieDataTableValidator.class)) {
+ assertDoesNotThrow(validator::run);
+ messages = logs.messages();
+ }
+ assertTrue(messages.contains(
+ "Data table validation failed due to dangling files count 1, found
before active timeline"),
+ messages.toString());
+ assertTrue(messages.stream().anyMatch(m -> m.startsWith("Dangling file:
") && m.endsWith(danglingFile)),
+ "the dangling file must be named in " + messages);
+ assertTrue(messages.contains("Data table validation failed."),
messages.toString());
+ } else {
+ HoodieException thrown = assertThrows(HoodieException.class,
validator::run);
+ assertTrue(thrown.getCause() instanceof HoodieValidationException, "got
" + thrown.getCause());
+ assertTrue(thrown.getCause().getMessage().contains("dangling files 1"),
thrown.getCause().getMessage());
+ }
+ }
+
+ /**
+ * A base file carrying the instant time of a completed commit, but absent
from that commit's metadata, is an
+ * extra file and fails the second check.
+ */
+ @Test
+ public void testExtraFileForCompletedCommitFailsValidation() throws
IOException {
+ String instantTime = writeOneCommit();
+ addUnaccountedBaseFile(instantTime);
+
+ HoodieDataTableValidator validator = new HoodieDataTableValidator(jsc,
validatorConfig(false));
+ HoodieException thrown = assertThrows(HoodieException.class,
validator::run);
+ assertTrue(thrown.getCause() instanceof HoodieValidationException, "got "
+ thrown.getCause());
+ assertTrue(thrown.getCause().getMessage().contains("dangling files 1"),
thrown.getCause().getMessage());
+ // the table itself is untouched by validation
+ assertEquals(1,
HoodieTableMetaClient.reload(metaClient).getActiveTimeline()
+ .filterCompletedInstants().countInstants());
+ }
+
+ /**
+ * In continuous mode the async service keeps validating until it fails;
with --ignore-failed off the very
+ * first round throws, which is what stops the job.
+ */
+ @Test
+ @Timeout(value = 2, unit = TimeUnit.MINUTES)
+ public void testContinuousModeStopsOnValidationFailure() throws IOException {
+ writeOneCommit();
+ addUnaccountedBaseFile("00000000000001");
+
+ HoodieDataTableValidator.Config cfg = validatorConfig(false);
+ cfg.continuous = true;
+ cfg.minValidateIntervalSeconds = 1;
+ HoodieDataTableValidator validator = new HoodieDataTableValidator(jsc,
cfg);
+
+ HoodieException thrown = assertThrows(HoodieException.class,
validator::run);
+ assertTrue(stackMessages(thrown).contains("dangling files 1"),
stackMessages(thrown));
+ }
+
+ @Test
+ public void testConfigEqualsHashCodeAndToString() {
+ HoodieDataTableValidator.Config cfg = validatorConfig(true);
+ cfg.basePath = "/tmp/table";
+ cfg.continuous = true;
+ cfg.minValidateIntervalSeconds = 30;
+
+ assertEquals(cfg, cfg);
+ assertNotEquals(cfg, null);
+ assertNotEquals(cfg, "not a config");
+ // a Config straight out of JCommander has no base path yet
+ assertEquals(new HoodieDataTableValidator.Config(), new
HoodieDataTableValidator.Config());
+ assertEquals(new HoodieDataTableValidator.Config().hashCode(), new
HoodieDataTableValidator.Config().hashCode());
+ // --help is not compared, so it must not be hashed either
+ HoodieDataTableValidator.Config askedForHelp = new
HoodieDataTableValidator.Config();
+ askedForHelp.help = true;
+ assertEquals(new HoodieDataTableValidator.Config(), askedForHelp);
+ assertEquals(new HoodieDataTableValidator.Config().hashCode(),
askedForHelp.hashCode());
+
+ HoodieDataTableValidator.Config same = validatorConfig(true);
+ same.basePath = "/tmp/table";
+ same.continuous = true;
+ same.minValidateIntervalSeconds = 30;
+ assertEquals(cfg, same);
+ assertEquals(cfg.hashCode(), same.hashCode());
+
+ same.minValidateIntervalSeconds = 60;
+ assertNotEquals(cfg, same);
+ assertNotEquals(cfg.hashCode(), same.hashCode());
+
+ String printed = cfg.toString();
+ assertTrue(printed.contains("--base-path /tmp/table"));
+ assertTrue(printed.contains("--continuous true"));
+ assertTrue(printed.contains("--ignore-failed true"));
+ assertTrue(printed.contains("--min-validate-interval-seconds 30"));
+ }
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieDropPartitionsTool.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieDropPartitionsTool.java
new file mode 100644
index 000000000000..25315924fdbf
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieDropPartitionsTool.java
@@ -0,0 +1,358 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.utilities;
+
+import org.apache.hudi.client.SparkRDDWriteClient;
+import org.apache.hudi.client.WriteClientTestUtils;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieReplaceCommitMetadata;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.view.FileSystemViewManager;
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.hive.HoodieHiveSyncException;
+import org.apache.hudi.testutils.HoodieSparkClientTestBase;
+import org.apache.hudi.utilities.testutils.CapturingLogAppender;
+
+import org.apache.spark.api.java.JavaRDD;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH;
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_SECOND_PARTITION_PATH;
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_THIRD_PARTITION_PATH;
+import static
org.apache.hudi.utilities.testutils.ToolTestUtils.latestBaseFileCount;
+import static org.apache.hudi.utilities.testutils.ToolTestUtils.stackMessages;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests {@link HoodieDropPartitionsTool} against a small three-partition COW
table.
+ */
+public class TestHoodieDropPartitionsTool extends HoodieSparkClientTestBase {
+
+ private static final int RECORDS_PER_PARTITION = 4;
+
+ private HoodieDropPartitionsTool.Config toolConfig(String mode, String
partitions) {
+ HoodieDropPartitionsTool.Config cfg = new
HoodieDropPartitionsTool.Config();
+ cfg.basePath = basePath;
+ cfg.tableName = metaClient.getTableConfig().getTableName();
+ cfg.runningMode = mode;
+ cfg.partitions = partitions;
+ cfg.parallelism = 2;
+ cfg.configs.add(HoodieWriteConfig.TBL_NAME.key() + "=" + cfg.tableName);
+ return cfg;
+ }
+
+ /**
+ * Writes two insert commits: the first spreads records over all three
partitions, the second adds a
+ * second file slice to the first partition.
+ */
+ private void writeThreePartitionTable() {
+ HoodieWriteConfig writeConfig = getConfigBuilder().build();
+ try (SparkRDDWriteClient client = getHoodieWriteClient(writeConfig)) {
+ String firstCommit = WriteClientTestUtils.createNewInstantTime();
+ List<HoodieRecord> firstBatch = new ArrayList<>();
+ for (String partition : Arrays.asList(
+ DEFAULT_FIRST_PARTITION_PATH, DEFAULT_SECOND_PARTITION_PATH,
DEFAULT_THIRD_PARTITION_PATH)) {
+ firstBatch.addAll(dataGen.generateInsertsForPartition(firstCommit,
RECORDS_PER_PARTITION, partition));
+ }
+ writeBatchAndCommit(client, firstCommit, firstBatch);
+
+ String secondCommit = WriteClientTestUtils.createNewInstantTime();
+ writeBatchAndCommit(client, secondCommit,
+ dataGen.generateInsertsForPartition(secondCommit,
RECORDS_PER_PARTITION, DEFAULT_FIRST_PARTITION_PATH));
+ }
+ }
+
+ private void writeBatchAndCommit(SparkRDDWriteClient client, String
instantTime, List<HoodieRecord> records) {
+ WriteClientTestUtils.startCommitWithTime(client, instantTime);
+ JavaRDD<WriteStatus> writeStatuses =
client.insert(jsc.parallelize(records, 1), instantTime);
+ client.commit(instantTime, writeStatuses);
+ }
+
+ private List<String> latestFileIds(String partition) {
+ HoodieTableMetaClient reloaded = HoodieTableMetaClient.reload(metaClient);
+ try (HoodieTableFileSystemView fsView =
FileSystemViewManager.createInMemoryFileSystemView(
+ context, reloaded,
HoodieMetadataConfig.newBuilder().enable(false).build())) {
+ return
fsView.getLatestBaseFiles(partition).map(HoodieBaseFile::getFileId).collect(Collectors.toList());
+ }
+ }
+
+ private List<String> completedInstants() {
+ return
HoodieTableMetaClient.reload(metaClient).getActiveTimeline().filterCompletedInstants()
+
.getInstantsAsStream().map(HoodieInstant::requestedTime).collect(Collectors.toList());
+ }
+
+ @Test
+ public void
testDryRunReportsTheFilesItWouldDeleteAndLeavesTheTableUntouched() {
+ writeThreePartitionTable();
+ List<String> instantsBefore = completedInstants();
+ // what the tool prints must be the file ids the two partitions really hold
+ Set<String> expectedReport = new HashSet<>(Arrays.asList(
+ "Partitions : " + DEFAULT_FIRST_PARTITION_PATH + ", corresponding data
file IDs : "
+ + latestFileIds(DEFAULT_FIRST_PARTITION_PATH),
+ "Partitions : " + DEFAULT_SECOND_PARTITION_PATH + ", corresponding
data file IDs : "
+ + latestFileIds(DEFAULT_SECOND_PARTITION_PATH)));
+
+ HoodieDropPartitionsTool.Config cfg = toolConfig("dry_run",
+ DEFAULT_FIRST_PARTITION_PATH + "," + DEFAULT_SECOND_PARTITION_PATH);
+ List<String> messages;
+ try (CapturingLogAppender logs =
CapturingLogAppender.attachTo(HoodieDropPartitionsTool.class)) {
+ new HoodieDropPartitionsTool(jsc, cfg).run();
+ messages = logs.messages();
+ }
+
+ assertTrue(messages.contains("Data files and partitions to delete : "),
messages.toString());
+ assertEquals(expectedReport,
+ messages.stream().filter(m -> m.startsWith("Partitions :
")).collect(Collectors.toSet()));
+ assertTrue(messages.stream().noneMatch(m ->
m.contains(DEFAULT_THIRD_PARTITION_PATH)),
+ "the partition that was not named must not be reported: " + messages);
+
+ assertEquals(instantsBefore, completedInstants(), "dry run must not add
any instant");
+ assertEquals(1, latestBaseFileCount(context, metaClient,
DEFAULT_FIRST_PARTITION_PATH));
+ assertEquals(1, latestBaseFileCount(context, metaClient,
DEFAULT_SECOND_PARTITION_PATH));
+ assertEquals(1, latestBaseFileCount(context, metaClient,
DEFAULT_THIRD_PARTITION_PATH));
+ }
+
+ @Test
+ public void testDeleteMasksOnlyTheRequestedPartitions() throws IOException {
+ writeThreePartitionTable();
+ int instantsBefore = completedInstants().size();
+
+ HoodieDropPartitionsTool.Config cfg = toolConfig("delete",
+ DEFAULT_FIRST_PARTITION_PATH + "," + DEFAULT_SECOND_PARTITION_PATH);
+ new HoodieDropPartitionsTool(jsc, cfg).run();
+
+ HoodieTableMetaClient reloaded = HoodieTableMetaClient.reload(metaClient);
+ assertEquals(instantsBefore + 1, completedInstants().size(), "delete must
add exactly one instant");
+ HoodieInstant replaceInstant =
reloaded.getActiveTimeline().getCompletedReplaceTimeline().lastInstant().get();
+ HoodieReplaceCommitMetadata replaceMetadata =
+ reloaded.getActiveTimeline().readReplaceCommitMetadata(replaceInstant);
+ assertEquals(
+ new HashSet<>(Arrays.asList(DEFAULT_FIRST_PARTITION_PATH,
DEFAULT_SECOND_PARTITION_PATH)),
+ replaceMetadata.getPartitionToReplaceFileIds().keySet());
+ // the file group of the first partition, written by both commits, is
masked
+ assertEquals(1,
replaceMetadata.getPartitionToReplaceFileIds().get(DEFAULT_FIRST_PARTITION_PATH).size());
+
+ assertEquals(0, latestBaseFileCount(context, metaClient,
DEFAULT_FIRST_PARTITION_PATH));
+ assertEquals(0, latestBaseFileCount(context, metaClient,
DEFAULT_SECOND_PARTITION_PATH));
+ assertEquals(1, latestBaseFileCount(context, metaClient,
DEFAULT_THIRD_PARTITION_PATH),
+ "the partition that was not named must survive");
+ }
+
+ /**
+ * The tool takes its write properties either from --props or from repeated
--hoodie-conf, and only defaults
+ * hoodie.meta.fields.mode from the table when the operator did not name it.
Both sources are checked by asking
+ * for a meta-fields mode the table does not have and expecting the write
config gate to reject it.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testWritePropertiesComeFromPropsFileAndHoodieConf(boolean
usePropsFile) throws IOException {
+ writeThreePartitionTable();
+
+ HoodieDropPartitionsTool.Config cfg = toolConfig("dry_run",
DEFAULT_THIRD_PARTITION_PATH);
+ String metaFieldsOverride = "hoodie.meta.fields.mode=NONE";
+ if (usePropsFile) {
+ // the file carries the mode, the --hoodie-conf entry already on the
config carries the table name, so
+ // both sources have to be merged for this run to reach the write config
gate
+ Path propsFile = tempDir.resolve("drop-partitions.properties");
+ Files.write(propsFile, Collections.singletonList(metaFieldsOverride),
StandardCharsets.UTF_8);
+ cfg.propsFilePath = propsFile.toAbsolutePath().toString();
+ } else {
+ cfg.configs.add(metaFieldsOverride);
+ }
+
+ HoodieDropPartitionsTool tool = new HoodieDropPartitionsTool(jsc, cfg);
+ Throwable thrown = assertThrows(HoodieException.class, tool::run);
+ assertTrue(stackMessages(thrown).contains("hoodie.meta.fields.mode"),
+ "expected the meta fields mode from the config source to reach the
write config, got: " + thrown);
+ }
+
+ @Test
+ public void testUnsupportedModeFails() {
+ writeThreePartitionTable();
+ HoodieDropPartitionsTool.Config cfg = toolConfig("purge",
DEFAULT_THIRD_PARTITION_PATH);
+ HoodieDropPartitionsTool tool = new HoodieDropPartitionsTool(jsc, cfg);
+
+ HoodieException thrown = assertThrows(HoodieException.class, tool::run);
+ assertTrue(thrown.getMessage().contains("Unable to delete table partitions
in " + basePath));
+ assertTrue(thrown.getCause() instanceof IllegalArgumentException, "got " +
thrown.getCause());
+ assertEquals(0,
HoodieTableMetaClient.reload(metaClient).getActiveTimeline()
+ .getCompletedReplaceTimeline().countInstants());
+ }
+
+ /**
+ * A missing --hive-database is caught before the delete runs, so the
partitions are still there afterwards.
+ */
+ @Test
+ public void testHiveSyncConfigIsVerifiedBeforeTheDrop() {
+ writeThreePartitionTable();
+ HoodieDropPartitionsTool.Config cfg = toolConfig("delete",
DEFAULT_THIRD_PARTITION_PATH);
+ cfg.syncToHive = true;
+ cfg.hiveDataBase = null;
+ HoodieDropPartitionsTool tool = new HoodieDropPartitionsTool(jsc, cfg);
+
+ HoodieException thrown = assertThrows(HoodieException.class, tool::run);
+ assertTrue(thrown.getCause() instanceof IllegalArgumentException, "got " +
thrown.getCause());
+ assertTrue(thrown.getCause().getMessage().contains("--hive-database"));
+ assertEquals(0,
HoodieTableMetaClient.reload(metaClient).getActiveTimeline()
+ .getCompletedReplaceTimeline().countInstants(), "nothing may be
dropped once the hive configs are bad");
+ assertEquals(1, latestBaseFileCount(context, metaClient,
DEFAULT_THIRD_PARTITION_PATH));
+ }
+
+ /**
+ * With the hive configs in place the sync props are built and the sync is
attempted for real; pointing it at a
+ * port nothing listens on keeps the test free of a metastore. The drop is
committed before that attempt, so a
+ * metastore that is down costs the sync, not the partitions.
+ */
+ @Test
+ public void testHiveSyncFailureLeavesTheDropCommitted() {
+ // the tool feeds the FileSystem's hadoop conf into the HiveConf, which is
the only way in for these
+ jsc.hadoopConfiguration().set("hive.metastore.connect.retries", "1");
+ jsc.hadoopConfiguration().set("hive.metastore.client.connect.retry.delay",
"0s");
+ jsc.hadoopConfiguration().set("hive.metastore.failure.retries", "0");
+ writeThreePartitionTable();
+ HoodieDropPartitionsTool.Config cfg = toolConfig("delete",
DEFAULT_THIRD_PARTITION_PATH);
+ cfg.syncToHive = true;
+ cfg.hiveDataBase = "db";
+ cfg.hiveTableName = "tbl";
+ cfg.hivePartitionsField = "partition_path";
+ cfg.hiveHMSUris = "thrift://localhost:1";
+ HoodieDropPartitionsTool tool = new HoodieDropPartitionsTool(jsc, cfg);
+
+ HoodieException thrown = assertThrows(HoodieException.class, tool::run);
+ assertTrue(thrown.getCause() instanceof HoodieHiveSyncException, "got " +
thrown.getCause());
+ assertTrue(stackMessages(thrown).contains("Failed to create
HiveMetaStoreClient"), stackMessages(thrown));
+ assertTrue(stackMessages(thrown).contains("Could not connect to meta store
using any of the URIs provided"),
+ stackMessages(thrown));
+
+ assertEquals(1,
HoodieTableMetaClient.reload(metaClient).getActiveTimeline()
+ .getCompletedReplaceTimeline().countInstants(), "the drop is committed
before hive sync runs");
+ assertEquals(0, latestBaseFileCount(context, metaClient,
DEFAULT_THIRD_PARTITION_PATH));
+ }
+
+ /**
+ * Dry run touches no partition, so it does not need the hive configs to be
sound: it still prints its listing.
+ */
+ @Test
+ public void testDryRunDoesNotNeedHiveConfigs() {
+ writeThreePartitionTable();
+ List<String> instantsBefore = completedInstants();
+ HoodieDropPartitionsTool.Config cfg = toolConfig("dry_run",
DEFAULT_THIRD_PARTITION_PATH);
+ cfg.syncToHive = true;
+ cfg.hiveDataBase = null;
+
+ List<String> messages;
+ try (CapturingLogAppender logs =
CapturingLogAppender.attachTo(HoodieDropPartitionsTool.class)) {
+ new HoodieDropPartitionsTool(jsc, cfg).run();
+ messages = logs.messages();
+ }
+
+ assertEquals(
+ Collections.singleton("Partitions : " + DEFAULT_THIRD_PARTITION_PATH +
", corresponding data file IDs : "
+ + latestFileIds(DEFAULT_THIRD_PARTITION_PATH)),
+ messages.stream().filter(m -> m.startsWith("Partitions :
")).collect(Collectors.toSet()));
+ assertEquals(instantsBefore, completedInstants(), "dry run must not add
any instant");
+ }
+
+ /**
+ * The partition fields are written into the sync props verbatim; empty,
hive sync silently skips every
+ * partition, so the tool refuses the run rather than dropping partitions
the metastore never hears about.
+ */
+ @Test
+ public void testHiveSyncWithoutPartitionFieldIsRejectedBeforeTheDrop() {
+ writeThreePartitionTable();
+ HoodieDropPartitionsTool.Config cfg = toolConfig("delete",
DEFAULT_THIRD_PARTITION_PATH);
+ cfg.syncToHive = true;
+ cfg.hiveDataBase = "db";
+ cfg.hiveTableName = "tbl";
+ // cfg.hivePartitionsField is left at its default, the empty string
+ HoodieDropPartitionsTool tool = new HoodieDropPartitionsTool(jsc, cfg);
+
+ HoodieException thrown = assertThrows(HoodieException.class, tool::run);
+ assertTrue(thrown.getCause() instanceof IllegalArgumentException, "got " +
thrown.getCause());
+
assertTrue(thrown.getCause().getMessage().contains("--hive-partition-field"),
+ thrown.getCause().getMessage());
+ assertEquals(0,
HoodieTableMetaClient.reload(metaClient).getActiveTimeline()
+ .getCompletedReplaceTimeline().countInstants(), "nothing may be
dropped once the hive configs are bad");
+ assertEquals(1, latestBaseFileCount(context, metaClient,
DEFAULT_THIRD_PARTITION_PATH));
+ }
+
+ @Test
+ public void testConfigEqualsHashCodeAndToString() {
+ HoodieDropPartitionsTool.Config left = new
HoodieDropPartitionsTool.Config();
+ left.basePath = "/tmp/table";
+ left.runningMode = "delete";
+ left.tableName = "t1";
+ left.partitions = "p1,p2";
+ left.configs = new ArrayList<>(Collections.singletonList("k=v"));
+
+ HoodieDropPartitionsTool.Config right = new
HoodieDropPartitionsTool.Config();
+ right.basePath = "/tmp/table";
+ right.runningMode = "delete";
+ right.tableName = "t1";
+ right.partitions = "p1,p2";
+ right.configs = new ArrayList<>(Collections.singletonList("k=v"));
+
+ assertEquals(left, left);
+ assertEquals(left, right);
+ assertEquals(left.hashCode(), right.hashCode());
+ assertNotEquals(left, null);
+ assertNotEquals(left, "not a config");
+ // a Config straight out of JCommander has no base path yet
+ assertEquals(new HoodieDropPartitionsTool.Config(), new
HoodieDropPartitionsTool.Config());
+ assertEquals(new HoodieDropPartitionsTool.Config().hashCode(), new
HoodieDropPartitionsTool.Config().hashCode());
+ // --help is not compared, so it must not be hashed either
+ HoodieDropPartitionsTool.Config askedForHelp = new
HoodieDropPartitionsTool.Config();
+ askedForHelp.help = true;
+ assertEquals(new HoodieDropPartitionsTool.Config(), askedForHelp);
+ assertEquals(new HoodieDropPartitionsTool.Config().hashCode(),
askedForHelp.hashCode());
+
+ right.hiveDataBase = "db";
+ assertNotEquals(left, right);
+ assertNotEquals(left.hashCode(), right.hashCode());
+
+ String printed = left.toString();
+ assertTrue(printed.contains("--base-path /tmp/table"));
+ assertTrue(printed.contains("--partitions p1,p2"));
+ assertTrue(printed.contains("--hoodie-conf [k=v]"));
+ assertTrue(printed.contains("--hive-user-name Masked"), "credentials must
not be printed");
+ }
+}
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 af8e3b4d49c6..26ca1b16278c 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
@@ -167,6 +167,24 @@ public class TestHoodieMetadataTableValidator extends
HoodieSparkClientTestBase
});
}
+ @Test
+ public void testConfigEqualsAndHashCodeIgnoreHelp() {
+ HoodieMetadataTableValidator.Config config = new
HoodieMetadataTableValidator.Config();
+ config.basePath = "/tmp/table";
+ HoodieMetadataTableValidator.Config askedForHelp = new
HoodieMetadataTableValidator.Config();
+ askedForHelp.basePath = "/tmp/table";
+ askedForHelp.help = true;
+
+ // --help is not compared, so it must not be hashed either
+ assertEquals(config, askedForHelp);
+ assertEquals(config.hashCode(), askedForHelp.hashCode());
+
+ // a Config straight out of JCommander has no base path yet
+ assertEquals(new HoodieMetadataTableValidator.Config(), new
HoodieMetadataTableValidator.Config());
+ assertEquals(new HoodieMetadataTableValidator.Config().hashCode(),
+ new HoodieMetadataTableValidator.Config().hashCode());
+ }
+
@Test
public void testAggregateColumnStats() {
HoodieColumnRangeMetadata<Comparable> fileColumn1Range1 =
HoodieColumnRangeMetadata.<Comparable>create(
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieTTLJob.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieTTLJob.java
new file mode 100644
index 000000000000..846c11c91729
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieTTLJob.java
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.utilities;
+
+import org.apache.hudi.client.SparkRDDWriteClient;
+import org.apache.hudi.client.WriteClientTestUtils;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieReplaceCommitMetadata;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.config.HoodieCleanConfig;
+import org.apache.hudi.config.HoodieTTLConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.testutils.HoodieSparkClientTestBase;
+
+import org.apache.spark.api.java.JavaRDD;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH;
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_SECOND_PARTITION_PATH;
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_THIRD_PARTITION_PATH;
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.getCommitTimeAtUTC;
+import static
org.apache.hudi.utilities.testutils.ToolTestUtils.latestBaseFileCount;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+/**
+ * Tests {@link HoodieTTLJob} on a table whose partitions were written at
different times.
+ */
+public class TestHoodieTTLJob extends HoodieSparkClientTestBase {
+
+ private static final int RECORDS_PER_PARTITION = 4;
+
+ /**
+ * Both constructors are covered: with an explicit props/meta client pair,
and with the (jsc, cfg) constructor
+ * that has to read --props and --hoodie-conf itself.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testTtlDropsOnlyExpiredPartitions(boolean
readPropsFromFileSystem) throws IOException {
+ writeOnePartitionPerInstant();
+
+ HoodieTTLJob.Config cfg = new HoodieTTLJob.Config();
+ cfg.basePath = basePath;
+ cfg.parallelism = 2;
+
+ HoodieTTLJob job;
+ if (readPropsFromFileSystem) {
+ Path propsFile = tempDir.resolve("ttl.properties");
+ Files.write(propsFile, Arrays.asList(
+ HoodieWriteConfig.TBL_NAME.key() + "=" +
metaClient.getTableConfig().getTableName(),
+ HoodieTTLConfig.PARTITION_TTL_STRATEGY_TYPE.key() +
"=KEEP_BY_TIME"), StandardCharsets.UTF_8);
+ cfg.propsFilePath = propsFile.toAbsolutePath().toString();
+ cfg.configs.add(HoodieTTLConfig.DAYS_RETAIN.key() + "=10");
+ job = new HoodieTTLJob(jsc, cfg);
+ } else {
+ TypedProperties props = new TypedProperties();
+ props.setProperty(HoodieWriteConfig.TBL_NAME.key(),
metaClient.getTableConfig().getTableName());
+ props.setProperty(HoodieTTLConfig.PARTITION_TTL_STRATEGY_TYPE.key(),
"KEEP_BY_TIME");
+ props.setProperty(HoodieTTLConfig.DAYS_RETAIN.key(), "10");
+ job = new HoodieTTLJob(jsc, cfg, props, metaClient);
+ // the job turns async cleaning off on the properties it was handed
+ assertEquals("false",
props.get(HoodieCleanConfig.ASYNC_CLEAN.key()).toString());
+ }
+
+ job.run();
+
+ HoodieTableMetaClient reloaded = HoodieTableMetaClient.reload(metaClient);
+ HoodieInstant replaceInstant =
+
reloaded.getActiveTimeline().getCompletedReplaceTimeline().lastInstant().get();
+ HoodieReplaceCommitMetadata replaceMetadata =
+ reloaded.getActiveTimeline().readReplaceCommitMetadata(replaceInstant);
+ assertEquals(
+ new HashSet<>(Arrays.asList(DEFAULT_FIRST_PARTITION_PATH,
DEFAULT_SECOND_PARTITION_PATH)),
+ replaceMetadata.getPartitionToReplaceFileIds().keySet(),
+ "only the partitions older than the retention are dropped");
+
+ assertEquals(0, latestBaseFileCount(context, metaClient,
DEFAULT_FIRST_PARTITION_PATH));
+ assertEquals(0, latestBaseFileCount(context, metaClient,
DEFAULT_SECOND_PARTITION_PATH));
+ assertEquals(1, latestBaseFileCount(context, metaClient,
DEFAULT_THIRD_PARTITION_PATH),
+ "the fresh partition must survive");
+
assertFalse(replaceMetadata.getPartitionToReplaceFileIds().containsKey(DEFAULT_THIRD_PARTITION_PATH));
+ }
+
+ private void writeOnePartitionPerInstant() {
+ HoodieWriteConfig writeConfig = getConfigBuilder().build();
+ try (SparkRDDWriteClient client = getHoodieWriteClient(writeConfig)) {
+ // two partitions written far in the past, one written now
+ writeRecordsForPartition(client, DEFAULT_FIRST_PARTITION_PATH,
getCommitTimeAtUTC(0));
+ writeRecordsForPartition(client, DEFAULT_SECOND_PARTITION_PATH,
getCommitTimeAtUTC(1000));
+ writeRecordsForPartition(client, DEFAULT_THIRD_PARTITION_PATH,
WriteClientTestUtils.createNewInstantTime());
+ }
+ }
+
+ private void writeRecordsForPartition(SparkRDDWriteClient client, String
partition, String instantTime) {
+ List<HoodieRecord> records =
+ new ArrayList<>(dataGen.generateInsertsForPartition(instantTime,
RECORDS_PER_PARTITION, partition));
+ WriteClientTestUtils.startCommitWithTime(client, instantTime);
+ JavaRDD<WriteStatus> writeStatuses =
client.insert(jsc.parallelize(records, 1), instantTime);
+ client.commit(instantTime, writeStatuses);
+ }
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestTableSizeStats.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestTableSizeStats.java
new file mode 100644
index 000000000000..d40a851befa1
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestTableSizeStats.java
@@ -0,0 +1,311 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.utilities;
+
+import org.apache.hudi.client.SparkRDDWriteClient;
+import org.apache.hudi.client.WriteClientTestUtils;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.testutils.HoodieSparkClientTestBase;
+import org.apache.hudi.utilities.testutils.CapturingLogAppender;
+
+import org.apache.spark.api.java.JavaRDD;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.LocalDate;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH;
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_SECOND_PARTITION_PATH;
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_THIRD_PARTITION_PATH;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests {@link TableSizeStats}. The tool reports through its log, so the
assertions read back the lines it
+ * logged for the table and for every partition it decided to include.
+ */
+public class TestTableSizeStats extends HoodieSparkClientTestBase {
+
+ private static final int RECORDS_PER_PARTITION = 4;
+ private static final String PARTITION_STATS_PREFIX = "Partition stats [name:
";
+ // the tool parses partition names as yyyy/M/d, so a partition from
yesterday is one --num-days can select
+ private static final String YESTERDAY_PARTITION_PATH =
+
LocalDate.now().minusDays(1).format(DateTimeFormatter.ofPattern("yyyy/M/d"));
+
+ private static Stream<Arguments> dateIntervalArgs() {
+ return Stream.of(
+ // everything on or after the start date: the 2016 partition and
yesterday's
+ Arguments.of("2016/1/1", null, 0L,
+ Arrays.asList(DEFAULT_FIRST_PARTITION_PATH,
YESTERDAY_PARTITION_PATH)),
+ // only the 2015 partitions are before the end date
+ Arguments.of(null, "2016/1/1", 0L,
+ Arrays.asList(DEFAULT_SECOND_PARTITION_PATH,
DEFAULT_THIRD_PARTITION_PATH)),
+ // half open interval [start, end): the start date is included, the
end date is not
+ Arguments.of("2015/3/16", "2015/3/17", 0L,
Collections.singletonList(DEFAULT_SECOND_PARTITION_PATH)),
+ // --num-days walks back from today: only yesterday's partition falls
inside a ten day window
+ Arguments.of(null, null, 10L,
Collections.singletonList(YESTERDAY_PARTITION_PATH)));
+ }
+
+ private TableSizeStats.Config statsConfig() {
+ TableSizeStats.Config cfg = new TableSizeStats.Config();
+ cfg.basePath = basePath;
+ cfg.parallelism = 2;
+ return cfg;
+ }
+
+ private void writeOneCommit(String... partitions) {
+ HoodieWriteConfig writeConfig = getConfigBuilder().build();
+ try (SparkRDDWriteClient client = getHoodieWriteClient(writeConfig)) {
+ String instantTime = WriteClientTestUtils.createNewInstantTime();
+ List<HoodieRecord> records = new ArrayList<>();
+ for (String partition : partitions) {
+ records.addAll(dataGen.generateInsertsForPartition(instantTime,
RECORDS_PER_PARTITION, partition));
+ }
+ WriteClientTestUtils.startCommitWithTime(client, instantTime);
+ JavaRDD<WriteStatus> writeStatuses =
client.insert(jsc.parallelize(records, 1), instantTime);
+ client.commit(instantTime, writeStatuses);
+ }
+ }
+
+ private void writeDefaultPartitions() {
+ writeOneCommit(DEFAULT_FIRST_PARTITION_PATH,
DEFAULT_SECOND_PARTITION_PATH, DEFAULT_THIRD_PARTITION_PATH);
+ }
+
+ private List<String> runAndCollectLogs(TableSizeStats.Config cfg) {
+ try (CapturingLogAppender logs =
CapturingLogAppender.attachTo(TableSizeStats.class)) {
+ new TableSizeStats(jsc, cfg).run();
+ return logs.messages();
+ }
+ }
+
+ private static Set<String> partitionStatHeaders(List<String> messages) {
+ return messages.stream().filter(m ->
m.startsWith(PARTITION_STATS_PREFIX)).collect(Collectors.toSet());
+ }
+
+ private static String lineAfter(List<String> messages, String header) {
+ int index = messages.indexOf(header);
+ assertTrue(index >= 0 && index + 1 < messages.size(), "missing log line ["
+ header + "] in " + messages);
+ return messages.get(index + 1);
+ }
+
+ @Test
+ public void testTableAndPartitionStatsCoverEveryPartition() {
+ writeDefaultPartitions();
+ TableSizeStats.Config cfg = statsConfig();
+ cfg.tableStats = true;
+ cfg.partitionStats = true;
+
+ List<String> messages = runAndCollectLogs(cfg);
+
+ Set<String> expectedHeaders = Stream.of(
+ DEFAULT_FIRST_PARTITION_PATH, DEFAULT_SECOND_PARTITION_PATH,
DEFAULT_THIRD_PARTITION_PATH)
+ .map(p -> PARTITION_STATS_PREFIX + p +
"]").collect(Collectors.toSet());
+ assertEquals(expectedHeaders, partitionStatHeaders(messages));
+ for (String header : expectedHeaders) {
+ assertEquals("Number of files: 1", lineAfter(messages, header));
+ }
+
+ String tableHeader = "Table stats [path: " + basePath + "]";
+ assertEquals("Number of files: 3", lineAfter(messages, tableHeader));
+ assertTrue(messages.stream().anyMatch(m -> m.matches("Total size:
\\d+\\.\\d{2} (B|KB|MB|GB|TB)")),
+ "expected a formatted total size in " + messages);
+ }
+
+ @Test
+ public void testTotalSizeOnlyWhenTableStatsAreOff() {
+ writeDefaultPartitions();
+ List<String> messages = runAndCollectLogs(statsConfig());
+
+ assertEquals(Collections.emptySet(), partitionStatHeaders(messages),
+ "partition stats must stay off unless asked for");
+ assertTrue(messages.stream().noneMatch(m -> m.startsWith("Table stats
[path: ")));
+ assertTrue(messages.stream().anyMatch(m -> m.matches("Total size:
\\d+\\.\\d{2} (B|KB|MB|GB|TB)")),
+ "expected a formatted total size in " + messages);
+ }
+
+ @ParameterizedTest
+ @MethodSource("dateIntervalArgs")
+ public void testOnlyPartitionsInsideTheDateIntervalAreCounted(String
startDate, String endDate, long numDays,
+ List<String>
expectedPartitions) {
+ writeOneCommit(DEFAULT_FIRST_PARTITION_PATH,
DEFAULT_SECOND_PARTITION_PATH, DEFAULT_THIRD_PARTITION_PATH,
+ YESTERDAY_PARTITION_PATH);
+ TableSizeStats.Config cfg = statsConfig();
+ cfg.partitionStats = true;
+ cfg.startDate = startDate;
+ cfg.endDate = endDate;
+ cfg.numDays = numDays;
+
+ List<String> messages = runAndCollectLogs(cfg);
+
+ Set<String> expectedHeaders = expectedPartitions.stream()
+ .map(p -> PARTITION_STATS_PREFIX + p + ", has date:
yes]").collect(Collectors.toSet());
+ assertEquals(expectedHeaders, partitionStatHeaders(messages));
+ }
+
+ @Test
+ public void testBasePathsAreReadFromThePropsFile() throws IOException {
+ writeDefaultPartitions();
+ Path propsFile = tempDir.resolve("base-paths.properties");
+ Files.write(propsFile, Collections.singletonList(basePath),
StandardCharsets.UTF_8);
+
+ TableSizeStats.Config cfg = statsConfig();
+ cfg.basePath = null;
+ cfg.propsFilePath = propsFile.toAbsolutePath().toString();
+ cfg.tableStats = true;
+
+ List<String> messages = runAndCollectLogs(cfg);
+ assertEquals("Number of files: 3", lineAfter(messages, "Table stats [path:
" + basePath + "]"));
+ }
+
+ /**
+ * --props-path is read twice: once by the constructor as a hoodie
properties file, and again by run() as the
+ * list of base paths. A file that is missing from the start never reaches
the second read.
+ */
+ @Test
+ public void testMissingPropsFileFailsInTheConstructor() {
+ TableSizeStats.Config cfg = statsConfig();
+ cfg.propsFilePath = tempDir.resolve("missing-" + UUID.randomUUID() +
".properties").toAbsolutePath().toString();
+ HoodieIOException thrown =
+ assertThrows(HoodieIOException.class, () -> new TableSizeStats(jsc,
cfg).run());
+ assertTrue(thrown.getMessage().contains("Properties file does not exist"),
thrown.getMessage());
+ }
+
+ @Test
+ public void testPropsFileRemovedAfterTheConstructorFailsTheRun() throws
IOException {
+ Path propsFile = tempDir.resolve("base-paths.properties");
+ Files.write(propsFile, Collections.singletonList(basePath),
StandardCharsets.UTF_8);
+ TableSizeStats.Config cfg = statsConfig();
+ cfg.propsFilePath = propsFile.toAbsolutePath().toString();
+
+ TableSizeStats stats = new TableSizeStats(jsc, cfg);
+ Files.delete(propsFile);
+
+ HoodieException thrown = assertThrows(HoodieException.class, stats::run);
+ assertTrue(thrown.getCause().getMessage().contains("Cannot read properties
from dfs from file"),
+ thrown.getCause().getMessage());
+ }
+
+ @Test
+ public void testMissingBasePathFails() {
+ TableSizeStats.Config cfg = statsConfig();
+ cfg.basePath = null;
+ HoodieException thrown = assertThrows(HoodieException.class, () -> new
TableSizeStats(jsc, cfg).run());
+ assertTrue(thrown.getCause().getMessage().contains("Base path needs to be
set."), thrown.getCause().toString());
+ }
+
+ @Test
+ public void testDateIntervalOnPartitionsWithoutDatesFails() {
+ writeOneCommit("country=us");
+ TableSizeStats.Config cfg = statsConfig();
+ cfg.numDays = 10;
+
+ HoodieException thrown = assertThrows(HoodieException.class, () -> new
TableSizeStats(jsc, cfg).run());
+ assertTrue(thrown.getCause().getMessage().contains("Cannot apply
--start-date, --end-date, or --num-days"),
+ thrown.getCause().getMessage());
+ assertTrue(thrown.getCause().getMessage().contains("country=us"),
thrown.getCause().getMessage());
+ }
+
+ @Test
+ public void testStartDateAfterEndDateFails() {
+ TableSizeStats.Config cfg = statsConfig();
+ cfg.startDate = "2017/1/1";
+ cfg.endDate = "2016/1/1";
+ HoodieException thrown = assertThrows(HoodieException.class, () -> new
TableSizeStats(jsc, cfg).run());
+ assertTrue(thrown.getCause().getMessage().contains("Starting date must be
before ending date"),
+ thrown.getCause().getMessage());
+ }
+
+ @Test
+ public void testNegativeNumDaysFails() {
+ TableSizeStats.Config cfg = statsConfig();
+ cfg.numDays = -1;
+ HoodieException thrown = assertThrows(HoodieException.class, () -> new
TableSizeStats(jsc, cfg).run());
+ assertTrue(thrown.getCause().getMessage().contains("--num-days must
specify a positive value"),
+ thrown.getCause().getMessage());
+ }
+
+ @Test
+ public void testUnparseableEndDateFails() {
+ TableSizeStats.Config cfg = statsConfig();
+ cfg.endDate = "yesterday";
+ HoodieException thrown = assertThrows(HoodieException.class, () -> new
TableSizeStats(jsc, cfg).run());
+ assertTrue(thrown.getCause().getMessage().contains("Unable to parse
--end-date"),
+ thrown.getCause().getMessage());
+ }
+
+ @Test
+ public void testConfigEqualsHashCodeAndToString() {
+ TableSizeStats.Config left = new TableSizeStats.Config();
+ left.basePath = "/tmp/table";
+ left.numDays = 3;
+ left.tableStats = true;
+ left.configs = new ArrayList<>(Collections.singletonList("k=v"));
+
+ TableSizeStats.Config right = new TableSizeStats.Config();
+ right.basePath = "/tmp/table";
+ right.numDays = 3;
+ right.tableStats = true;
+ right.configs = new ArrayList<>(Collections.singletonList("k=v"));
+
+ assertEquals(left, left);
+ assertEquals(left, right);
+ assertEquals(left.hashCode(), right.hashCode());
+ assertNotEquals(left, null);
+ assertNotEquals(left, "not a config");
+ // a Config straight out of JCommander has no base path yet
+ assertEquals(new TableSizeStats.Config(), new TableSizeStats.Config());
+ assertEquals(new TableSizeStats.Config().hashCode(), new
TableSizeStats.Config().hashCode());
+ // --help is not compared, so it must not be hashed either
+ TableSizeStats.Config askedForHelp = new TableSizeStats.Config();
+ askedForHelp.help = true;
+ assertEquals(new TableSizeStats.Config(), askedForHelp);
+ assertEquals(new TableSizeStats.Config().hashCode(),
askedForHelp.hashCode());
+
+ right.endDate = "2016/1/1";
+ assertNotEquals(left, right);
+ assertNotEquals(left.hashCode(), right.hashCode());
+
+ String printed = left.toString();
+ assertTrue(printed.contains("--base-path /tmp/table"));
+ assertTrue(printed.contains("--num-days 3"));
+ assertTrue(printed.contains("--enable-table-stats true"));
+ assertTrue(printed.contains("--hoodie-conf [k=v]"));
+ }
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/testutils/CapturingLogAppender.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/testutils/CapturingLogAppender.java
new file mode 100644
index 000000000000..feb36ba107bc
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/testutils/CapturingLogAppender.java
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.utilities.testutils;
+
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.Logger;
+import org.apache.logging.log4j.core.appender.AbstractAppender;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+
+/**
+ * Collects the messages one class logs, for tools whose only output is their
log.
+ *
+ * <p>{@code addAppender} gives the named logger a {@link
org.apache.logging.log4j.core.config.LoggerConfig} of
+ * its own, cloned from the nearest configured ancestor, and refreshes every
live logger from it. That refresh
+ * drops any level set beforehand, so the level is raised only after the
appender is in place. Events of child
+ * loggers still reach the appender through that config, hence the logger name
check in
+ * {@link #append(LogEvent)}.
+ */
+public class CapturingLogAppender extends AbstractAppender implements
AutoCloseable {
+
+ private final String loggerName;
+ private final Logger logger;
+ private final Level previousLevel;
+ private final List<String> messages = Collections.synchronizedList(new
ArrayList<>());
+
+ private CapturingLogAppender(String loggerName, Level level) {
+ super("Capture-" + loggerName + "-" + UUID.randomUUID(), null, null,
false, null);
+ this.loggerName = loggerName;
+ this.logger = (Logger) LogManager.getLogger(loggerName);
+ this.previousLevel = logger.getLevel();
+ start();
+ logger.addAppender(this);
+ logger.setLevel(level);
+ }
+
+ /**
+ * Starts capturing everything {@code loggerClass} logs at INFO or above,
until the returned appender is closed.
+ */
+ public static CapturingLogAppender attachTo(Class<?> loggerClass) {
+ return new CapturingLogAppender(loggerClass.getName(), Level.INFO);
+ }
+
+ @Override
+ public void append(LogEvent event) {
+ if (loggerName.equals(event.getLoggerName())) {
+ messages.add(event.getMessage().getFormattedMessage());
+ }
+ }
+
+ /**
+ * The formatted messages captured so far, in the order they were logged.
+ */
+ public List<String> messages() {
+ return new ArrayList<>(messages);
+ }
+
+ @Override
+ public void close() {
+ logger.removeAppender(this);
+ logger.setLevel(previousLevel);
+ stop();
+ }
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/testutils/ToolTestUtils.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/testutils/ToolTestUtils.java
new file mode 100644
index 000000000000..805f4ffa8dcb
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/testutils/ToolTestUtils.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.utilities.testutils;
+
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.view.FileSystemViewManager;
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
+
+/**
+ * Assertion helpers shared by the tests of the standalone spark tools.
+ */
+public final class ToolTestUtils {
+
+ private ToolTestUtils() {
+ }
+
+ /**
+ * The message of {@code throwable} and of every cause under it, one per
line. The tools wrap what they catch,
+ * so the detail a test cares about is usually a few causes down.
+ */
+ public static String stackMessages(Throwable throwable) {
+ StringBuilder sb = new StringBuilder();
+ for (Throwable t = throwable; t != null; t = t.getCause()) {
+ sb.append(t.getMessage()).append('\n');
+ }
+ return sb.toString();
+ }
+
+ /**
+ * How many base files the latest file slices of {@code partition} hold,
listed off the file system rather than
+ * the metadata table so that a partition dropped by a replacecommit reads
as empty.
+ */
+ public static long latestBaseFileCount(HoodieEngineContext context,
HoodieTableMetaClient metaClient,
+ String partition) {
+ HoodieTableMetaClient reloaded = HoodieTableMetaClient.reload(metaClient);
+ try (HoodieTableFileSystemView fsView =
FileSystemViewManager.createInMemoryFileSystemView(
+ context, reloaded,
HoodieMetadataConfig.newBuilder().enable(false).build())) {
+ return fsView.getLatestBaseFiles(partition).count();
+ }
+ }
+}
