yihua commented on code in PR #12569:
URL: https://github.com/apache/hudi/pull/12569#discussion_r1908046879
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestAvroKafkaSource.java:
##########
@@ -153,6 +155,30 @@ private Properties getProducerProperties() {
return props;
}
+ @Test
+ void testKafkaSource_InvalidHostException() throws IOException {
+ UtilitiesTestBase.Helpers.saveStringsToDFS(
+ new String[] {dataGen.generateGenericRecord().getSchema().toString()},
hoodieStorage(),
+ SCHEMA_PATH);
+ final String topic = TEST_TOPIC_PREFIX + "testKafkaOffsetAppend";
+ TypedProperties props = createPropsForKafkaSource(topic, null, "earliest");
+
+ props.put("hoodie.deltastreamer.schemaprovider.source.schema.file",
SCHEMA_PATH);
Review Comment:
```suggestion
props.put("hoodie.streamer.schemaprovider.source.schema.file",
SCHEMA_PATH);
```
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/KafkaSource.java:
##########
@@ -75,6 +78,11 @@ protected InputBatch<T>
readFromCheckpoint(Option<Checkpoint> lastCheckpoint, lo
lastCheckpoint, sourceLimit));
} catch (org.apache.kafka.common.errors.TimeoutException e) {
throw new HoodieSourceTimeoutException("Kafka Source timed out " +
e.getMessage());
+ } catch (KafkaException ex) {
+ if (hasConfigException(ex)) {
+ throw new HoodieSourceConfigException("kafka source config issue ",
ex);
+ }
+ throw ex;
Review Comment:
#8638 already introduced `HoodieReadFromSourceException` and
`HiveIncrPullSource#readFromCheckpoint` throws `HoodieReadFromSourceException`.
Could `HoodieReadFromSourceException` be reused? Is a new exception required?
--
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]
