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 08eb2d9e8429 fix: Parallelize cloud object existence checks in
S3EventsHoodieIncrSource (#18252)
08eb2d9e8429 is described below
commit 08eb2d9e84294f1eba38c08a9c8b919597c198a9
Author: Vinish Reddy <[email protected]>
AuthorDate: Mon Aug 17 16:16:55 2026 +0530
fix: Parallelize cloud object existence checks in S3EventsHoodieIncrSource
(#18252)
* [HUDI-XXXXX] Parallelize cloud object existence checks in
S3EventsHoodieIncrSource
Repartition by totalExecutorCores and run a per-task thread pool for
concurrent HEAD requests during file existence checks. Previously all
checks ran sequentially in a single partition due to upstream Window
coalescing, causing ~5hr latency for 176K+ files.
- Add EXISTS_CHECK_PARALLELISM config (default 32 threads per task)
- Repartition distinct() output by totalCores for cluster utilization
- Add thread pool in getCloudObjectMetadataPerPartition using
CompletableFuture.supplyAsync with explicit ExecutorService
- Extract processRow helper, fix per-file INFO log to DEBUG
- Add tests for parallel and sequential exists check paths
* [HUDI-XXXXX] fix: Use allOf().join() to prevent executor shutdown race in
parallel exists check
Replace sequential per-future join() with CompletableFuture.allOf().join()
to
wait for all futures concurrently inside the try block. This ensures
shutdownNow()
only fires after all work is done, preventing interrupted threads when the
caller
consumes results. Also adds proper CompletionException handling on the
fan-in join.
* Fix static import order in CloudObjectsSelectorCommon
* review(18252): fail-fast pool, defaultParallelism repartition, config
validation, tests
- getObjectMetadata: drop the SparkConf core arithmetic (yielded 1
partition or repartition(0)
in common deployments); repartition(jsc.defaultParallelism()) only when
the exists check is on
- getCloudObjectMetadataPerPartition: FutureUtils.allOf for fail-fast +
cancel, named daemon
threads via CustomizedThreadFactory, unwrap CompletionException so both
branches throw the
same HoodieException; keep the sequential branch eager
- EXISTS_CHECK_PARALLELISM: default 16, sinceVersion 1.3.0, drop dead
deltastreamer alias,
document the fs.s3a.connection.maximum interaction; validate >= 1
- tests: wired S3 getObjectMetadata test with the exists check on
(S3_FS_PREFIX + parallelism
from props, distinct, missing file dropped, 0 rejected); parameterized
{1,8} happy path and
failure path for the partition function
* review(18252): address round-2 review
- add EXISTS_CHECK_PARTITIONS (0 = spark default parallelism) and clamp the
count to >= 1;
extract existsCheckNumPartitions
- unwrapExistsCheckFailure rethrows any RuntimeException so both branches
surface the same
exception; size the pool by min(parallelism, rows) and short-circuit
empty partitions
- getUrlForFile copies the Configuration only when the exists check runs
- config doc: 500 connections on Hadoop 3.4+; javadoc contract for
existsCheckParallelism
- tests: URL-encoded event keys (a fixture with a space covers the decode),
a recording
LocalFileSystem proves the pooled vs caller thread per branch, failure
test also covers a
non-numeric size, explicit dedupe assert, existsCheckNumPartitions
asserted, @TempDir param
form like the rest of the file
---------
Co-authored-by: Lokesh Jain <[email protected]>
Co-authored-by: voon <[email protected]>
---
.../hudi/utilities/config/CloudSourceConfig.java | 21 +++
.../helpers/CloudObjectsSelectorCommon.java | 173 ++++++++++++++++-----
.../helpers/TestCloudObjectsSelectorCommon.java | 139 +++++++++++++++++
3 files changed, 295 insertions(+), 38 deletions(-)
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/CloudSourceConfig.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/CloudSourceConfig.java
index e2242e6a3c87..79dc309d1f65 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/CloudSourceConfig.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/CloudSourceConfig.java
@@ -217,4 +217,25 @@ public class CloudSourceConfig extends HoodieConfig {
+ "Note: the per-read mergeSchema option is honored by Spark's
native Parquet reader and by Spark's "
+ "native ORC reader (Spark 3.0+, default ORC impl since Spark 2.4).
On older runtimes the option is "
+ "silently ignored.");
+
+ public static final ConfigProperty<Integer> EXISTS_CHECK_PARALLELISM =
ConfigProperty
+ .key(STREAMER_CONFIG_PREFIX +
"source.cloud.data.check.file.exists.parallelism")
+ .defaultValue(16)
+ .markAdvanced()
+ .sinceVersion("1.3.0")
+ .withDocumentation("Number of threads per Spark task used to check cloud
object existence concurrently when "
+ + ENABLE_EXISTS_CHECK.key() + " is enabled. Must be >= 1; 1 checks
sequentially. All tasks on an executor "
+ + "share one cached FileSystem client, so keep executor cores x this
value within the client's connection "
+ + "pool (fs.s3a.connection.maximum, 96 by default on Hadoop 3.3, 500
on 3.4+) to avoid connection pool timeouts.");
+
+ public static final ConfigProperty<Integer> EXISTS_CHECK_PARTITIONS =
ConfigProperty
+ .key(STREAMER_CONFIG_PREFIX +
"source.cloud.data.check.file.exists.partitions")
+ .defaultValue(0)
+ .markAdvanced()
+ .sinceVersion("1.3.0")
+ .withDocumentation("Number of Spark partitions the cloud object
existence checks are spread over when "
+ + ENABLE_EXISTS_CHECK.key() + " is enabled; each partition checks
with " + EXISTS_CHECK_PARALLELISM.key()
+ + " threads. The default 0 sizes to the cluster (spark default
parallelism, i.e. the registered executor "
+ + "cores). Set explicitly to cap the total concurrency (partitions x
threads) against the storage client's "
+ + "connection pool and request rate limits.");
}
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/CloudObjectsSelectorCommon.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/CloudObjectsSelectorCommon.java
index c5a9a6e7a923..be048007f534 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/CloudObjectsSelectorCommon.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/CloudObjectsSelectorCommon.java
@@ -24,8 +24,11 @@ import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.schema.HoodieSchemaField;
import org.apache.hudi.common.schema.HoodieSchemaType;
+import org.apache.hudi.common.util.CustomizedThreadFactory;
+import org.apache.hudi.common.util.FutureUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
@@ -64,15 +67,22 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.function.Function;
import java.util.stream.Collectors;
import static org.apache.hudi.common.util.CollectionUtils.isNullOrEmpty;
import static org.apache.hudi.common.util.ConfigUtils.containsConfigProperty;
import static org.apache.hudi.common.util.ConfigUtils.getBooleanWithAltKeys;
+import static org.apache.hudi.common.util.ConfigUtils.getIntWithAltKeys;
import static org.apache.hudi.common.util.ConfigUtils.getStringWithAltKeys;
import static
org.apache.hudi.utilities.config.CloudSourceConfig.CLOUD_DATAFILE_EXTENSION;
import static
org.apache.hudi.utilities.config.CloudSourceConfig.CLOUD_INCREMENTAL_MERGE_SCHEMA;
+import static
org.apache.hudi.utilities.config.CloudSourceConfig.EXISTS_CHECK_PARALLELISM;
+import static
org.apache.hudi.utilities.config.CloudSourceConfig.EXISTS_CHECK_PARTITIONS;
import static
org.apache.hudi.utilities.config.CloudSourceConfig.IGNORE_RELATIVE_PATH_PREFIX;
import static
org.apache.hudi.utilities.config.CloudSourceConfig.IGNORE_RELATIVE_PATH_SUBSTR;
import static
org.apache.hudi.utilities.config.CloudSourceConfig.PATH_BASED_PARTITION_FIELDS;
@@ -117,35 +127,94 @@ public class CloudObjectsSelectorCommon {
* @param storageUrlSchemePrefix Eg: s3:// or gs://. The
storage-provider-specific prefix to use within the URL.
* @param storageConf storage configuration.
* @param checkIfExists check if each file exists, before adding
it to the returned list
- * @return
+ * @param existsCheckParallelism number of threads per task for the
existence checks (getObjectMetadata validates it is >= 1); 1 checks sequentially
*/
public static MapPartitionsFunction<Row, CloudObjectMetadata>
getCloudObjectMetadataPerPartition(
- String storageUrlSchemePrefix, StorageConfiguration<Configuration>
storageConf, boolean checkIfExists) {
+ String storageUrlSchemePrefix, StorageConfiguration<Configuration>
storageConf,
+ boolean checkIfExists, int existsCheckParallelism) {
return rows -> {
- List<CloudObjectMetadata> cloudObjectMetadataPerPartition = new
ArrayList<>();
- rows.forEachRemaining(row -> {
- Option<String> filePathUrl = getUrlForFile(row,
storageUrlSchemePrefix, storageConf, checkIfExists);
- filePathUrl.ifPresent(url -> {
- log.info("Adding file: {}", url);
- long size;
- Object obj = row.get(2);
- if (obj instanceof String) {
- size = Long.parseLong((String) obj);
- } else if (obj instanceof Integer) {
- size = ((Integer) obj).longValue();
- } else if (obj instanceof Long) {
- size = (long) obj;
- } else {
- throw new HoodieIOException("unexpected object size's type in
Cloud storage events: " + obj.getClass());
- }
- cloudObjectMetadataPerPartition.add(new CloudObjectMetadata(url,
size));
- });
- });
+ if (!checkIfExists || existsCheckParallelism <= 1) {
+ List<CloudObjectMetadata> cloudObjectMetadataPerPartition = new
ArrayList<>();
+ rows.forEachRemaining(row ->
+ processRow(row, storageUrlSchemePrefix, storageConf,
checkIfExists).ifPresent(cloudObjectMetadataPerPartition::add));
+ return cloudObjectMetadataPerPartition.iterator();
+ }
- return cloudObjectMetadataPerPartition.iterator();
+ List<Row> rowList = new ArrayList<>();
+ rows.forEachRemaining(rowList::add);
+ if (rowList.isEmpty()) {
+ return Collections.emptyIterator();
+ }
+ ExecutorService executor =
Executors.newFixedThreadPool(Math.min(existsCheckParallelism, rowList.size()),
+ new CustomizedThreadFactory("cloud-exists-check", true));
+ try {
+ List<CompletableFuture<Option<CloudObjectMetadata>>> futures =
rowList.stream()
+ .map(row -> CompletableFuture.supplyAsync(
+ () -> processRow(row, storageUrlSchemePrefix, storageConf,
true), executor))
+ .collect(Collectors.toList());
+ List<Option<CloudObjectMetadata>> results;
+ try {
+ // fails fast: the first failed check completes the returned future
exceptionally and cancels the rest,
+ // and every future is complete before this returns, so the pool can
be shut down right after
+ results = FutureUtils.allOf(futures).join();
+ } catch (CompletionException e) {
+ throw unwrapExistsCheckFailure(e);
+ }
+ return results.stream()
+ .filter(Option::isPresent)
+ .map(Option::get)
+ .collect(Collectors.toList())
+ .iterator();
+ } finally {
+ // on the failure path this also interrupts the in-flight checks that
FutureUtils.allOf only cancelled
+ executor.shutdownNow();
+ }
};
}
+ /**
+ * {@link FutureUtils#allOf} wraps the failure of a check in one or more
{@link CompletionException}s; rethrow the
+ * original exception so callers see the same failure the sequential path
throws, whatever the parallelism.
+ */
+ private static RuntimeException unwrapExistsCheckFailure(CompletionException
e) {
+ Throwable cause = e;
+ while (cause instanceof CompletionException && cause.getCause() != null) {
+ cause = cause.getCause();
+ }
+ if (cause instanceof Error) {
+ throw (Error) cause;
+ }
+ return cause instanceof RuntimeException
+ ? (RuntimeException) cause
+ : new HoodieException("Failed during parallel cloud object existence
check", cause);
+ }
+
+ /**
+ * Process a single row to build a {@link CloudObjectMetadata}. Optionally
checks if the file exists.
+ */
+ private static Option<CloudObjectMetadata> processRow(Row row, String
storageUrlSchemePrefix,
+
StorageConfiguration<Configuration> storageConf,
+ boolean checkIfExists)
{
+ Option<String> filePathUrl = getUrlForFile(row, storageUrlSchemePrefix,
storageConf, checkIfExists);
+ if (!filePathUrl.isPresent()) {
+ return Option.empty();
+ }
+ String url = filePathUrl.get();
+ log.debug("Adding file: {}", url);
+ long size;
+ Object obj = row.get(2);
+ if (obj instanceof String) {
+ size = Long.parseLong((String) obj);
+ } else if (obj instanceof Integer) {
+ size = ((Integer) obj).longValue();
+ } else if (obj instanceof Long) {
+ size = (long) obj;
+ } else {
+ throw new HoodieIOException("unexpected object size's type in Cloud
storage events: " + obj.getClass());
+ }
+ return Option.of(new CloudObjectMetadata(url, size));
+ }
+
/**
* Construct a full qualified URL string to a cloud file from a given Row.
Optionally check if the file exists.
* Here Row is assumed to have the schema [bucket_name,
filepath_relative_to_bucket].
@@ -157,8 +226,6 @@ public class CloudObjectsSelectorCommon {
private static Option<String> getUrlForFile(Row row, String
storageUrlSchemePrefix,
StorageConfiguration<Configuration> storageConf,
boolean checkIfExists) {
- final Configuration configuration = storageConf.unwrapCopy();
-
String bucket = row.getString(0);
String filePath = storageUrlSchemePrefix + bucket + StoragePath.SEPARATOR
+ row.getString(1);
@@ -167,7 +234,7 @@ public class CloudObjectsSelectorCommon {
if (!checkIfExists) {
return Option.of(filePathUrl);
}
- boolean exists = checkIfFileExists(storageUrlSchemePrefix, bucket,
filePathUrl, configuration);
+ boolean exists = checkIfFileExists(storageUrlSchemePrefix, bucket,
filePathUrl, storageConf.unwrapCopy());
return exists ? Option.of(filePathUrl) : Option.empty();
} catch (Exception exception) {
log.error("Failed to generate path to cloud file {}", filePath,
exception);
@@ -256,22 +323,52 @@ public class CloudObjectsSelectorCommon {
TypedProperties props
) {
StorageConfiguration<Configuration> storageConf =
HadoopFSUtils.getStorageConfWithCopy(jsc.hadoopConfiguration());
+ int existsCheckParallelism = getIntWithAltKeys(props,
EXISTS_CHECK_PARALLELISM);
+ ValidationUtils.checkArgument(existsCheckParallelism >= 1,
+ EXISTS_CHECK_PARALLELISM.key() + " must be >= 1, got: " +
existsCheckParallelism);
+
+ String prefix;
+ String bucketCol;
+ String keyCol;
+ String sizeCol;
if (type == Type.GCS) {
- return cloudObjectMetadataDF
- .select("bucket", "name", "size")
- .distinct()
- .mapPartitions(getCloudObjectMetadataPerPartition(GCS_PREFIX,
storageConf, checkIfExists), Encoders.kryo(CloudObjectMetadata.class))
- .collectAsList();
+ prefix = GCS_PREFIX;
+ bucketCol = "bucket";
+ keyCol = GCS_OBJECT_KEY;
+ sizeCol = GCS_OBJECT_SIZE;
} else if (type == Type.S3) {
String s3FS = getStringWithAltKeys(props, S3_FS_PREFIX,
true).toLowerCase();
- String s3Prefix = s3FS + "://";
- return cloudObjectMetadataDF
- .select(CloudObjectsSelectorCommon.S3_BUCKET_NAME,
CloudObjectsSelectorCommon.S3_OBJECT_KEY,
CloudObjectsSelectorCommon.S3_OBJECT_SIZE)
- .distinct()
- .mapPartitions(getCloudObjectMetadataPerPartition(s3Prefix,
storageConf, checkIfExists), Encoders.kryo(CloudObjectMetadata.class))
- .collectAsList();
- }
- throw new UnsupportedOperationException("Invalid cloud type " + type);
+ prefix = s3FS + "://";
+ bucketCol = S3_BUCKET_NAME;
+ keyCol = S3_OBJECT_KEY;
+ sizeCol = S3_OBJECT_SIZE;
+ } else {
+ throw new UnsupportedOperationException("Invalid cloud type " + type);
+ }
+
+ Dataset<Row> distinctObjects = cloudObjectMetadataDF.select(bucketCol,
keyCol, sizeCol).distinct();
+ if (checkIfExists) {
+ // The upstream Window.orderBy() in IncrSourceHelper collapses the
dataset to one partition and AQE keeps the
+ // distinct() output there, which would serialize every existence check
on one task. Spread the checks over
+ // the cluster: repartition(n) is not coalesced by AQE, and
defaultParallelism tracks the registered cores.
+ int numPartitions = existsCheckNumPartitions(props, jsc);
+ log.info("Checking cloud object existence over {} partitions with {}
threads per task", numPartitions, existsCheckParallelism);
+ distinctObjects = distinctObjects.repartition(numPartitions);
+ }
+ return distinctObjects
+ .mapPartitions(
+ getCloudObjectMetadataPerPartition(prefix, storageConf,
checkIfExists, existsCheckParallelism),
+ Encoders.kryo(CloudObjectMetadata.class))
+ .collectAsList();
+ }
+
+ /**
+ * Number of partitions the existence checks are spread over: {@link
CloudSourceConfig#EXISTS_CHECK_PARTITIONS}
+ * when set to a positive value, otherwise the Spark default parallelism.
Never below 1 (repartition rejects 0).
+ */
+ static int existsCheckNumPartitions(TypedProperties props, JavaSparkContext
jsc) {
+ int configured = getIntWithAltKeys(props, EXISTS_CHECK_PARTITIONS);
+ return Math.max(1, configured > 0 ? configured : jsc.defaultParallelism());
}
public Option<Dataset<Row>> loadAsDataset(SparkSession spark,
List<CloudObjectMetadata> cloudObjectMetadata,
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestCloudObjectsSelectorCommon.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestCloudObjectsSelectorCommon.java
index f7cb663fa320..3085d3db08e6 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestCloudObjectsSelectorCommon.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestCloudObjectsSelectorCommon.java
@@ -23,12 +23,20 @@ import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.schema.HoodieSchemaField;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
import org.apache.hudi.testutils.HoodieSparkClientTestHarness;
import org.apache.hudi.utilities.config.CloudSourceConfig;
+import org.apache.hudi.utilities.config.S3EventsHoodieIncrSourceConfig;
import org.apache.hudi.utilities.schema.FilebasedSchemaProvider;
import org.apache.hudi.utilities.schema.RowBasedSchemaProvider;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.LocalFileSystem;
+import org.apache.spark.api.java.function.MapPartitionsFunction;
import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.types.DataTypes;
@@ -40,6 +48,7 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
import java.io.File;
@@ -50,8 +59,10 @@ import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
@@ -329,6 +340,104 @@ public class TestCloudObjectsSelectorCommon extends
HoodieSparkClientTestHarness
Assertions.assertEquals(Collections.singletonList(RowFactory.create(expectedPath,
"some data")), result.get().collectAsList());
}
+ @Test
+ void s3ObjectMetadataDedupesAndFiltersMissingFilesWithExistsCheck(@TempDir
Path tempDir) throws IOException {
+ // a space in the name: the event key is percent-encoded and getUrlForFile
must decode it
+ Path existingFile1 = Files.write(tempDir.resolve("we ird.json"),
Collections.singletonList("{}"));
+ Path existingFile2 = Files.write(tempDir.resolve("file2.json"),
Collections.singletonList("{}"));
+ Path missingFile = tempDir.resolve("missing.json");
+ // bucket "" + key = absolute path without its leading separator, so
prefix + bucket + "/" + key is a file:// URL
+ List<String> jsonRecords = Arrays.asList(
+ s3EventJson(existingFile1, 100),
+ s3EventJson(existingFile2, 200),
+ s3EventJson(missingFile, 300),
+ // duplicate event, dropped by distinct()
+ s3EventJson(existingFile1, 100));
+ Dataset<Row> cloudObjectMetadataDF =
sparkSession.read().json(sparkSession.createDataset(jsonRecords,
Encoders.STRING()));
+ TypedProperties props = new TypedProperties();
+ props.put(S3EventsHoodieIncrSourceConfig.S3_FS_PREFIX.key(), "file");
+ props.put(CloudSourceConfig.EXISTS_CHECK_PARALLELISM.key(), "0");
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
CloudObjectsSelectorCommon.getObjectMetadata(
+ CloudObjectsSelectorCommon.Type.S3, jsc, cloudObjectMetadataDF, true,
props));
+
+ props.put(CloudSourceConfig.EXISTS_CHECK_PARALLELISM.key(), "4");
+ Assertions.assertEquals(jsc.defaultParallelism(),
CloudObjectsSelectorCommon.existsCheckNumPartitions(props, jsc));
+ props.put(CloudSourceConfig.EXISTS_CHECK_PARTITIONS.key(), "2");
+ Assertions.assertEquals(2,
CloudObjectsSelectorCommon.existsCheckNumPartitions(props, jsc));
+ List<CloudObjectMetadata> result =
CloudObjectsSelectorCommon.getObjectMetadata(
+ CloudObjectsSelectorCommon.Type.S3, jsc, cloudObjectMetadataDF, true,
props);
+
+ Assertions.assertEquals(2, result.size(), "distinct() should have
collapsed the duplicate event");
+ Map<String, Long> pathToSize = result.stream()
+ .collect(Collectors.toMap(CloudObjectMetadata::getPath,
CloudObjectMetadata::getSize));
+ Map<String, Long> expected = new HashMap<>();
+ expected.put("file://" + existingFile1.toAbsolutePath(), 100L);
+ expected.put("file://" + existingFile2.toAbsolutePath(), 200L);
+ Assertions.assertEquals(expected, pathToSize);
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {1, 8})
+ void existsCheckDropsMissingFiles(int parallelism, @TempDir Path tempDir)
throws Exception {
+ // a space in the name: the event key is percent-encoded and getUrlForFile
must decode it
+ Path existingFile1 = Files.write(tempDir.resolve("we ird.json"),
Collections.singletonList("{}"));
+ Path existingFile2 = Files.write(tempDir.resolve("file2.json"),
Collections.singletonList("{}"));
+ Path missingFile = tempDir.resolve("missing.json");
+ List<Row> rows = Arrays.asList(
+ cloudEventRow(existingFile1, 100L),
+ cloudEventRow(missingFile, 300L),
+ cloudEventRow(existingFile2, 200L));
+
+ Configuration conf = storageConf.unwrapCopy();
+ conf.setClass("fs.file.impl", ThreadRecordingLocalFileSystem.class,
FileSystem.class);
+ conf.setBoolean("fs.file.impl.disable.cache", true);
+ ThreadRecordingLocalFileSystem.EXISTS_CALL_THREADS.clear();
+
+ List<CloudObjectMetadata> result = new ArrayList<>();
+ CloudObjectsSelectorCommon.getCloudObjectMetadataPerPartition(
+ "file://", new HadoopStorageConfiguration(conf), true, parallelism)
+ .call(rows.iterator()).forEachRemaining(result::add);
+
+ // both branches must produce the same objects, in input order
+ Assertions.assertEquals(
+ Arrays.asList("file://" + existingFile1.toAbsolutePath(), "file://" +
existingFile2.toAbsolutePath()),
+
result.stream().map(CloudObjectMetadata::getPath).collect(Collectors.toList()));
+ Assertions.assertEquals(Arrays.asList(100L, 200L),
+
result.stream().map(CloudObjectMetadata::getSize).collect(Collectors.toList()));
+
+ // one exists() per row; on pool threads for the parallel branch, on the
caller thread for the sequential one
+ List<String> threads = new
ArrayList<>(ThreadRecordingLocalFileSystem.EXISTS_CALL_THREADS);
+ Assertions.assertEquals(3, threads.size(), threads.toString());
+ if (parallelism > 1) {
+ Assertions.assertTrue(threads.stream().allMatch(t ->
t.startsWith("cloud-exists-check-")), threads.toString());
+ } else {
+
Assertions.assertEquals(Collections.singleton(Thread.currentThread().getName()),
new HashSet<>(threads));
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({"1, escape", "8, escape", "1, size", "8, size"})
+ void existsCheckSurfacesRowFailure(int parallelism, String failure, @TempDir
Path tempDir) throws Exception {
+ // both branches must throw the same exception instead of dropping the
file: a key with a malformed percent
+ // escape fails URL decoding inside getUrlForFile (HoodieException), a
non-numeric size string fails
+ // Long.parseLong in processRow (raw NumberFormatException)
+ Path existingFile = Files.write(tempDir.resolve("file1.json"),
Collections.singletonList("{}"));
+ Row badRow = failure.equals("escape")
+ ? RowFactory.create("", "path/bad%zz.json", 200L)
+ : RowFactory.create("", localFileKey(existingFile), "not-a-number");
+ List<Row> rows = Arrays.asList(cloudEventRow(existingFile, 100L), badRow);
+ MapPartitionsFunction<Row, CloudObjectMetadata> fn =
+
CloudObjectsSelectorCommon.getCloudObjectMetadataPerPartition("file://",
storageConf, true, parallelism);
+
+ if (failure.equals("escape")) {
+ HoodieException e = Assertions.assertThrows(HoodieException.class, () ->
fn.call(rows.iterator()));
+ Assertions.assertTrue(e.getMessage().contains("path/bad%zz.json"),
e.getMessage());
+ Assertions.assertInstanceOf(IllegalArgumentException.class,
e.getCause());
+ } else {
+ Assertions.assertThrows(NumberFormatException.class, () ->
fn.call(rows.iterator()));
+ }
+ }
+
/**
* Asserts that a Dataset contains expected rows; when the source path
column is enabled it is expected
* to be appended last, nullable, and to hold the file URI of the row's
source file.
@@ -365,4 +474,34 @@ public class TestCloudObjectsSelectorCommon extends
HoodieSparkClientTestHarness
private static void setIncludeSourcePathField(TypedProperties properties,
boolean include) {
properties.put(CloudSourceConfig.INCLUDE_SOURCE_PATH_FIELD.key(),
String.valueOf(include));
}
+
+ /** Row with the [bucket, key, size] shape
getCloudObjectMetadataPerPartition expects; see s3EventJson for the key. */
+ private static Row cloudEventRow(Path file, long size) {
+ return RowFactory.create("", localFileKey(file), size);
+ }
+
+ /** S3 event notification whose bucket is empty and whose key is the
URL-encoded absolute path without the leading
+ * separator (the shape S3 event notifications carry), so that prefix
"file://" + bucket + "/" + key decodes to the
+ * local file. */
+ private static String s3EventJson(Path file, long size) {
+ return "{\"s3\":{\"bucket\":{\"name\":\"\"},\"object\":{\"key\":\"" +
localFileKey(file) + "\",\"size\":" + size + "}}}";
+ }
+
+ private static String localFileKey(Path file) {
+ return file.toUri().getRawPath().substring(1);
+ }
+
+ /**
+ * LocalFileSystem that records the thread each exists() call ran on, so the
tests can tell the pooled branch
+ * from the sequential one. Registered as fs.file.impl with the FileSystem
cache disabled.
+ */
+ public static class ThreadRecordingLocalFileSystem extends LocalFileSystem {
+ static final List<String> EXISTS_CALL_THREADS =
Collections.synchronizedList(new ArrayList<>());
+
+ @Override
+ public boolean exists(org.apache.hadoop.fs.Path f) throws IOException {
+ EXISTS_CALL_THREADS.add(Thread.currentThread().getName());
+ return super.exists(f);
+ }
+ }
}
