[HUDI-8811] Handle Kafka source configuration-related failures with specific exception handling #12569
Conversation
vamsikarnika
changed the title
| @@ -75,6 +77,8 @@ 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 (ConfigException ex) { | |||
| throw new HoodieSourceConfigException("kafka source config issue" + ex.getMessage()); | |||
There was a problem hiding this comment.
For these errors, let's pass in the source exception so we can see the context.
Also what kind of errors will this catch? Adding the specifics to the PR description will be helpful for the reviewer.
There was a problem hiding this comment.
This catches errors related to kafka configs like invalid kafka host url, invalid port, unresolvable broker addresses(all provided brokers are unresolvable).
Also in some cases missing required configs. For example when io.confluent.kafka.serializers.KafkaAvroDeserializer is used as kafka deserializer and schema.registry.url config is missing in the provided configs, KafkaConsumer throws io.confluent.common.config.ConfigException during initialization. We're catching those exceptions and categorizing them as HoodieSourceConfigExceptions here.
vamsikarnika
changed the title
Change Logs
Catch KafkaException instances and handle configuration-related errors by wrapping them in a HoodieSourceConfigException. Specifically, handle exceptions such as missing host or port, invalid or unresolved broker addresses, invalid host URLs.
Also in some cases missing required configs. For example when io.confluent.kafka.serializers.KafkaAvroDeserializer(default deserializer) is used as kafka deserializer and schema.registry.url config is missing in the provided configs, KafkaConsumer throws io.confluent.common.config.ConfigException during initialization. We're catching those exceptions and categorising them as HoodieSourceConfigExceptions here.
Impact
Add new exception type for HoodieSourceConfigException to catch issues specific to kafka configs
Risk level (write none, low medium or high below)
low
Documentation Update
NA
Contributor's checklist