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 3e08b72fe7a5 test(utilities): Tail sweep: low-coverage small classes
for hudi-utilities (#19416)
3e08b72fe7a5 is described below
commit 3e08b72fe7a5b596f9ce463db42234f5330bdf5f
Author: voonhous <[email protected]>
AuthorDate: Fri Jul 31 19:06:54 2026 +0800
test(utilities): Tail sweep: low-coverage small classes for hudi-utilities
(#19416)
* test(utilities): cover exception classes and small helper POJOs
Add grouped unit tests for the utilities exception constructors, the
deprecated cloud-ingestion config shims, DocumentParserType resolution,
QueryInfo and CheckpointWithPredicates accessors, and the two reflective
strategy loader utils in the streamer package.
* test(utilities): extend unit tests for schema, metrics, and validator
classes
Cover the SchemaPostProcessor bridge overloads and deprecated config
constants, schema provider failure paths and refresh, the validation
context timeline accessors, PubsubQueueClient request builders, the
deprecated delta streamer metrics alias, small source POJOs, and the
abstract error table writer and ingestion metrics constructors.
* test(utilities): cover small Spark-backed sources and callback branches
Add a standalone ORCDFSSource test, cover the SqlSource meta-column
drop, S3EventsSource commit and close handling, the Maxwell post
processor ordering-field branches driven without a broker, Kafka source
constructor validation, the Kafka commit callback partition routing,
and the AWS DMS transformer Op-present branch.
* test(utilities): address self-review findings
Drop the ORC read test that trips the HUDI-8081 hive-exec vs
hive-storage-api classpath collision, keeping the source config
assertions in a dedicated config test. Make the deprecated-constant
assertions falsifiable by pinning literal keys, make the schema
refresh, retry backoff, and S3 inference tests discriminating, and
assert exact messages in the Kafka source constructor guards. Remove
the unfalsifiable constructor-mock tests, replace reference identity
with content equality in the delete-support test, move the source and
row-schema-provider tests to their conventional homes, cover both
callback partition routing branches through a single embedded broker,
and delete a superseded no-op regex block plus duplicated maxwell
fixtures.
* test(utilities): address review comments
- TestKafkaCallbackProvider: drop assertDoesNotThrow around call() (it
swallows send failures), assert each consumed message body carries the
commit time that produced it, and name the poll timeout constant
- TestRowBasedSchemaProvider: drop the props-ctor test; nothing resolves
that constructor reflectively and it asserted nothing meaningful
- TestFilebasedSchemaProvider: cover the no-target-schema-file branch of
getTargetSchema() before and after refresh()
- TestS3EventsSource: add glacierEventData (absent from s3-metadata.avsc)
to the SQS fixture so the no-provider test pins schema inference
---
.../hudi/utilities/TestSchemaPostProcessor.java | 92 +++++++++++
.../callback/TestKafkaCallbackProvider.java | 80 +++++++++-
.../utilities/config/TestORCDFSSourceConfig.java | 45 ++++++
.../exception/TestUtilitiesExceptions.java | 169 +++++++++++++++++++++
.../TestAWSDatabaseMigrationServiceSource.java | 17 +++
.../schema/TestFilebasedSchemaProvider.java | 85 +++++++++++
.../schema/TestProtoClassBasedSchemaProvider.java | 33 ++++
.../schema/TestRowBasedSchemaProvider.java | 44 ++++++
.../schema/TestSchemaRegistryProvider.java | 8 +
.../utilities/sources/TestAvroKafkaSource.java | 13 ++
.../sources/TestJsonKafkaSourcePostProcessor.java | 111 +++++++++++---
.../utilities/sources/TestProtoKafkaSource.java | 32 ++++
.../hudi/utilities/sources/TestS3EventsSource.java | 63 +++++++-
.../apache/hudi/utilities/sources/TestSource.java | 25 ++-
.../hudi/utilities/sources/TestSqlSource.java | 32 ++++
.../TestDeprecatedCloudIngestionConfigs.java | 79 ++++++++++
.../sources/helpers/TestIncrSourceHelper.java | 14 ++
.../utilities/sources/helpers/TestQueryInfo.java | 109 +++++++++++++
.../sources/helpers/gcs/TestPubsubQueueClient.java | 53 +++++++
.../unstructured/TestDocumentParserType.java | 81 ++++++++++
.../TestConfigurationHotUpdateStrategyUtils.java | 89 +++++++++++
.../streamer/TestHoodieStreamerMetrics.java | 26 ++++
.../streamer/TestTerminationStrategyUtils.java | 62 ++++++++
.../validator/TestSparkValidationContext.java | 42 +++++
.../utilities/testutils/CloudObjectTestUtils.java | 5 +-
.../TestOpenAICompatibleEmbeddingProvider.java | 27 +++-
26 files changed, 1402 insertions(+), 34 deletions(-)
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestSchemaPostProcessor.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestSchemaPostProcessor.java
index d25878445a0d..d6a14fc30e42 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestSchemaPostProcessor.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestSchemaPostProcessor.java
@@ -32,6 +32,8 @@ import
org.apache.hudi.utilities.schema.postprocessor.add.AddPrimitiveColumnSche
import org.apache.hudi.utilities.testutils.UtilitiesTestBase;
import org.apache.hudi.utilities.transform.FlatteningTransformer;
+import org.apache.avro.Schema;
+import org.apache.spark.api.java.JavaSparkContext;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -149,6 +151,46 @@ public class TestSchemaPostProcessor extends
UtilitiesTestBase {
assertNotNull(targetSchema.getField("_row_key").orElse(null));
}
+ @Test
+ public void testDeleteSupportSkipsSchemaAlreadyCarryingTheDeleteMarker() {
+ DeleteSupportSchemaPostProcessor processor = new
DeleteSupportSchemaPostProcessor(properties, null);
+ // deprecated avro overload
+ Schema avroSchema =
processor.processSchema(HoodieSchema.parse(ORIGINAL_SCHEMA).toAvroSchema());
+ HoodieSchema withDeleteMarker = HoodieSchema.fromAvroSchema(avroSchema);
+
assertNotNull(withDeleteMarker.getField("_hoodie_is_deleted").orElse(null));
+
+ // reprocessing yields the same schema instead of adding the column twice
+ HoodieSchema reprocessed = processor.processSchema(withDeleteMarker);
+ assertEquals(withDeleteMarker, reprocessed);
+ assertEquals(1, reprocessed.getFields().stream()
+ .filter(field -> "_hoodie_is_deleted".equals(field.name())).count());
+ }
+
+ @Test
+ public void testProcessSchemaBridgesToDeprecatedOverload() {
+ SchemaPostProcessor processor = new
RenamingSchemaPostProcessor(properties, null);
+ HoodieSchema targetSchema =
processor.processSchema(HoodieSchema.parse(ORIGINAL_SCHEMA));
+
+ assertEquals("renamedRec", targetSchema.getName());
+ assertEquals(HoodieSchema.parse(ORIGINAL_SCHEMA).getFields().size(),
targetSchema.getFields().size());
+ }
+
+ @Test
+ public void testProcessSchemaThrowsWhenNeitherOverloadIsImplemented() {
+ SchemaPostProcessor processor = new
UnimplementedSchemaPostProcessor(properties, null);
+ HoodieSchema schema = HoodieSchema.parse(ORIGINAL_SCHEMA);
+
+ Assertions.assertThrows(UnsupportedOperationException.class, () ->
processor.processSchema(schema));
+ }
+
+ @Test
+ public void testDeprecatedConfigConstants() {
+ assertEquals("hoodie.streamer.schemaprovider.schema_post_processor",
+ SchemaPostProcessor.Config.SCHEMA_POST_PROCESSOR_PROP);
+
assertEquals("hoodie.streamer.schemaprovider.schema_post_processor.delete.columns",
+
DropColumnSchemaPostProcessor.Config.DELETE_COLUMN_POST_PROCESSOR_COLUMN_PROP);
+ }
+
@Test
public void testDeleteColumnThrows() {
// remove all columns from source schema
@@ -159,6 +201,18 @@ public class TestSchemaPostProcessor extends
UtilitiesTestBase {
Assertions.assertThrows(HoodieSchemaPostProcessException.class, () ->
processor.processSchema(schema));
}
+ @Test
+ public void testDeleteColumnWithEmptyParam() {
+ // configured but empty: nothing is deleted and the source schema is
returned unchanged
+
properties.put(SchemaProviderPostProcessorConfig.DELETE_COLUMN_POST_PROCESSOR_COLUMN.key(),
"");
+ DropColumnSchemaPostProcessor processor = new
DropColumnSchemaPostProcessor(properties, null);
+ HoodieSchema schema = HoodieSchema.parse(ORIGINAL_SCHEMA);
+ HoodieSchema targetSchema = processor.processSchema(schema);
+
+ assertEquals(schema.getFields().size(), targetSchema.getFields().size());
+ assertNotNull(targetSchema.getField("rider").orElse(null));
+ }
+
@ParameterizedTest
@MethodSource("configParams")
public void testAddPrimitiveTypeColumn(String type) {
@@ -183,4 +237,42 @@ public class TestSchemaPostProcessor extends
UtilitiesTestBase {
newColumn = targetSchema.getField("primitive_column").get();
assertEquals(type, newColumn.schema().getType().name().toLowerCase());
}
+
+ @Test
+ public void testAddPrimitiveTypeColumnWithDeprecatedOverload() {
+
properties.put(SchemaProviderPostProcessorConfig.SCHEMA_POST_PROCESSOR_ADD_COLUMN_NAME_PROP.key(),
"primitive_column");
+
properties.put(SchemaProviderPostProcessorConfig.SCHEMA_POST_PROCESSOR_ADD_COLUMN_TYPE_PROP.key(),
"string");
+
+ AddPrimitiveColumnSchemaPostProcessor processor = new
AddPrimitiveColumnSchemaPostProcessor(properties, null);
+ Schema targetSchema =
processor.processSchema(HoodieSchema.parse(ORIGINAL_SCHEMA).toAvroSchema());
+
+ assertNotNull(targetSchema.getField("primitive_column"));
+ assertNotNull(targetSchema.getField("_row_key"));
+ }
+
+ /**
+ * Implements only the deprecated avro overload, so that calls to
+ * {@link SchemaPostProcessor#processSchema(HoodieSchema)} go through the
base class bridge.
+ */
+ private static class RenamingSchemaPostProcessor extends SchemaPostProcessor
{
+
+ RenamingSchemaPostProcessor(TypedProperties props, JavaSparkContext jssc) {
+ super(props, jssc);
+ }
+
+ @Override
+ public Schema processSchema(Schema schema) {
+ return new
Schema.Parser().parse(schema.toString().replace("tripUberRec", "renamedRec"));
+ }
+ }
+
+ /**
+ * Implements neither overload, so the base class rejects the call.
+ */
+ private static class UnimplementedSchemaPostProcessor extends
SchemaPostProcessor {
+
+ UnimplementedSchemaPostProcessor(TypedProperties props, JavaSparkContext
jssc) {
+ super(props, jssc);
+ }
+ }
}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/callback/TestKafkaCallbackProvider.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/callback/TestKafkaCallbackProvider.java
index 92421a03eaf4..450e5fd04855 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/callback/TestKafkaCallbackProvider.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/callback/TestKafkaCallbackProvider.java
@@ -30,6 +30,11 @@ import
org.apache.hudi.utilities.callback.kafka.HoodieWriteCommitKafkaCallbackCo
import org.apache.hudi.utilities.testutils.KafkaTestUtils;
import org.apache.hudi.utilities.testutils.UtilitiesTestBase;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.StringDeserializer;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
@@ -37,14 +42,27 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.io.IOException;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
+import java.util.Properties;
import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
import static
org.apache.hudi.common.testutils.HoodieTestTable.makeNewCommitTime;
import static
org.apache.hudi.common.testutils.HoodieTestUtils.generateFakeHoodieWriteStat;
-import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class TestKafkaCallbackProvider extends UtilitiesTestBase {
+ private static final long POLL_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(60);
+
private final String testTopicName = "hoodie_test_" + UUID.randomUUID();
private KafkaTestUtils testUtils;
@@ -72,20 +90,68 @@ public class TestKafkaCallbackProvider extends
UtilitiesTestBase {
@Test
public void testCallbackMessage() {
- testUtils.createTopic(testTopicName, 2);
-
- HoodieWriteConfig hoodieConfig = createConfigForKafkaCallback();
- HoodieWriteCommitCallback commitCallback =
HoodieCommitCallbackFactory.create(hoodieConfig);
+ int numPartitions = 2;
+ testUtils.createTopic(testTopicName, numPartitions);
List<HoodieWriteStat> stats = generateFakeHoodieWriteStat(1);
- assertDoesNotThrow(() -> commitCallback.call(new
HoodieWriteCommitCallbackMessage(makeNewCommitTime(),
hoodieConfig.getTableName(), hoodieConfig.getBasePath(), stats)));
+ // without a partition config the message is routed by hashing the table
name key
+ HoodieWriteConfig defaultRoutedConfig = createConfigForKafkaCallback(null);
+ HoodieWriteCommitCallback defaultRoutedCallback =
HoodieCommitCallbackFactory.create(defaultRoutedConfig);
+ String defaultRoutedCommitTime = makeNewCommitTime();
+ defaultRoutedCallback.call(new HoodieWriteCommitCallbackMessage(
+ defaultRoutedCommitTime, defaultRoutedConfig.getTableName(),
defaultRoutedConfig.getBasePath(), stats));
+
+ // an explicit partition config overrides the key hashing
+ HoodieWriteConfig pinnedConfig = createConfigForKafkaCallback("1");
+ HoodieWriteCommitCallback pinnedCallback =
HoodieCommitCallbackFactory.create(pinnedConfig);
+ String pinnedCommitTime = makeNewCommitTime(Instant.now().plusSeconds(1));
+ pinnedCallback.call(new HoodieWriteCommitCallbackMessage(
+ pinnedCommitTime, pinnedConfig.getTableName(),
pinnedConfig.getBasePath(), stats));
+
+ // call() swallows send failures, so consuming the topic is the only proof
the sends went through;
+ // hashing the table name key routes to partition 0, so partition 1 can
only come from the config
+ List<ConsumerRecord<String, String>> consumed =
consumeCallbackMessages(numPartitions, 2);
+ assertEquals(Arrays.asList(0, 1),
+
consumed.stream().map(ConsumerRecord::partition).sorted().collect(Collectors.toList()));
+ Map<Integer, String> expectedCommitTimeByPartition = new HashMap<>();
+ expectedCommitTimeByPartition.put(0, defaultRoutedCommitTime);
+ expectedCommitTimeByPartition.put(1, pinnedCommitTime);
+ for (ConsumerRecord<String, String> record : consumed) {
+
assertTrue(record.value().contains(expectedCommitTimeByPartition.get(record.partition())),
+ () -> "unexpected callback message on partition " +
record.partition() + ": " + record.value());
+ }
+ }
+
+ private List<ConsumerRecord<String, String>> consumeCallbackMessages(int
numPartitions, int expectedCount) {
+ Properties consumerProps = new Properties();
+ consumerProps.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
testUtils.brokerAddress());
+ consumerProps.setProperty(ConsumerConfig.GROUP_ID_CONFIG,
"test-kafka-callback-" + UUID.randomUUID());
+ consumerProps.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
+ consumerProps.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
+
+ List<ConsumerRecord<String, String>> records = new ArrayList<>();
+ try (KafkaConsumer<String, String> consumer = new
KafkaConsumer<>(consumerProps)) {
+ List<TopicPartition> partitions = IntStream.range(0, numPartitions)
+ .mapToObj(partition -> new TopicPartition(testTopicName, partition))
+ .collect(Collectors.toList());
+ consumer.assign(partitions);
+ consumer.seekToBeginning(partitions);
+ long deadline = System.currentTimeMillis() + POLL_TIMEOUT_MS;
+ while (records.size() < expectedCount && System.currentTimeMillis() <
deadline) {
+ consumer.poll(Duration.ofSeconds(1)).forEach(records::add);
+ }
+ }
+ return records;
}
- private HoodieWriteConfig createConfigForKafkaCallback() {
+ private HoodieWriteConfig createConfigForKafkaCallback(String partition) {
TypedProperties props = new TypedProperties();
props.setProperty(HoodieWriteCommitKafkaCallbackConfig.TOPIC.key(),
testTopicName);
props.setProperty(HoodieWriteCommitKafkaCallbackConfig.BOOTSTRAP_SERVERS.key(),
testUtils.brokerAddress());
+ if (partition != null) {
+ props.setProperty(HoodieWriteCommitKafkaCallbackConfig.PARTITION.key(),
partition);
+ }
HoodieWriteConfig hoodieWriteConfig = HoodieWriteConfig.newBuilder()
.withCallbackConfig(
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/config/TestORCDFSSourceConfig.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/config/TestORCDFSSourceConfig.java
new file mode 100644
index 000000000000..331eb36e1e6d
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/config/TestORCDFSSourceConfig.java
@@ -0,0 +1,45 @@
+/*
+ * 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.config;
+
+import org.apache.hudi.common.util.Option;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Unit tests for {@link ORCDFSSourceConfig}.
+ *
+ * <p>ORCDFSSource itself stays untested until the HUDI-8081
hive-exec/hive-storage-api class-path
+ * collision (VectorizedRowBatch#isSelectedInUse) is fixed for this module.
+ */
+class TestORCDFSSourceConfig {
+
+ @Test
+ void testMergeSchemaConfig() {
+ assertEquals("hoodie.streamer.source.orc.dfs.merge.schema.enable",
+ ORCDFSSourceConfig.ORC_DFS_MERGE_SCHEMA.key());
+ assertTrue(ORCDFSSourceConfig.ORC_DFS_MERGE_SCHEMA.defaultValue());
+ assertTrue(ORCDFSSourceConfig.ORC_DFS_MERGE_SCHEMA.isAdvanced());
+ assertEquals(Option.of("1.2.0"),
ORCDFSSourceConfig.ORC_DFS_MERGE_SCHEMA.getSinceVersion());
+ }
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/exception/TestUtilitiesExceptions.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/exception/TestUtilitiesExceptions.java
new file mode 100644
index 000000000000..e49ebb4b061b
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/exception/TestUtilitiesExceptions.java
@@ -0,0 +1,169 @@
+/*
+ * 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.exception;
+
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.utilities.ingestion.HoodieIngestionException;
+
+import org.junit.jupiter.api.Test;
+
+import java.sql.SQLException;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests the thin hudi-utilities exception types that wrap {@link
HoodieException}.
+ */
+class TestUtilitiesExceptions {
+
+ @Test
+ void sourceTimeoutExceptionPreservesMessageAndCause() {
+ Throwable cause = new InterruptedException("waited too long");
+ HoodieSourceTimeoutException withCause = new
HoodieSourceTimeoutException("source timed out", cause);
+ assertEquals("source timed out", withCause.getMessage());
+ assertSame(cause, withCause.getCause());
+ assertTrue(withCause instanceof HoodieException);
+
+ HoodieSourceTimeoutException messageOnly = new
HoodieSourceTimeoutException("source timed out");
+ assertEquals("source timed out", messageOnly.getMessage());
+ assertNull(messageOnly.getCause());
+ }
+
+ @Test
+ void streamerExceptionsPreserveMessageAndCause() {
+ Throwable cause = new IllegalStateException("boom");
+ HoodieStreamerException withCause = new HoodieStreamerException("streamer
failed", cause);
+ assertEquals("streamer failed", withCause.getMessage());
+ assertSame(cause, withCause.getCause());
+ assertTrue(withCause instanceof HoodieException);
+
+ HoodieStreamerException messageOnly = new
HoodieStreamerException("streamer failed");
+ assertEquals("streamer failed", messageOnly.getMessage());
+ assertNull(messageOnly.getCause());
+
+ HoodieStreamerWriteException writeWithCause = new
HoodieStreamerWriteException("write failed", cause);
+ assertEquals("write failed", writeWithCause.getMessage());
+ assertSame(cause, writeWithCause.getCause());
+ assertTrue(writeWithCause instanceof HoodieStreamerException);
+
+ HoodieStreamerWriteException writeMessageOnly = new
HoodieStreamerWriteException("write failed");
+ assertEquals("write failed", writeMessageOnly.getMessage());
+ assertNull(writeMessageOnly.getCause());
+ }
+
+ @Test
+ void ingestionExceptionPreservesMessageAndCause() {
+ HoodieIngestionException messageOnly = new
HoodieIngestionException("ingestion failed");
+ assertEquals("ingestion failed", messageOnly.getMessage());
+ assertNull(messageOnly.getCause());
+ assertTrue(messageOnly instanceof HoodieException);
+
+ Throwable cause = new IllegalArgumentException("bad config");
+ HoodieIngestionException causeOnly = new HoodieIngestionException(cause);
+ assertSame(cause, causeOnly.getCause());
+ // Throwable-only ctor derives the message from the cause.
+ assertEquals(cause.toString(), causeOnly.getMessage());
+ }
+
+ @Test
+ void transformPlanExceptionPreservesMessageAndCause() {
+ Throwable cause = new RuntimeException("bad sql");
+ HoodieTransformPlanException withCause = new
HoodieTransformPlanException("planning failed", cause);
+ assertEquals("planning failed", withCause.getMessage());
+ assertSame(cause, withCause.getCause());
+ assertTrue(withCause instanceof HoodieTransformException);
+
+ HoodieTransformPlanException messageOnly = new
HoodieTransformPlanException("planning failed");
+ assertEquals("planning failed", messageOnly.getMessage());
+ assertNull(messageOnly.getCause());
+ }
+
+ @Test
+ void sourcePostProcessExceptionPreservesMessageAndCause() {
+ Throwable cause = new RuntimeException("post process");
+ HoodieSourcePostProcessException withCause = new
HoodieSourcePostProcessException("source post process failed", cause);
+ assertEquals("source post process failed", withCause.getMessage());
+ assertSame(cause, withCause.getCause());
+ assertTrue(withCause instanceof HoodieException);
+
+ HoodieSourcePostProcessException messageOnly = new
HoodieSourcePostProcessException("source post process failed");
+ assertEquals("source post process failed", messageOnly.getMessage());
+ assertNull(messageOnly.getCause());
+ }
+
+ @Test
+ void schemaPostProcessExceptionPreservesMessageAndCause() {
+ Throwable cause = new RuntimeException("post process");
+ HoodieSchemaPostProcessException withCause = new
HoodieSchemaPostProcessException("schema post process failed", cause);
+ assertEquals("schema post process failed", withCause.getMessage());
+ assertSame(cause, withCause.getCause());
+ assertTrue(withCause instanceof HoodieException);
+
+ HoodieSchemaPostProcessException messageOnly = new
HoodieSchemaPostProcessException("schema post process failed");
+ assertEquals("schema post process failed", messageOnly.getMessage());
+ assertNull(messageOnly.getCause());
+ }
+
+ @Test
+ void schemaProviderExceptionsPreserveMessageAndCause() {
+ Throwable cause = new RuntimeException("registry down");
+ HoodieSchemaProviderException withCause = new
HoodieSchemaProviderException("provider failed", cause);
+ assertEquals("provider failed", withCause.getMessage());
+ assertSame(cause, withCause.getCause());
+ assertTrue(withCause instanceof HoodieException);
+
+ HoodieSchemaProviderException messageOnly = new
HoodieSchemaProviderException("provider failed");
+ assertEquals("provider failed", messageOnly.getMessage());
+ assertNull(messageOnly.getCause());
+
+ HoodieSchemaFetchException fetchWithCause = new
HoodieSchemaFetchException("fetch failed", cause);
+ assertEquals("fetch failed", fetchWithCause.getMessage());
+ assertSame(cause, fetchWithCause.getCause());
+ assertTrue(fetchWithCause instanceof HoodieSchemaProviderException);
+
+ HoodieSchemaFetchException fetchMessageOnly = new
HoodieSchemaFetchException("fetch failed");
+ assertEquals("fetch failed", fetchMessageOnly.getMessage());
+ assertNull(fetchMessageOnly.getCause());
+ }
+
+ @Test
+ void incrementalPullExceptionsPreserveMessageAndCause() {
+ SQLException cause = new SQLException("syntax error");
+ HoodieIncrementalPullException withCause = new
HoodieIncrementalPullException("pull failed", cause);
+ assertEquals("pull failed", withCause.getMessage());
+ assertSame(cause, withCause.getCause());
+ assertTrue(withCause instanceof HoodieException);
+
+ HoodieIncrementalPullException messageOnly = new
HoodieIncrementalPullException("pull failed");
+ assertEquals("pull failed", messageOnly.getMessage());
+ assertNull(messageOnly.getCause());
+
+ HoodieIncrementalPullSQLException sqlWithCause = new
HoodieIncrementalPullSQLException("sql pull failed", cause);
+ assertEquals("sql pull failed", sqlWithCause.getMessage());
+ assertSame(cause, sqlWithCause.getCause());
+ assertTrue(sqlWithCause instanceof HoodieIncrementalPullException);
+
+ HoodieIncrementalPullSQLException sqlMessageOnly = new
HoodieIncrementalPullSQLException("sql pull failed");
+ assertEquals("sql pull failed", sqlMessageOnly.getMessage());
+ assertNull(sqlMessageOnly.getCause());
+ }
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/functional/TestAWSDatabaseMigrationServiceSource.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/functional/TestAWSDatabaseMigrationServiceSource.java
index 3b1da6d00b51..c5066eba4b8d 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/functional/TestAWSDatabaseMigrationServiceSource.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/functional/TestAWSDatabaseMigrationServiceSource.java
@@ -34,6 +34,9 @@ import java.io.IOException;
import java.io.Serializable;
import java.util.Arrays;
+import static org.apache.spark.sql.functions.lit;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -88,4 +91,18 @@ public class TestAWSDatabaseMigrationServiceSource extends
SparkClientFunctional
assertTrue(outputFrame.select(AWSDmsAvroPayload.OP_FIELD).collectAsList().stream()
.allMatch(r -> r.getString(0).equals("")));
}
+
+ @Test
+ public void testTransformerWithOpFieldAlreadyPresent() {
+ AWSDmsTransformer transformer = new AWSDmsTransformer();
+ Dataset<Row> inputFrame = spark().createDataFrame(Arrays.asList(
+ new Record("1", 3433L),
+ new Record("2", 3433L)), Record.class)
+ .withColumn(AWSDmsAvroPayload.OP_FIELD, lit("D"));
+
+ Dataset<Row> outputFrame = transformer.apply(jsc(), spark(), inputFrame,
null);
+ // the dataset is handed back untouched, the existing operation values are
kept as they are
+ assertArrayEquals(inputFrame.columns(), outputFrame.columns());
+ assertEquals(inputFrame.collectAsList(), outputFrame.collectAsList());
+ }
}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/schema/TestFilebasedSchemaProvider.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/schema/TestFilebasedSchemaProvider.java
index cfb7aac85005..a34b81516d35 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/schema/TestFilebasedSchemaProvider.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/schema/TestFilebasedSchemaProvider.java
@@ -19,11 +19,15 @@
package org.apache.hudi.utilities.schema;
import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.exception.HoodieAvroSchemaException;
+import org.apache.hudi.utilities.config.FilebasedSchemaProviderConfig;
import org.apache.hudi.utilities.config.HoodieSchemaProviderConfig;
+import org.apache.hudi.utilities.exception.HoodieSchemaProviderException;
import org.apache.hudi.utilities.schema.converter.JsonToAvroSchemaConverter;
import org.apache.hudi.utilities.testutils.UtilitiesTestBase;
+import io.confluent.kafka.schemaregistry.ParsedSchema;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -37,6 +41,7 @@ import static
org.apache.hudi.utilities.testutils.SanitizationTestUtils.generate
import static
org.apache.hudi.utilities.testutils.SanitizationTestUtils.generateRenamedSchemaWithDefaultReplacement;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* Unit tests for {@link FilebasedSchemaProvider}.
@@ -97,4 +102,84 @@ class TestFilebasedSchemaProvider extends UtilitiesTestBase
{
assertEquals(filebasedSchemaProvider.getSourceHoodieSchema(),
jsonFilebasedSchemaProvider.getSourceHoodieSchema());
}
+
+ @Test
+ void testJsonSchemaWithUnknownConverterClass() throws IOException {
+ TypedProperties props = Helpers.setupSchemaOnDFS("streamer-config",
"source_uber_encoded_decimal.json");
+ props.setProperty(HoodieSchemaProviderConfig.SCHEMA_CONVERTER.key(),
"org.apache.hudi.utilities.NoSuchSchemaConverter");
+ Throwable t = assertThrows(HoodieSchemaProviderException.class, () -> new
FilebasedSchemaProvider(props, jsc));
+ assertTrue(t.getMessage().contains("Error loading json schema converter"),
t.getMessage());
+ }
+
+ @Test
+ void testJsonSchemaWithFailingConverter() throws IOException {
+ TypedProperties props = Helpers.setupSchemaOnDFS("streamer-config",
"source_uber_encoded_decimal.json");
+ props.setProperty(HoodieSchemaProviderConfig.SCHEMA_CONVERTER.key(),
FailingSchemaConverter.class.getName());
+ Throwable t = assertThrows(HoodieSchemaProviderException.class, () -> new
FilebasedSchemaProvider(props, jsc));
+ assertTrue(t.getMessage().contains("Error converting json schema"),
t.getMessage());
+ }
+
+ @Test
+ void testMissingSchemaFile() {
+ TypedProperties props = new TypedProperties();
+ props.setProperty(FilebasedSchemaProviderConfig.SOURCE_SCHEMA_FILE.key(),
basePath + "/no_such_schema.avsc");
+ Throwable t = assertThrows(HoodieSchemaProviderException.class, () -> new
FilebasedSchemaProvider(props, jsc));
+ assertTrue(t.getMessage().contains("Error reading schema from file"),
t.getMessage());
+ }
+
+ @Test
+ void testRefreshPicksUpRewrittenSourceAndTargetSchemaFiles() throws
IOException {
+ TypedProperties targetProps = Helpers.setupSchemaOnDFS("streamer-config",
"source_uber_encoded_decimal.avsc");
+ TypedProperties props = Helpers.setupSchemaOnDFS("streamer-config",
"file_schema_provider_valid.avsc");
+ props.setProperty(FilebasedSchemaProviderConfig.TARGET_SCHEMA_FILE.key(),
+
targetProps.getString(FilebasedSchemaProviderConfig.SOURCE_SCHEMA_FILE.key()));
+ this.schemaProvider = new FilebasedSchemaProvider(props, jsc);
+ assertEquals(this.schemaProvider.getSourceHoodieSchema(),
generateProperFormattedSchema());
+
+ // rewrite the configured source schema file in place with an unrelated
schema: refresh() has to
+ // re-read the file rather than serve the schema cached at construction
time
+ HoodieSchema rewrittenSchema = new FilebasedSchemaProvider(
+ Helpers.setupSchemaOnDFS("streamer-config", "source_uber.avsc"),
jsc).getSourceHoodieSchema();
+ Helpers.copyToDFS("streamer-config/source_uber.avsc", storage,
+
props.getString(FilebasedSchemaProviderConfig.SOURCE_SCHEMA_FILE.key()));
+
+ this.schemaProvider.refresh();
+
+ assertEquals(rewrittenSchema, this.schemaProvider.getSourceHoodieSchema());
+ assertEquals(this.schemaProvider.getTargetHoodieSchema(),
+ new FilebasedSchemaProvider(targetProps, jsc).getSourceHoodieSchema());
+ }
+
+ @Test
+ void testRefreshWithoutTargetSchemaFileConfigured() throws IOException {
+ TypedProperties props = Helpers.setupSchemaOnDFS("streamer-config",
"file_schema_provider_valid.avsc");
+ FilebasedSchemaProvider provider = new FilebasedSchemaProvider(props, jsc);
+ // without a target schema file the constructor leaves the target schema
unset, so the target
+ // falls back to the source schema
+ assertEquals(generateProperFormattedSchema(),
provider.getTargetHoodieSchema());
+
+ // refresh() populates the target schema from the target file, which
defaults to the source
+ // file, so a rewrite of that file now surfaces through the populated
branch of getTargetSchema()
+ HoodieSchema rewrittenSchema = new FilebasedSchemaProvider(
+ Helpers.setupSchemaOnDFS("streamer-config", "source_uber.avsc"),
jsc).getSourceHoodieSchema();
+ Helpers.copyToDFS("streamer-config/source_uber.avsc", storage,
+
props.getString(FilebasedSchemaProviderConfig.SOURCE_SCHEMA_FILE.key()));
+ provider.refresh();
+
+ assertEquals(rewrittenSchema, provider.getTargetHoodieSchema());
+ }
+
+ /**
+ * Json schema converter whose conversion always fails, to exercise the
conversion failure path.
+ */
+ public static class FailingSchemaConverter implements
SchemaRegistryProvider.SchemaConverter {
+
+ public FailingSchemaConverter(TypedProperties props) {
+ }
+
+ @Override
+ public String convert(ParsedSchema schema) throws IOException {
+ throw new IOException("simulated json schema conversion failure");
+ }
+ }
}
\ No newline at end of file
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/schema/TestProtoClassBasedSchemaProvider.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/schema/TestProtoClassBasedSchemaProvider.java
index fe0bb619d24b..f3f0f2f545e6 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/schema/TestProtoClassBasedSchemaProvider.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/schema/TestProtoClassBasedSchemaProvider.java
@@ -21,6 +21,7 @@ package org.apache.hudi.utilities.schema;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.internal.HoodieSchemaException;
import org.apache.hudi.utilities.config.ProtoClassBasedSchemaProviderConfig;
import org.apache.hudi.utilities.test.proto.Parent;
import org.apache.hudi.utilities.test.proto.Sample;
@@ -83,4 +84,36 @@ public class TestProtoClassBasedSchemaProvider {
HoodieSchema expectedSchema = new
HoodieSchema.Parser().parse(getClass().getClassLoader().getResourceAsStream("schema-provider/proto/oneof_schema.avsc"));
Assertions.assertEquals(expectedSchema, protoSchema);
}
+
+ @Test
+ public void validateTargetSchemaFallsBackToSourceSchema() {
+ TypedProperties properties = new TypedProperties();
+
properties.setProperty(ProtoClassBasedSchemaProviderConfig.PROTO_SCHEMA_CLASS_NAME.key(),
Sample.class.getName());
+ ProtoClassBasedSchemaProvider protoToAvroSchemaProvider = new
ProtoClassBasedSchemaProvider(properties, null);
+ HoodieSchema expectedSchema = new
HoodieSchema.Parser().parse(getClass().getClassLoader().getResourceAsStream("schema-provider/proto/sample_schema_defaults.avsc"));
+ // no target schema is configurable for this provider, so both accessors
must serve the source schema
+ Assertions.assertEquals(expectedSchema,
HoodieSchema.fromAvroSchema(protoToAvroSchemaProvider.getSourceSchema()));
+ Assertions.assertEquals(expectedSchema,
HoodieSchema.fromAvroSchema(protoToAvroSchemaProvider.getTargetSchema()));
+ }
+
+ @Test
+ public void validateUnknownProtoClassFailsOnConstruction() {
+ TypedProperties properties = new TypedProperties();
+
properties.setProperty(ProtoClassBasedSchemaProviderConfig.PROTO_SCHEMA_CLASS_NAME.key(),
+ "org.apache.hudi.utilities.test.proto.NoSuchMessage");
+ // the proto class is loaded eagerly, so an unknown class is rejected
before any schema is served
+ Assertions.assertThrows(HoodieSchemaException.class, () -> new
ProtoClassBasedSchemaProvider(properties, null));
+ }
+
+ @Test
+ public void validateDeprecatedConfigConstants() {
+ Assertions.assertEquals("hoodie.streamer.schemaprovider.proto.class.name",
+ ProtoClassBasedSchemaProvider.Config.PROTO_SCHEMA_CLASS_NAME.key());
+
Assertions.assertEquals("hoodie.streamer.schemaprovider.proto.flatten.wrappers",
+
ProtoClassBasedSchemaProvider.Config.PROTO_SCHEMA_WRAPPED_PRIMITIVES_AS_RECORDS.key());
+
Assertions.assertEquals("hoodie.streamer.schemaprovider.proto.timestamps.as.records",
+
ProtoClassBasedSchemaProvider.Config.PROTO_SCHEMA_TIMESTAMPS_AS_RECORDS.key());
+
Assertions.assertEquals("hoodie.streamer.schemaprovider.proto.max.recursion.depth",
+
ProtoClassBasedSchemaProvider.Config.PROTO_SCHEMA_MAX_RECURSION_DEPTH.key());
+ }
}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/schema/TestRowBasedSchemaProvider.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/schema/TestRowBasedSchemaProvider.java
new file mode 100644
index 000000000000..605d5444106e
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/schema/TestRowBasedSchemaProvider.java
@@ -0,0 +1,44 @@
+/*
+ * 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.schema;
+
+import org.apache.avro.Schema;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructType;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+/**
+ * Unit tests for {@link RowBasedSchemaProvider}.
+ */
+class TestRowBasedSchemaProvider {
+
+ @Test
+ void testSourceSchemaIsDerivedFromRowStruct() {
+ Schema sourceSchema = new RowBasedSchemaProvider(
+ new StructType().add("id", DataTypes.LongType,
false)).getSourceSchema();
+
+ assertEquals(RowBasedSchemaProvider.HOODIE_RECORD_STRUCT_NAME,
sourceSchema.getName());
+ assertEquals(RowBasedSchemaProvider.HOODIE_RECORD_NAMESPACE,
sourceSchema.getNamespace());
+ assertNotNull(sourceSchema.getField("id"));
+ assertEquals(Schema.Type.LONG,
sourceSchema.getField("id").schema().getType());
+ }
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/schema/TestSchemaRegistryProvider.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/schema/TestSchemaRegistryProvider.java
index 644ca30e3da5..ac4d432dbbe0 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/schema/TestSchemaRegistryProvider.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/schema/TestSchemaRegistryProvider.java
@@ -34,6 +34,7 @@ import java.util.Collections;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doReturn;
@@ -181,4 +182,11 @@ class TestSchemaRegistryProvider {
// Verify that the fallback schema is returned.
assertEquals(HoodieSchema.parse(FALLBACK_SCHEMA), schema);
}
+
+ @Test
+ public void testNullTargetProviderReturnsNullTargetSchema() {
+ // the registry is never contacted: the url is only validated at
construction time
+ NullTargetSchemaRegistryProvider underTest = new
NullTargetSchemaRegistryProvider(getProps(), null);
+ assertNull(underTest.getTargetHoodieSchema());
+ }
}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestAvroKafkaSource.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestAvroKafkaSource.java
index 7bb14e2d3e84..54fe737eebec 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestAvroKafkaSource.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestAvroKafkaSource.java
@@ -310,4 +310,17 @@ public class TestAvroKafkaSource extends
SparkClientFunctionalTestHarness {
String schemaHash =
Base64.encode(HashID.hash(schemaProvider.getSourceHoodieSchema().toString(),
HashID.Size.BITS_128));
assertEquals(StringUtils.concatenateWithThreshold(String.format("%s_",
groupId), schemaHash, GROUP_ID_MAX_BYTES_LENGTH), newGroupId);
}
+
+ @Test
+ void testUnknownValueDeserializerClass() {
+ final String topic = TEST_TOPIC_PREFIX + "testUnknownValueDeserializer";
+ TypedProperties props = createPropsForKafkaSource(topic, null, "earliest");
+
+ props.put("hoodie.streamer.source.kafka.value.deserializer.class",
"org.apache.hudi.NotADeserializer");
+ // the deserializer class is resolved while constructing the source, i.e.
before touching kafka
+ HoodieReadFromSourceException exception =
assertThrows(HoodieReadFromSourceException.class,
+ () -> new AvroKafkaSource(props, jsc(), spark(), schemaProvider,
metrics));
+ assertTrue(exception.getMessage().contains(
+ "Could not load custom avro kafka deserializer:
org.apache.hudi.NotADeserializer"), exception.getMessage());
+ }
}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestJsonKafkaSourcePostProcessor.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestJsonKafkaSourcePostProcessor.java
index 1f22f6712be1..21c142bd3b5e 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestJsonKafkaSourcePostProcessor.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestJsonKafkaSourcePostProcessor.java
@@ -38,16 +38,20 @@ import org.apache.hudi.utilities.testutils.KafkaTestUtils;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.avro.generic.GenericRecord;
+import org.apache.spark.SparkException;
import org.apache.spark.api.java.JavaRDD;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
+import java.util.List;
import java.util.Objects;
import static
org.apache.hudi.utilities.config.JsonKafkaPostProcessorConfig.JSON_KAFKA_PROCESSOR_CLASS;
@@ -55,11 +59,18 @@ import static
org.apache.hudi.utilities.testutils.UtilitiesTestBase.Helpers.json
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
public class TestJsonKafkaSourcePostProcessor extends
SparkClientFunctionalTestHarness {
+ // database hudi, table hudi_maxwell_01, delete
+ private static final String MAXWELL_DELETE_RECORD =
"{\"database\":\"hudi\",\"table\":\"hudi_maxwell_01\","
+ + "\"type\":\"delete\",\"ts\":1647074555,\"xid\":6631,\"commit\":true,"
+ +
"\"data\":{\"id\":\"6018220e39e74477b45c7cf42f66bdc0\",\"name\":\"mathieu\",\"age\":20,"
+ + "\"insert_time\":\"2022-03-12 04:40:02\",\"update_time\":\"2022-03-12
04:42:25\"}}";
+
private static KafkaTestUtils testUtils;
private final HoodieIngestionMetrics metrics =
mock(HoodieIngestionMetrics.class);
@@ -171,21 +182,14 @@ public class TestJsonKafkaSourcePostProcessor extends
SparkClientFunctionalTestH
// Maxwell data
// ------------------------------------------------------------------------
- // database hudi, table hudi_maxwell_01 (insert, update and delete)
- String hudiMaxwell01Insert =
"{\"database\":\"hudi\",\"table\":\"hudi_maxwell_01\",\"type\":\"insert\","
- +
"\"ts\":1647074402,\"xid\":6233,\"commit\":true,\"data\":{\"id\":\"6018220e39e74477b45c7cf42f66bdc0\","
- + "\"name\":\"mathieu\",\"age\":18,\"insert_time\":\"2022-03-12
08:40:02\","
- + "\"update_time\":\"2022-03-12 08:40:02\"}}";
+ // database hudi, table hudi_maxwell_01 (insert and update, delete is
MAXWELL_DELETE_RECORD)
+ String hudiMaxwell01Insert = maxwellInsertRecord("hudi",
"hudi_maxwell_01");
String hudiMaxwell01Update =
"{\"database\":\"hudi\",\"table\":\"hudi_maxwell_01\",\"type\":\"update\","
+
"\"ts\":1647074482,\"xid\":6440,\"commit\":true,\"data\":{\"id\":\"6018220e39e74477b45c7cf42f66bdc0\","
+ "\"name\":\"mathieu\",\"age\":20,\"insert_time\":\"2022-03-12
04:40:02\",\"update_time\":\"2022-03-12 04:42:25\"},"
+ "\"old\":{\"age\":18,\"insert_time\":\"2022-03-12
08:40:02\",\"update_time\":\"2022-03-12 08:40:02\"}}";
- String hudiMaxwell01Delete =
"{\"database\":\"hudi\",\"table\":\"hudi_maxwell_01\",\"type\":\"delete\","
- +
"\"ts\":1647074555,\"xid\":6631,\"commit\":true,\"data\":{\"id\":\"6018220e39e74477b45c7cf42f66bdc0\","
- + "\"name\":\"mathieu\",\"age\":20,\"insert_time\":\"2022-03-12
04:40:02\",\"update_time\":\"2022-03-12 04:42:25\"}}";
-
String hudiMaxwell01Ddl =
"{\"type\":\"table-alter\",\"database\":\"hudi\",\"table\":\"hudi_maxwell_01\","
+
"\"old\":{\"database\":\"hudi\",\"charset\":\"utf8\",\"table\":\"hudi_maxwell_01\","
+
"\"primary-key\":[\"id\"],\"columns\":[{\"type\":\"varchar\",\"name\":\"id\",\"charset\":\"utf8\"},"
@@ -211,12 +215,6 @@ public class TestJsonKafkaSourcePostProcessor extends
SparkClientFunctionalTestH
+ "\"name\":\"andy\",\"age\":17,\"insert_time\":\"2022-03-12
08:31:56\","
+ "\"update_time\":\"2022-03-12 08:31:56\"}}";
- // database hudi_02, table hudi_maxwell_01, insert
- String hudi02Maxwell01Insert =
"{\"database\":\"hudi_02\",\"table\":\"hudi_maxwell_01\",\"type\":\"insert\","
- +
"\"ts\":1647073916,\"xid\":4990,\"commit\":true,\"data\":{\"id\":\"9bb17f316ee8488cb107621ddf0f3cb0\","
- + "\"name\":\"andy\",\"age\":17,\"insert_time\":\"2022-03-12
08:31:56\","
- + "\"update_time\":\"2022-03-12 08:31:56\"}}";
-
// ------------------------------------------------------------------------
// Tests
// ------------------------------------------------------------------------
@@ -244,9 +242,9 @@ public class TestJsonKafkaSourcePostProcessor extends
SparkClientFunctionalTestH
props.setProperty(JsonKafkaPostProcessorConfig.ORDERING_FIELDS_FORMAT.key(),
"yyyy-MM-dd HH:mm:ss");
props.setProperty(HoodieWriteConfig.PRECOMBINE_FIELD_NAME.key(),
"update_time");
- JavaRDD<String> inputDelete =
jsc().parallelize(Collections.singletonList(hudiMaxwell01Delete));
+ JavaRDD<String> inputDelete =
jsc().parallelize(Collections.singletonList(MAXWELL_DELETE_RECORD));
- long ts = mapper.readTree(hudiMaxwell01Delete).get("ts").longValue();
+ long ts = mapper.readTree(MAXWELL_DELETE_RECORD).get("ts").longValue();
String formatTs = DateTimeUtils.formatUnixTimestamp(ts, "yyyy-MM-dd
HH:mm:ss");
new MaxwellJsonKafkaSourcePostProcessor(props)
@@ -265,7 +263,7 @@ public class TestJsonKafkaSourcePostProcessor extends
SparkClientFunctionalTestH
props.setProperty(JsonKafkaPostProcessorConfig.ORDERING_FIELDS_TYPE.key(),
"NON_TIMESTAMP");
props.setProperty(HoodieWriteConfig.PRECOMBINE_FIELD_NAME.key(), "id");
- JavaRDD<String> inputDelete2 =
jsc().parallelize(Collections.singletonList(hudiMaxwell01Delete));
+ JavaRDD<String> inputDelete2 =
jsc().parallelize(Collections.singletonList(MAXWELL_DELETE_RECORD));
String updateTimeInUpdate =
mapper.readTree(hudiMaxwell01Update).get("data").get("update_time").textValue();
new MaxwellJsonKafkaSourcePostProcessor(props)
@@ -287,14 +285,81 @@ public class TestJsonKafkaSourcePostProcessor extends
SparkClientFunctionalTestH
// ddl data will be ignored, ths count should be 0
long ddlDataNum = processor.process(ddlData).count();
assertEquals(0, ddlDataNum);
+ }
+
+ /**
+ * The delete time of a maxwell record is rewritten with `ts` for every
timestamp-like ordering
+ * field type, using the numeric representation the type asks for.
+ */
+ @ParameterizedTest
+ @CsvSource({"EPOCHMILLISECONDS, 1647074555000", "UNIX_TIMESTAMP,
1647074555"})
+ public void testMaxwellPostProcessorNumericOrderingFieldsType(String
orderingFieldsType,
+ long
expectedOrderingValue) throws IOException {
+ TypedProperties props = maxwellProps();
+ props.setProperty(JsonKafkaPostProcessorConfig.ORDERING_FIELDS_TYPE.key(),
orderingFieldsType);
+ props.setProperty(HoodieWriteConfig.PRECOMBINE_FIELD_NAME.key(),
"update_time");
+
+ List<String> processed = new MaxwellJsonKafkaSourcePostProcessor(props)
+
.process(jsc().parallelize(Collections.singletonList(MAXWELL_DELETE_RECORD))).collect();
+
+ assertEquals(1, processed.size());
+ JsonNode record = new ObjectMapper().readTree(processed.get(0));
+
assertTrue(record.get(HoodieRecord.HOODIE_IS_DELETED_FIELD).booleanValue());
+ assertEquals(expectedOrderingValue, record.get("update_time").longValue());
+ }
+
+ /**
+ * An unknown ordering field type is rejected while resolving the enum, i.e.
before the processor
+ * gets a chance to reach its unsupported-format branch.
+ */
+ @Test
+ public void testMaxwellPostProcessorWithUnknownOrderingFieldsType() {
+ TypedProperties props = maxwellProps();
+ props.setProperty(JsonKafkaPostProcessorConfig.ORDERING_FIELDS_TYPE.key(),
"EPOCH_SECONDS");
+ props.setProperty(HoodieWriteConfig.PRECOMBINE_FIELD_NAME.key(),
"update_time");
+ MaxwellJsonKafkaSourcePostProcessor processor = new
MaxwellJsonKafkaSourcePostProcessor(props);
+ JavaRDD<String> input =
jsc().parallelize(Collections.singletonList(MAXWELL_DELETE_RECORD));
+
+ // the ordering field type is read inside the map task, so spark wraps up
the failure
+ SparkException exception = Assertions.assertThrows(SparkException.class,
() -> processor.process(input).collect());
+ Throwable cause = exception;
+ while (cause != null && !(cause instanceof IllegalArgumentException)) {
+ cause = cause.getCause();
+ }
+ assertNotNull(cause);
+ assertTrue(cause.getMessage().contains("EPOCH_SECONDS"));
+ }
+
+ /**
+ * Without a database regex, records are kept based on the table name alone.
+ */
+ @Test
+ public void testMaxwellPostProcessorWithoutDatabaseRegex() {
+ TypedProperties props = new TypedProperties();
+ props.setProperty(JsonKafkaPostProcessorConfig.TABLE_NAME_REGEX.key(),
"hudi_maxwell(_)?[0-9]{0,2}");
+ MaxwellJsonKafkaSourcePostProcessor processor = new
MaxwellJsonKafkaSourcePostProcessor(props);
- // test table regex without database regex
- props.remove(JsonKafkaPostProcessorConfig.DATABASE_NAME_REGEX.key());
+ JavaRDD<String> input = jsc().parallelize(Arrays.asList(
+ maxwellInsertRecord("hudi", "hudi_maxwell_01"),
+ maxwellInsertRecord("not_a_hudi_database", "hudi_maxwell_02"),
+ maxwellInsertRecord("hudi", "not_a_target_table")));
+
+ // both target tables are kept, whatever the database they come from
+ assertEquals(2, processor.process(input).count());
+ }
+
+ private static TypedProperties maxwellProps() {
+ TypedProperties props = new TypedProperties();
+ props.setProperty(JsonKafkaPostProcessorConfig.DATABASE_NAME_REGEX.key(),
"hudi(_)?[0-9]{0,2}");
props.setProperty(JsonKafkaPostProcessorConfig.TABLE_NAME_REGEX.key(),
"hudi_maxwell(_)?[0-9]{0,2}");
+ return props;
+ }
- JavaRDD<String> dataWithoutDatabaseRegex =
jsc().parallelize(Arrays.asList(hudiMaxwell01Insert, hudi02Maxwell01Insert));
- long countWithoutDatabaseRegex =
processor.process(dataWithoutDatabaseRegex).count();
- assertEquals(2, countWithoutDatabaseRegex);
+ private static String maxwellInsertRecord(String database, String table) {
+ return "{\"database\":\"" + database + "\",\"table\":\"" + table +
"\",\"type\":\"insert\","
+ + "\"ts\":1647074402,\"xid\":6233,\"commit\":true,"
+ +
"\"data\":{\"id\":\"6018220e39e74477b45c7cf42f66bdc0\",\"name\":\"mathieu\",\"age\":18,"
+ + "\"insert_time\":\"2022-03-12
08:40:02\",\"update_time\":\"2022-03-12 08:40:02\"}}";
}
/**
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestProtoKafkaSource.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestProtoKafkaSource.java
index a5cabddf9a31..7600797354f9 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestProtoKafkaSource.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestProtoKafkaSource.java
@@ -21,8 +21,10 @@ package org.apache.hudi.utilities.sources;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieErrorTableConfig;
+import org.apache.hudi.utilities.config.HoodieStreamerConfig;
import org.apache.hudi.utilities.config.KafkaSourceConfig;
import org.apache.hudi.utilities.config.ProtoClassBasedSchemaProviderConfig;
+import org.apache.hudi.utilities.exception.HoodieReadFromSourceException;
import org.apache.hudi.utilities.schema.ProtoClassBasedSchemaProvider;
import org.apache.hudi.utilities.schema.SchemaProvider;
import org.apache.hudi.utilities.schema.SchemaRegistryProvider;
@@ -53,10 +55,12 @@ import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
@@ -74,6 +78,8 @@ import java.util.stream.IntStream;
import static org.apache.hudi.common.util.StringUtils.getUTF8Bytes;
import static
org.apache.hudi.utilities.config.KafkaSourceConfig.KAFKA_PROTO_VALUE_DESERIALIZER_CLASS;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* Tests against {@link ProtoKafkaSource}.
@@ -178,6 +184,32 @@ public class TestProtoKafkaSource extends
BaseTestKafkaSource {
assertEquals(Option.empty(), fetch4AsRows.getBatch());
}
+ @Test
+ public void testProtoKafkaSourceWithUnsupportedDeserializer() {
+ TypedProperties props = createPropsForKafkaSource(TEST_TOPIC_PREFIX +
"test_proto_unsupported_deserializer", null, "earliest");
+ props.put(KAFKA_PROTO_VALUE_DESERIALIZER_CLASS.key(),
StringDeserializer.class.getName());
+ SchemaProvider schemaProvider = new ProtoClassBasedSchemaProvider(props,
jsc());
+
+ // the deserializer is validated while constructing the source, i.e.
before touching kafka
+ HoodieReadFromSourceException exception =
assertThrows(HoodieReadFromSourceException.class,
+ () -> new ProtoKafkaSource(props, jsc(), spark(), schemaProvider,
metrics));
+ assertTrue(exception.getMessage().contains(
+ "Only ByteArrayDeserializer and KafkaProtobufDeserializer are
supported for ProtoKafkaSource"), exception.getMessage());
+ }
+
+ @Test
+ public void testProtoKafkaSourceWithKafkaOffsetsAppended() {
+ TypedProperties props = createPropsForKafkaSource(TEST_TOPIC_PREFIX +
"test_proto_kafka_offsets_appended", null, "earliest");
+ props.put(HoodieStreamerConfig.KAFKA_APPEND_OFFSETS.key(), "true");
+ SchemaProvider schemaProvider = new ProtoClassBasedSchemaProvider(props,
jsc());
+
+ // appending kafka offsets is not supported for proto sources
+ HoodieReadFromSourceException exception =
assertThrows(HoodieReadFromSourceException.class,
+ () -> new ProtoKafkaSource(props, jsc(), spark(), schemaProvider,
metrics));
+ assertTrue(exception.getMessage().contains(
+ "Appending kafka offsets to ProtoKafkaSource is not supported"),
exception.getMessage());
+ }
+
private static List<Sample> createSampleMessages(int count) {
return IntStream.range(0, count).mapToObj(unused -> {
Sample.Builder builder = Sample.newBuilder()
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestS3EventsSource.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestS3EventsSource.java
index 764d7a571280..21a6d0358459 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestS3EventsSource.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestS3EventsSource.java
@@ -20,20 +20,29 @@ package org.apache.hudi.utilities.sources;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.table.checkpoint.Checkpoint;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.utilities.schema.FilebasedSchemaProvider;
+import org.apache.hudi.utilities.schema.SchemaProvider;
import org.apache.hudi.utilities.streamer.SourceFormatAdapter;
+import org.apache.hudi.utilities.testutils.CloudObjectTestUtils;
import
org.apache.hudi.utilities.testutils.sources.AbstractCloudObjectsSourceTestBase;
import org.apache.avro.generic.GenericRecord;
import org.apache.hadoop.fs.Path;
import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
+import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchRequest;
import java.io.IOException;
+import java.util.Arrays;
import java.util.List;
import static
org.apache.hudi.config.HoodieErrorTableConfig.ERROR_TABLE_PERSIST_SOURCE_RDD;
@@ -42,7 +51,11 @@ import static
org.apache.hudi.utilities.config.S3SourceConfig.S3_SOURCE_QUEUE_RE
import static
org.apache.hudi.utilities.config.S3SourceConfig.S3_SOURCE_QUEUE_URL;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
/**
* Basic tests for {@link S3EventsSource}.
@@ -103,9 +116,57 @@ public class TestS3EventsSource extends
AbstractCloudObjectsSourceTestBase {
verifyRddsArePersisted(sourceFormatAdapter, fetch2, persistSourceRdd);
}
+ /**
+ * Without a schema provider the event schema is inferred from the SQS
payload instead of being
+ * applied on read.
+ */
+ @Test
+ public void testReadingFromSourceWithoutSchemaProvider() {
+ S3EventsSource source = prepareS3EventsSource(generateProperties(false),
null);
+ generateMessageInQueue("1");
+
+ Pair<Option<Dataset<Row>>, Checkpoint> batch =
source.fetchNextBatch(Option.empty(), Long.MAX_VALUE);
+ Dataset<Row> eventRecords = batch.getLeft().get();
+ assertEquals(1, eventRecords.count());
+ // inference has to recover the whole sqs event, nested structs included:
a payload spark fails
+ // to infer surfaces as a single _corrupt_record column instead;
glacierEventData is not declared
+ // in s3-metadata.avsc, so its presence pins the schema to inference
rather than the provider
+ assertEquals(Arrays.asList("awsRegion", "eventName", "eventSource",
"eventTime", "eventVersion",
+ "glacierEventData", "requestParameters", "s3", "userIdentity"),
Arrays.asList(eventRecords.columns()));
+ Row eventRecord = eventRecords.select("s3.object.key",
"s3.object.size").first();
+ assertEquals("1.parquet", eventRecord.getString(0));
+ assertEquals(123L, eventRecord.getLong(1));
+ assertNotNull(batch.getRight());
+ }
+
+ /**
+ * Messages picked up by a fetch are deleted from the queue on commit,
exactly once, and the sqs
+ * client is released when the source is closed.
+ */
+ @Test
+ public void testOnCommitDeletesProcessedMessagesAndClose() throws
IOException {
+ S3EventsSource source = prepareS3EventsSource(generateProperties(false),
schemaProvider);
+ generateMessageInQueue("1");
+ CloudObjectTestUtils.deleteMessagesInQueue(sqs);
+ source.fetchNextBatch(Option.empty(), Long.MAX_VALUE);
+
+ source.onCommit("1");
+ verify(sqs,
times(1)).deleteMessageBatch(any(DeleteMessageBatchRequest.class));
+ // the processed messages are cleared on commit, so a second commit has
nothing left to delete
+ source.onCommit("2");
+ verify(sqs,
times(1)).deleteMessageBatch(any(DeleteMessageBatchRequest.class));
+
+ source.close();
+ verify(sqs).close();
+ }
+
@Override
public Source prepareCloudObjectSource(TypedProperties props) {
- S3EventsSource dfsSource = new S3EventsSource(props, jsc, sparkSession,
schemaProvider);
+ return prepareS3EventsSource(props, schemaProvider);
+ }
+
+ private S3EventsSource prepareS3EventsSource(TypedProperties props,
SchemaProvider sourceSchemaProvider) {
+ S3EventsSource dfsSource = new S3EventsSource(props, jsc, sparkSession,
sourceSchemaProvider);
dfsSource.sqs = this.sqs;
return dfsSource;
}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestSource.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestSource.java
index e6e4897e05f1..6ad38c3c0fe9 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestSource.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestSource.java
@@ -25,9 +25,11 @@ import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
/**
- * Unit tests for {@link Source#releaseResources()}.
+ * Unit tests for {@link Source}, its {@link Source#releaseResources()}
contract and the typed
+ * {@link JsonSource}/{@link AvroSource} base classes.
*
* <p>releaseResources() is invoked from StreamSync.syncOnce()'s finally
block, after the
* write/commit has already completed. A transient Spark failure while
unpersisting the
@@ -70,4 +72,25 @@ public class TestSource {
"A transient unpersist failure in cleanup must not propagate out of
releaseResources()");
assertEquals(1, source.unpersistCalls, "unpersist should have been
attempted exactly once");
}
+
+ /**
+ * The typed source base classes advertise their source type and reject the
deprecated
+ * fetchNewData() entry point, which their subclasses replace with
readFromCheckpoint().
+ */
+ @Test
+ public void jsonAndAvroSourcesRejectDeprecatedFetchNewData() {
+ JsonSource jsonSource = new JsonSource(new TypedProperties(), null, null,
null) {
+ };
+ AvroSource avroSource = new AvroSource(new TypedProperties(), null, null,
null) {
+ };
+
+ assertEquals(Source.SourceType.JSON, jsonSource.getSourceType());
+ assertEquals(Source.SourceType.AVRO, avroSource.getSourceType());
+ assertThrows(UnsupportedOperationException.class, () ->
jsonSource.fetchNewData(Option.empty(), 100L));
+ assertThrows(UnsupportedOperationException.class, () ->
avroSource.fetchNewData(Option.empty(), 100L));
+
+ // commit callback and cleanup are no-ops for sources that cache nothing
+ assertDoesNotThrow(() -> jsonSource.onCommit("00000000000001"));
+ assertDoesNotThrow(jsonSource::releaseResources);
+ }
}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestSqlSource.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestSqlSource.java
index 8a393e8b982e..0d6a2a6000de 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestSqlSource.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestSqlSource.java
@@ -20,6 +20,7 @@ package org.apache.hudi.utilities.sources;
import org.apache.hudi.AvroConversionUtils;
import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.utilities.ingestion.HoodieIngestionMetrics;
@@ -33,6 +34,9 @@ import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.sql.AnalysisException;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
+import org.apache.spark.sql.RowFactory;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructType;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
@@ -40,6 +44,8 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
@@ -191,6 +197,32 @@ public class TestSqlSource extends UtilitiesTestBase {
assertEquals(0, fetch1AsRows.getBatch().get().count());
}
+ /**
+ * Runs the test scenario of reading data from a source that already carries
hoodie meta columns.
+ * All meta columns but the partition path are expected to be dropped from
the fetched dataset.
+ */
+ @Test
+ public void testSqlSourceDropsHoodieMetaColumns() {
+ StructType schema = new StructType();
+ for (String metaColumn : HoodieRecord.HOODIE_META_COLUMNS) {
+ schema = schema.add(metaColumn, DataTypes.StringType, true);
+ }
+ schema = schema.add("id", DataTypes.StringType, true);
+ Row row = RowFactory.create("001", "001_0_1", "key1", "2022/03/12",
"f1_1-0-1_001.parquet", "key1");
+ sparkSession.createDataFrame(Collections.singletonList(row), schema)
+ .createOrReplaceTempView("test_sql_meta_table");
+
+ props.setProperty(sqlSourceConfig, "select * from test_sql_meta_table");
+ sqlSource = new SqlSource(props, jsc, sparkSession, schemaProvider,
metrics);
+ sourceFormatAdapter = new SourceFormatAdapter(sqlSource);
+
+ Dataset<Row> fetchedRows =
+ sourceFormatAdapter.fetchNewDataInRowFormat(Option.empty(),
Long.MAX_VALUE).getBatch().get();
+ assertEquals(Arrays.asList(HoodieRecord.PARTITION_PATH_METADATA_FIELD,
"id"),
+ Arrays.asList(fetchedRows.columns()));
+ assertEquals(1, fetchedRows.count());
+ }
+
/**
* Runs the test scenario of reading data from the source in row format.
* Source table doesn't exists.
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestDeprecatedCloudIngestionConfigs.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestDeprecatedCloudIngestionConfigs.java
new file mode 100644
index 000000000000..53d4136d03ea
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestDeprecatedCloudIngestionConfigs.java
@@ -0,0 +1,79 @@
+/*
+ * 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.sources.helpers;
+
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.utilities.config.CloudSourceConfig;
+import org.apache.hudi.utilities.config.GCSEventsSourceConfig;
+import org.apache.hudi.utilities.deltastreamer.NoNewDataTerminationStrategy;
+import org.apache.hudi.utilities.sources.helpers.gcs.GcsIngestionConfig;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Guards the deprecated cloud-ingestion shims against drift from the {@link
CloudSourceConfig} /
+ * {@link GCSEventsSourceConfig} definitions they now delegate to.
+ */
+@SuppressWarnings("deprecation")
+class TestDeprecatedCloudIngestionConfigs {
+
+ @Test
+ void cloudStoreIngestionConfigDelegatesToCloudSourceConfig() {
+ // covers the implicit constructor of the deprecated constant holder
+ assertNotNull(new CloudStoreIngestionConfig());
+
+ assertEquals("hoodie.streamer.source.cloud.meta.batch.size",
CloudStoreIngestionConfig.BATCH_SIZE_CONF);
+ assertEquals(10, CloudStoreIngestionConfig.DEFAULT_BATCH_SIZE);
+ assertEquals("hoodie.streamer.source.cloud.meta.ack",
CloudStoreIngestionConfig.ACK_MESSAGES);
+ assertTrue(CloudStoreIngestionConfig.ACK_MESSAGES_DEFAULT_VALUE);
+ assertEquals("hoodie.streamer.source.cloud.data.check.file.exists",
CloudStoreIngestionConfig.ENABLE_EXISTS_CHECK);
+ assertFalse(CloudStoreIngestionConfig.DEFAULT_ENABLE_EXISTS_CHECK);
+ assertEquals("hoodie.streamer.source.cloud.data.select.relpath.prefix",
CloudStoreIngestionConfig.SELECT_RELATIVE_PATH_PREFIX);
+ assertEquals("hoodie.streamer.source.cloud.data.ignore.relpath.prefix",
CloudStoreIngestionConfig.IGNORE_RELATIVE_PATH_PREFIX);
+ assertEquals("hoodie.streamer.source.cloud.data.ignore.relpath.substring",
CloudStoreIngestionConfig.IGNORE_RELATIVE_PATH_SUBSTR);
+ assertEquals("hoodie.streamer.source.cloud.data.datasource.options",
CloudStoreIngestionConfig.SPARK_DATASOURCE_OPTIONS);
+ assertEquals("hoodie.streamer.source.cloud.data.select.file.extension",
CloudStoreIngestionConfig.CLOUD_DATAFILE_EXTENSION);
+ assertEquals("hoodie.streamer.source.cloud.data.datafile.format",
CloudStoreIngestionConfig.DATAFILE_FORMAT);
+ }
+
+ @Test
+ void gcsIngestionConfigDelegatesToGcsEventsSourceConfig() {
+ // covers the implicit constructor of the deprecated constant holder
+ assertNotNull(new GcsIngestionConfig());
+
+ assertEquals("hoodie.streamer.source.gcs.project.id",
GcsIngestionConfig.GOOGLE_PROJECT_ID);
+ assertEquals("hoodie.streamer.source.gcs.subscription.id",
GcsIngestionConfig.PUBSUB_SUBSCRIPTION_ID);
+ }
+
+ @Test
+ void deprecatedNoNewDataTerminationStrategyInheritsShutdownBehavior() {
+ NoNewDataTerminationStrategy strategy = new
NoNewDataTerminationStrategy(new TypedProperties());
+
+ // Default is 3 consecutive empty rounds before shutdown.
+ assertFalse(strategy.shouldShutdown(Option.empty()));
+ assertFalse(strategy.shouldShutdown(Option.empty()));
+ assertTrue(strategy.shouldShutdown(Option.empty()));
+ }
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestIncrSourceHelper.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestIncrSourceHelper.java
index cb4e6a8708ed..4432ad4c7c7c 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestIncrSourceHelper.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestIncrSourceHelper.java
@@ -128,6 +128,20 @@ class TestIncrSourceHelper extends
SparkClientFunctionalTestHarness {
assertTrue(!result.getRight().isPresent());
}
+ @Test
+ void testCloudObjectIncrCheckpointSerialization() {
+ // neither commit nor key: falls back to the sentinel start timestamp
+ assertEquals(IncrSourceHelper.DEFAULT_START_TIMESTAMP, new
CloudObjectIncrCheckpoint(null, null).toString());
+
+ CloudObjectIncrCheckpoint commitOnly = new
CloudObjectIncrCheckpoint("commit1", null);
+ assertEquals("commit1", commitOnly.toString());
+ assertEquals("commit1", commitOnly.getCommit());
+
+ CloudObjectIncrCheckpoint commitAndKey = new
CloudObjectIncrCheckpoint("commit1", "path/to/file1.json");
+ assertEquals("commit1#path/to/file1.json", commitAndKey.toString());
+ assertEquals("path/to/file1.json", commitAndKey.getKey());
+ }
+
@Test
void testSingleObjectExceedingSourceLimit() {
List<Triple<String, Long, String>> filePathSizeAndCommitTime = new
ArrayList<>();
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestQueryInfo.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestQueryInfo.java
new file mode 100644
index 000000000000..96d8e3b53336
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestQueryInfo.java
@@ -0,0 +1,109 @@
+/*
+ * 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.sources.helpers;
+
+import
org.apache.hudi.utilities.sources.SnapshotLoadQuerySplitter.CheckpointWithPredicates;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+
+import static
org.apache.hudi.DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL;
+import static
org.apache.hudi.DataSourceReadOptions.QUERY_TYPE_SNAPSHOT_OPT_VAL;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests the {@link QueryInfo} value object used by the cloud incremental
sources.
+ */
+class TestQueryInfo {
+
+ private static final String PREVIOUS_INSTANT = "20240101000000000";
+ private static final String START_INSTANT = "20240101010000000";
+ private static final String END_INSTANT = "20240101020000000";
+ private static final String ORDER_COLUMN = "_hoodie_commit_time";
+ private static final String KEY_COLUMN = "_hoodie_record_key";
+ private static final String LIMIT_COLUMN = "s3.object.size";
+
+ private static QueryInfo incrementalQueryInfo() {
+ return new QueryInfo(QUERY_TYPE_INCREMENTAL_OPT_VAL(), PREVIOUS_INSTANT,
START_INSTANT, END_INSTANT,
+ ORDER_COLUMN, KEY_COLUMN, LIMIT_COLUMN);
+ }
+
+ @Test
+ void incrementalQueryInfoHasNoPredicateFilter() {
+ QueryInfo queryInfo = incrementalQueryInfo();
+
+ assertEquals(QUERY_TYPE_INCREMENTAL_OPT_VAL(), queryInfo.getQueryType());
+ assertEquals(PREVIOUS_INSTANT, queryInfo.getPreviousInstant());
+ assertEquals(START_INSTANT, queryInfo.getStartInstant());
+ assertEquals(END_INSTANT, queryInfo.getEndInstant());
+ assertEquals(ORDER_COLUMN, queryInfo.getOrderColumn());
+ assertEquals(KEY_COLUMN, queryInfo.getKeyColumn());
+ assertEquals(LIMIT_COLUMN, queryInfo.getLimitColumn());
+ assertEquals(Arrays.asList(ORDER_COLUMN, KEY_COLUMN),
queryInfo.getOrderByColumns());
+
+ // The 7-arg ctor defaults the predicate filter to the empty string, which
reads back as absent.
+ assertFalse(queryInfo.getPredicateFilter().isPresent());
+ assertTrue(queryInfo.isIncremental());
+ assertFalse(queryInfo.isSnapshot());
+ assertFalse(queryInfo.areStartAndEndInstantsEqual());
+ }
+
+ @Test
+ void snapshotQueryInfoExposesPredicateFilter() {
+ QueryInfo queryInfo = new QueryInfo(QUERY_TYPE_SNAPSHOT_OPT_VAL(),
PREVIOUS_INSTANT, START_INSTANT, START_INSTANT,
+ "partition_path = 'a'", ORDER_COLUMN, KEY_COLUMN, LIMIT_COLUMN);
+
+ assertTrue(queryInfo.isSnapshot());
+ assertFalse(queryInfo.isIncremental());
+ assertTrue(queryInfo.areStartAndEndInstantsEqual());
+ assertEquals("partition_path = 'a'", queryInfo.getPredicateFilter().get());
+ }
+
+ @Test
+ void withUpdatedEndInstantMovesTheEndInstantAndDropsThePredicateFilter() {
+ QueryInfo queryInfo = new QueryInfo(QUERY_TYPE_INCREMENTAL_OPT_VAL(),
PREVIOUS_INSTANT, START_INSTANT, END_INSTANT,
+ "partition_path = 'a'", ORDER_COLUMN, KEY_COLUMN, LIMIT_COLUMN);
+ assertTrue(queryInfo.getPredicateFilter().isPresent());
+
+ QueryInfo updated = queryInfo.withUpdatedEndInstant("20240101030000000");
+
+ assertEquals("20240101030000000", updated.getEndInstant());
+ assertEquals(START_INSTANT, updated.getStartInstant());
+ assertEquals(PREVIOUS_INSTANT, updated.getPreviousInstant());
+ assertEquals(QUERY_TYPE_INCREMENTAL_OPT_VAL(), updated.getQueryType());
+ // withUpdatedEndInstant routes through the 7-arg ctor, so any predicate
filter is dropped
+ assertFalse(updated.getPredicateFilter().isPresent());
+ }
+
+ @Test
+ void withUpdatedCheckpointAppliesEndTimeAndPredicate() {
+ CheckpointWithPredicates checkpoint = new
CheckpointWithPredicates("20240101040000000", "partition_path > 'b'");
+ assertEquals("20240101040000000", checkpoint.getEndCompletionTime());
+ assertEquals("partition_path > 'b'", checkpoint.getPredicateFilter());
+
+ QueryInfo updated =
incrementalQueryInfo().withUpdatedCheckpoint(checkpoint);
+
+ assertEquals("20240101040000000", updated.getEndInstant());
+ assertEquals("partition_path > 'b'", updated.getPredicateFilter().get());
+ assertEquals(START_INSTANT, updated.getStartInstant());
+ }
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/gcs/TestPubsubQueueClient.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/gcs/TestPubsubQueueClient.java
index 86fe45002379..a625b40256b8 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/gcs/TestPubsubQueueClient.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/gcs/TestPubsubQueueClient.java
@@ -18,13 +18,22 @@
package org.apache.hudi.utilities.sources.helpers.gcs;
+import com.google.api.gax.rpc.UnaryCallable;
import com.google.cloud.ServiceOptions;
import com.google.cloud.monitoring.v3.MetricServiceClient;
+import com.google.cloud.pubsub.v1.stub.GrpcSubscriberStub;
+import com.google.cloud.pubsub.v1.stub.SubscriberStub;
+import com.google.cloud.pubsub.v1.stub.SubscriberStubSettings;
import com.google.monitoring.v3.ListTimeSeriesRequest;
import com.google.monitoring.v3.Point;
import com.google.monitoring.v3.TimeSeries;
import com.google.monitoring.v3.TypedValue;
+import com.google.protobuf.Empty;
import com.google.protobuf.util.Timestamps;
+import com.google.pubsub.v1.AcknowledgeRequest;
+import com.google.pubsub.v1.PullRequest;
+import com.google.pubsub.v1.PullResponse;
+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;
@@ -32,13 +41,16 @@ import org.mockito.ArgumentCaptor;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.stream.Stream;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
class TestPubsubQueueClient {
@@ -76,4 +88,45 @@ class TestPubsubQueueClient {
assertEquals("metric.type=\"pubsub.googleapis.com/subscription/num_undelivered_messages\"
AND resource.label.subscription_id=\"subscriptionId\"", request.getFilter());
}
}
+
+ @Test
+ void makePullRequest() {
+ SubscriberStub subscriber = mock(SubscriberStub.class);
+ UnaryCallable<PullRequest, PullResponse> pullCallable =
mock(UnaryCallable.class);
+ PullResponse expectedResponse = PullResponse.newBuilder().build();
+ ArgumentCaptor<PullRequest> requestCaptor =
ArgumentCaptor.forClass(PullRequest.class);
+ when(subscriber.pullCallable()).thenReturn(pullCallable);
+
when(pullCallable.call(requestCaptor.capture())).thenReturn(expectedResponse);
+
+ PullResponse actualResponse = new
PubsubQueueClient().makePullRequest(subscriber,
"projects/project/subscriptions/subscription", 10);
+
+ assertSame(expectedResponse, actualResponse);
+ assertEquals("projects/project/subscriptions/subscription",
requestCaptor.getValue().getSubscription());
+ assertEquals(10, requestCaptor.getValue().getMaxMessages());
+ }
+
+ @Test
+ void makeAckRequest() {
+ SubscriberStub subscriber = mock(SubscriberStub.class);
+ UnaryCallable<AcknowledgeRequest, Empty> acknowledgeCallable =
mock(UnaryCallable.class);
+ ArgumentCaptor<AcknowledgeRequest> requestCaptor =
ArgumentCaptor.forClass(AcknowledgeRequest.class);
+ when(subscriber.acknowledgeCallable()).thenReturn(acknowledgeCallable);
+
+ new PubsubQueueClient().makeAckRequest(subscriber,
"projects/project/subscriptions/subscription", Arrays.asList("ack1", "ack2"));
+
+ verify(acknowledgeCallable).call(requestCaptor.capture());
+ assertEquals("projects/project/subscriptions/subscription",
requestCaptor.getValue().getSubscription());
+ assertEquals(Arrays.asList("ack1", "ack2"),
requestCaptor.getValue().getAckIdsList());
+ }
+
+ @Test
+ void getSubscriber() throws Exception {
+ SubscriberStubSettings subscriberStubSettings =
mock(SubscriberStubSettings.class);
+ GrpcSubscriberStub expectedSubscriber = mock(GrpcSubscriberStub.class);
+ try (MockedStatic<GrpcSubscriberStub> mockedStaticSubscriberStub =
Mockito.mockStatic(GrpcSubscriberStub.class)) {
+ mockedStaticSubscriberStub.when(() ->
GrpcSubscriberStub.create(subscriberStubSettings)).thenReturn(expectedSubscriber);
+
+ assertSame(expectedSubscriber, new
PubsubQueueClient().getSubscriber(subscriberStubSettings));
+ }
+ }
}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/unstructured/TestDocumentParserType.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/unstructured/TestDocumentParserType.java
new file mode 100644
index 000000000000..54cd56e69805
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/unstructured/TestDocumentParserType.java
@@ -0,0 +1,81 @@
+/*
+ * 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.sources.helpers.unstructured;
+
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.utilities.config.UnstructuredFileSourceConfig;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Verifies how {@link DocumentParserType} resolves the configured parser name
to a class name.
+ */
+class TestDocumentParserType {
+
+ private static final String PARSER_KEY =
UnstructuredFileSourceConfig.DOCUMENT_PARSER.key();
+ private static final String PARSER_CLASS_KEY =
UnstructuredFileSourceConfig.PARSER_CLASS.key();
+
+ @Test
+ void tikaResolvesRegardlessOfCaseAndPadding() {
+ // Unset falls back to the TIKA default.
+ assertEquals(TikaDocumentParser.class.getName(),
+ DocumentParserType.resolveParserClass(new TypedProperties()));
+
+ TypedProperties props = new TypedProperties();
+ props.setProperty(PARSER_KEY, " tika ");
+ assertEquals(TikaDocumentParser.class.getName(),
DocumentParserType.resolveParserClass(props));
+ }
+
+ @Test
+ void unknownParserTypeThrows() {
+ TypedProperties props = new TypedProperties();
+ props.setProperty(PARSER_KEY, "docling");
+
+ HoodieException e = assertThrows(HoodieException.class, () ->
DocumentParserType.resolveParserClass(props));
+ assertTrue(e.getMessage().contains("Unknown " + PARSER_KEY),
e.getMessage());
+ assertTrue(e.getMessage().contains("docling"), e.getMessage());
+ }
+
+ @Test
+ void customWithoutParserClassThrows() {
+ TypedProperties props = new TypedProperties();
+ props.setProperty(PARSER_KEY, "CUSTOM");
+
+ HoodieException unset = assertThrows(HoodieException.class, () ->
DocumentParserType.resolveParserClass(props));
+ assertTrue(unset.getMessage().contains(PARSER_CLASS_KEY),
unset.getMessage());
+
+ props.setProperty(PARSER_CLASS_KEY, " ");
+ HoodieException blank = assertThrows(HoodieException.class, () ->
DocumentParserType.resolveParserClass(props));
+ assertTrue(blank.getMessage().contains(PARSER_CLASS_KEY),
blank.getMessage());
+ }
+
+ @Test
+ void customReturnsTrimmedParserClass() {
+ TypedProperties props = new TypedProperties();
+ props.setProperty(PARSER_KEY, " custom ");
+ props.setProperty(PARSER_CLASS_KEY, " com.example.MyDocumentParser ");
+
+ assertEquals("com.example.MyDocumentParser",
DocumentParserType.resolveParserClass(props));
+ }
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestConfigurationHotUpdateStrategyUtils.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestConfigurationHotUpdateStrategyUtils.java
new file mode 100644
index 000000000000..33139f708034
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestConfigurationHotUpdateStrategyUtils.java
@@ -0,0 +1,89 @@
+/*
+ * 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.streamer;
+
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieException;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests {@link ConfigurationHotUpdateStrategyUtils}.
+ */
+class TestConfigurationHotUpdateStrategyUtils {
+
+ private final HoodieStreamer.Config cfg = new HoodieStreamer.Config();
+ private final TypedProperties props = new TypedProperties();
+
+ @Test
+ void unsetStrategyClassYieldsEmptyOption() {
+ assertFalse(ConfigurationHotUpdateStrategyUtils
+ .createConfigurationHotUpdateStrategy(null, cfg, props).isPresent());
+ assertFalse(ConfigurationHotUpdateStrategyUtils
+ .createConfigurationHotUpdateStrategy("", cfg, props).isPresent());
+ }
+
+ @Test
+ void configuredStrategyClassIsInstantiatedReflectivelyWithCfgAndProps() {
+ Option<ConfigurationHotUpdateStrategy> strategy =
ConfigurationHotUpdateStrategyUtils
+
.createConfigurationHotUpdateStrategy(EchoHotUpdateStrategy.class.getName(),
cfg, props);
+
+ assertTrue(strategy.isPresent());
+ assertTrue(strategy.get() instanceof EchoHotUpdateStrategy);
+ // The reflective ctor call must have handed both args down to the base
class.
+ assertSame(cfg, ((EchoHotUpdateStrategy) strategy.get()).getCfg());
+ assertSame(props, strategy.get().updateProperties(new
TypedProperties()).get());
+ }
+
+ @Test
+ void unresolvableStrategyClassThrows() {
+ String bogusClass =
"org.apache.hudi.utilities.streamer.DoesNotExistHotUpdateStrategy";
+
+ HoodieException e = assertThrows(HoodieException.class, () ->
ConfigurationHotUpdateStrategyUtils
+ .createConfigurationHotUpdateStrategy(bogusClass, cfg, props));
+ assertTrue(e.getMessage().contains("Could not create configuration hot
update strategy class"), e.getMessage());
+ assertTrue(e.getMessage().contains(bogusClass), e.getMessage());
+ }
+
+ /**
+ * Strategy whose ctor signature matches exactly what {@code
ReflectionUtils.loadClass} infers from
+ * {@link HoodieStreamer.Config} and {@link TypedProperties}, and which
echoes back what it was given.
+ */
+ public static class EchoHotUpdateStrategy extends
ConfigurationHotUpdateStrategy {
+
+ public EchoHotUpdateStrategy(HoodieStreamer.Config cfg, TypedProperties
properties) {
+ super(cfg, properties);
+ }
+
+ HoodieStreamer.Config getCfg() {
+ return cfg;
+ }
+
+ @Override
+ public Option<TypedProperties> updateProperties(TypedProperties
currentProps) {
+ return Option.of(properties);
+ }
+ }
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestHoodieStreamerMetrics.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestHoodieStreamerMetrics.java
index 23ba3e50144d..1fa85ad8cfc2 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestHoodieStreamerMetrics.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestHoodieStreamerMetrics.java
@@ -21,6 +21,8 @@ package org.apache.hudi.utilities.streamer;
import org.apache.hudi.common.config.metrics.HoodieMetricsConfig;
import org.apache.hudi.common.util.HoodieStorageUtils;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamerMetrics;
import com.codahale.metrics.MetricRegistry;
import com.codahale.metrics.Timer;
@@ -115,4 +117,28 @@ public class TestHoodieStreamerMetrics {
metrics.emitStreamerJobFailedMetrics();
assertNull(metrics.getMetrics());
}
+
+ @Test
+ public void testDeprecatedDeltaStreamerMetricsAlias() {
+ HoodieMetricsConfig metricsConfig = HoodieMetricsConfig.newBuilder()
+ .on(true)
+ .withPath("/tmp/path6")
+ .withReporterType("INMEMORY")
+ .build();
+ HoodieDeltaStreamerMetrics metrics = new HoodieDeltaStreamerMetrics(
+ metricsConfig, HoodieStorageUtils.getStorage(getDefaultStorageConf()));
+ metrics.emitStreamerJobSuccessMetrics();
+ assertEquals(".deltastreamer.success",
metrics.getMetrics().getRegistry().getGauges().firstKey());
+
+ // the write config overload reports against the metrics config derived
from the write config
+ HoodieWriteConfig writeConfig = HoodieWriteConfig.newBuilder()
+ .withPath("/tmp/path7")
+
.withMetricsConfig(HoodieMetricsConfig.newBuilder().on(true).withReporterType("INMEMORY").build())
+ .build();
+ HoodieDeltaStreamerMetrics metricsFromWriteConfig = new
HoodieDeltaStreamerMetrics(
+ writeConfig, HoodieStorageUtils.getStorage(getDefaultStorageConf()));
+ metricsFromWriteConfig.emitStreamerJobFailedMetrics();
+ assertEquals(".deltastreamer.failure",
+
metricsFromWriteConfig.getMetrics().getRegistry().getGauges().firstKey());
+ }
}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestTerminationStrategyUtils.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestTerminationStrategyUtils.java
new file mode 100644
index 000000000000..9fedc760fc43
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestTerminationStrategyUtils.java
@@ -0,0 +1,62 @@
+/*
+ * 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.streamer;
+
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieException;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests {@link TerminationStrategyUtils}.
+ */
+class TestTerminationStrategyUtils {
+
+ private final TypedProperties props = new TypedProperties();
+
+ @Test
+ void unsetStrategyClassYieldsEmptyOption() {
+
assertFalse(TerminationStrategyUtils.createPostWriteTerminationStrategy(props,
null).isPresent());
+
assertFalse(TerminationStrategyUtils.createPostWriteTerminationStrategy(props,
"").isPresent());
+ }
+
+ @Test
+ void configuredStrategyClassIsInstantiatedReflectively() {
+ Option<PostWriteTerminationStrategy> strategy =
TerminationStrategyUtils.createPostWriteTerminationStrategy(
+ props, NoNewDataTerminationStrategy.class.getName());
+
+ assertTrue(strategy.isPresent());
+ assertTrue(strategy.get() instanceof NoNewDataTerminationStrategy);
+ }
+
+ @Test
+ void unresolvableStrategyClassThrows() {
+ String bogusClass =
"org.apache.hudi.utilities.streamer.DoesNotExistTerminationStrategy";
+
+ HoodieException e = assertThrows(HoodieException.class,
+ () ->
TerminationStrategyUtils.createPostWriteTerminationStrategy(props, bogusClass));
+ assertTrue(e.getMessage().contains("Could not create"), e.getMessage());
+ assertTrue(e.getMessage().contains(bogusClass), e.getMessage());
+ }
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/validator/TestSparkValidationContext.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/validator/TestSparkValidationContext.java
index 7f94262e98c4..d809a91db31c 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/validator/TestSparkValidationContext.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/validator/TestSparkValidationContext.java
@@ -21,7 +21,11 @@ package org.apache.hudi.utilities.streamer.validator;
import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.checkpoint.StreamerCheckpointV1;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.util.Option;
import org.junit.jupiter.api.Test;
@@ -32,7 +36,11 @@ import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
/**
* Tests for {@link SparkValidationContext}.
@@ -141,6 +149,40 @@ public class TestSparkValidationContext {
ctx.getPreviousCommitMetadata().get().getMetadata(StreamerCheckpointV1.STREAMER_CHECKPOINT_KEY_V1));
}
+ @Test
+ public void testTimelineAccessWithoutMetaClient() {
+ SparkValidationContext ctx = new SparkValidationContext(
+ "20260320120000000",
+ Option.of(new HoodieCommitMetadata()),
+ Option.of(Collections.emptyList()),
+ Option.empty());
+
+ assertThrows(UnsupportedOperationException.class, ctx::getActiveTimeline);
+ assertFalse(ctx.getPreviousCommitInstant().isPresent());
+ }
+
+ @Test
+ public void testTimelineAccessWithMetaClient() {
+ HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class);
+ HoodieActiveTimeline activeTimeline = mock(HoodieActiveTimeline.class);
+ HoodieTimeline writeTimeline = mock(HoodieTimeline.class);
+ HoodieInstant lastInstant = mock(HoodieInstant.class);
+ when(metaClient.getActiveTimeline()).thenReturn(activeTimeline);
+ when(activeTimeline.getWriteTimeline()).thenReturn(writeTimeline);
+ when(writeTimeline.filterCompletedInstants()).thenReturn(writeTimeline);
+ when(writeTimeline.lastInstant()).thenReturn(Option.of(lastInstant));
+
+ SparkValidationContext ctx = new SparkValidationContext(
+ "20260320120000000",
+ Option.of(new HoodieCommitMetadata()),
+ Option.of(Collections.emptyList()),
+ Option.empty(),
+ metaClient);
+
+ assertSame(activeTimeline, ctx.getActiveTimeline());
+ assertSame(lastInstant, ctx.getPreviousCommitInstant().get());
+ }
+
@Test
public void testEmptyWriteStats() {
SparkValidationContext ctx = new SparkValidationContext(
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/testutils/CloudObjectTestUtils.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/testutils/CloudObjectTestUtils.java
index 9a848463df6d..19bb49c13330 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/testutils/CloudObjectTestUtils.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/testutils/CloudObjectTestUtils.java
@@ -56,6 +56,8 @@ public class CloudObjectTestUtils {
String approximateNumberOfMessages = "0";
if (path != null) {
+ // glacierEventData is deliberately absent from
streamer-config/s3-metadata.avsc: a test that
+ // sees it as a column proves the schema was inferred from this payload,
not schema-provided
String body =
"{\n \"Type\" : \"Notification\",\n \"MessageId\" : \"1\",\n
\"TopicArn\" : \"arn:aws:sns:foo:123:"
+ "foo-bar\",\n \"Subject\" : \"Amazon S3 Notification\",\n
\"Message\" : \"{\\\"Records\\\":"
@@ -63,7 +65,8 @@ public class CloudObjectTestUtils {
+
"-west-2\\\",\\\"eventTime\\\":\\\"2021-07-27T09:05:36.755Z\\\",\\\"eventName\\\":\\\"ObjectCreated"
+
":Copy\\\",\\\"userIdentity\\\":{\\\"principalId\\\":\\\"AWS:test\\\"},\\\"requestParameters\\\":"
+
"{\\\"sourceIPAddress\\\":\\\"0.0.0.0\\\"},\\\"responseElements\\\":{\\\"x-amz-request-id\\\":\\\""
- +
"test\\\",\\\"x-amz-id-2\\\":\\\"foobar\\\"},\\\"s3\\\":{\\\"s3SchemaVersion\\\":\\\"1.0\\\",\\\""
+ +
"test\\\",\\\"x-amz-id-2\\\":\\\"foobar\\\"},\\\"glacierEventData\\\":{\\\"restoreEventData\\\":"
+ +
"{\\\"lifecycleRestoreStorageClass\\\":\\\"STANDARD\\\"}},\\\"s3\\\":{\\\"s3SchemaVersion\\\":\\\"1.0\\\",\\\""
+
"configurationId\\\":\\\"foobar\\\",\\\"bucket\\\":{\\\"name\\\":\\\""
+ path.getParent().toString().replace("hdfs://", "")
+
"\\\",\\\"ownerIdentity\\\":{\\\"principalId\\\":\\\"foo\\\"},\\\"arn\\\":\\\"arn:aws:s3:::foo\\\"}"
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/embedding/TestOpenAICompatibleEmbeddingProvider.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/embedding/TestOpenAICompatibleEmbeddingProvider.java
index 466738524796..36303b4a9f22 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/embedding/TestOpenAICompatibleEmbeddingProvider.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/embedding/TestOpenAICompatibleEmbeddingProvider.java
@@ -32,11 +32,13 @@ import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -55,6 +57,7 @@ public class TestOpenAICompatibleEmbeddingProvider {
private static volatile int failureStatus = 500;
private static volatile boolean abortConnection = false;
private static volatile int vectorsToReturn = -1; // -1 = one per input
+ private static volatile String retryAfterHeader = "1";
private static final AtomicReference<String> LAST_BODY = new
AtomicReference<>();
private static final AtomicReference<String> LAST_AUTH = new
AtomicReference<>();
@@ -72,7 +75,7 @@ public class TestOpenAICompatibleEmbeddingProvider {
exchange.close(); // no response -> client-side IOException
return;
}
- exchange.getResponseHeaders().add("Retry-After", "1");
+ exchange.getResponseHeaders().add("Retry-After", retryAfterHeader);
exchange.sendResponseHeaders(failureStatus, -1);
exchange.close();
return;
@@ -107,6 +110,7 @@ public class TestOpenAICompatibleEmbeddingProvider {
failureStatus = 500;
abortConnection = false;
vectorsToReturn = -1;
+ retryAfterHeader = "1";
}
private OpenAICompatibleEmbeddingProvider provider(String apiKeyEnv) {
@@ -178,4 +182,25 @@ public class TestOpenAICompatibleEmbeddingProvider {
assertThrows(HoodieException.class,
() -> provider(null).embed(Arrays.asList("one", "two")));
}
+
+ @Test
+ public void testUnparsableRetryAfterFallsBackToExponentialBackoff() {
+ // two 429s whose Retry-After cannot be parsed as seconds, then success
+ REMAINING_FAILURES.set(2);
+ failureStatus = 429;
+ retryAfterHeader = "abc";
+ long startMs = System.currentTimeMillis();
+ assertEquals(1, provider(null).embed(Arrays.asList("throttled")).size());
+ assertEquals(3, REQUEST_COUNT.get());
+ // the unparsable header is ignored in favour of exponential backoff:
1000ms then 2000ms.
+ // honoring "abc" as one second would only add up to 2000ms in total.
+ assertTrue(System.currentTimeMillis() - startMs >= 3000);
+ }
+
+ @Test
+ public void testProviderInitDefaultsToNoOp() {
+ // an implementation that needs no configuration inherits the interface's
no-op init
+ EmbeddingProvider inMemoryProvider = texts ->
Collections.singletonList(new float[] {1.0f});
+ assertDoesNotThrow(() -> inMemoryProvider.init(new TypedProperties()));
+ }
}
