karthitect commented on code in PR #52:
URL:
https://github.com/apache/flink-connector-aws/pull/52#discussion_r1108970489
##########
flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/table/KinesisDynamicSource.java:
##########
@@ -219,4 +245,29 @@ public int hashCode() {
consumerProperties,
decodingFormat);
}
+
+ private KinesisShardAssigner getShardAssigner(String
shardAssignerIdentifier) {
+ try {
+ ServiceLoader<KinesisDynamicShardAssignerFactory> loader =
+
ServiceLoader.load(KinesisDynamicShardAssignerFactory.class);
+ for (KinesisDynamicShardAssignerFactory factory : loader) {
+ if
(factory.shardAssignerIdentifer().equals(shardAssignerIdentifier)) {
+ return factory.getShardAssigner();
+ }
+ }
+ } catch (Exception e) {
Review Comment:
There is some discussion about this here:
https://github.com/eclipse/jetty.project/issues/4340
There's quite a bit of complexity to handling the ServiceConfigurationError
in the iterator. I'm not sure that's warranted here. But I took a crack at it.
Note that I had to use `while(true)` because it's the iterator `next/hasNext`
calls that [throw
exceptions](https://docs.oracle.com/en/java/javase/16/docs/api/java.base/java/util/ServiceLoader.html#load(java.lang.Class))
(so I can't use a for loop directly).
--
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]
