wombatu-kun commented on code in PR #19416:
URL: https://github.com/apache/hudi/pull/19416#discussion_r3687901455
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/callback/TestKafkaCallbackProvider.java:
##########
@@ -72,20 +84,58 @@ public static void cleanupClass() throws IOException {
@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);
+ assertDoesNotThrow(() -> defaultRoutedCallback.call(new
HoodieWriteCommitCallbackMessage(
+ makeNewCommitTime(), defaultRoutedConfig.getTableName(),
defaultRoutedConfig.getBasePath(), stats)));
+
+ // an explicit partition config overrides the key hashing
+ HoodieWriteConfig pinnedConfig = createConfigForKafkaCallback("1");
+ HoodieWriteCommitCallback pinnedCallback =
HoodieCommitCallbackFactory.create(pinnedConfig);
+ assertDoesNotThrow(() -> pinnedCallback.call(new
HoodieWriteCommitCallbackMessage(
Review Comment:
HoodieWriteCommitKafkaCallback.call() catches Exception and only logs, so
both assertDoesNotThrow wrappers here pass even when the send fails outright.
The partition assertion is the only one that can fail - consider dropping the
wrappers or asserting the consumed message bodies instead.
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/schema/TestRowBasedSchemaProvider.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.hudi.common.config.TypedProperties;
+
+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.assertDoesNotThrow;
+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());
+ }
+
+ @Test
+ void testPropsBasedConstructor() {
+ // the (props, jssc) constructor is the signature HoodieStreamer resolves
reflectively
Review Comment:
RowBasedSchemaProvider is only ever built through
UtilHelpers.createRowBasedSchemaProvider from RowSource, so nothing resolves
the (props, jssc) ctor reflectively and this comment describes a path that does
not exist. That instance also leaves rowStruct null, so getSourceSchema()
throws on it - assert that instead of assertDoesNotThrow, or drop the test.
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestS3EventsSource.java:
##########
@@ -103,9 +116,56 @@ public void testReadingFromSource(boolean
persistSourceRdd) throws IOException {
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
+ assertEquals(Arrays.asList("awsRegion", "eventName", "eventSource",
"eventTime", "eventVersion",
Review Comment:
This column list is field-for-field identical to
streamer-config/s3-metadata.avsc, so every assertion here holds unchanged on
the schema-provider path and none of them pins the inference branch. Is it
meant to prove inference or just to execute it - if the former, the
CloudObjectTestUtils fixture needs a field the avsc does not declare.
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/schema/TestFilebasedSchemaProvider.java:
##########
@@ -97,4 +102,65 @@ void testJsonSchema() throws IOException {
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();
Review Comment:
This test always sets TARGET_SCHEMA_FILE, so the case where it is unset
stays uncovered: the ctor leaves targetSchema null while refresh() populates it
from the source file, which flips the branch getTargetSchema() takes. Worth a
second assertion with no target file configured - follow-up, not a blocker.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
