voonhous commented on code in PR #19416:
URL: https://github.com/apache/hudi/pull/19416#discussion_r3689401091
##########
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(
+ makeNewCommitTime(), pinnedConfig.getTableName(),
pinnedConfig.getBasePath(), stats)));
+
+ List<ConsumerRecord<String, String>> consumed =
consumeCallbackMessages(numPartitions, 2);
+ // hashing the table name key routes to partition 0, so partition 1 can
only come from the config
+ assertEquals(Arrays.asList(0, 1),
+
consumed.stream().map(ConsumerRecord::partition).sorted().collect(Collectors.toList()));
+ }
+
+ 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);
Review Comment:
Done -- hoisted into `POLL_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(60)`.
##########
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:
Good catch -- dropped the wrappers. The test now asserts each partition's
consumed body carries the commit time that produced it, so a swallowed send
failure fails the test.
##########
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:
Added `testRefreshWithoutTargetSchemaFileConfigured` covering the fallback
before `refresh()` and the populated branch after.
--
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]
