XComp commented on code in PR #24:
URL:
https://github.com/apache/flink-connector-pulsar/pull/24#discussion_r1102601255
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarPartitionSplitReader.java:
##########
@@ -231,7 +232,11 @@ public void
handleSplitsChanges(SplitsChange<PulsarPartitionSplit> splitsChanges
}
// Create pulsar consumer.
- this.pulsarConsumer =
createPulsarConsumer(registeredSplit.getPartition());
+ try {
+ this.pulsarConsumer =
createPulsarConsumer(registeredSplit.getPartition());
+ } catch (PulsarClientException e) {
Review Comment:
It's odd that we have to transform the `PulsarClientException` here. I see
two reasons:
* Either we should expose exceptions in SplitReader.handleSplitsChanges` on
the Flink side
* Or we're not implementing the interface properly on the Pulsar connector
side
I'm curious what your opinion is on that one because I'm not that familiar
with the SDK part of the code
--
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]
