hlteoh37 commented on code in PR #76:
URL:
https://github.com/apache/flink-connector-aws/pull/76#discussion_r1825626808
##########
flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/KinesisStreamsSource.java:
##########
@@ -178,6 +185,84 @@ public SimpleVersionedSerializer<KinesisShardSplit>
getSplitSerializer() {
return new KinesisStreamsSourceEnumeratorStateSerializer(new
KinesisShardSplitSerializer());
}
+ private Supplier<SplitReader<Record, KinesisShardSplit>>
getKinesisShardSplitReaderSupplier(
+ Configuration sourceConfig, Map<String, KinesisShardMetrics>
shardMetricGroupMap) {
+ KinesisStreamsSourceConfigConstants.ReaderType readerType =
sourceConfig.get(READER_TYPE);
+ switch (readerType) {
+ case POLLING:
+ return getPollingKinesisShardSplitReaderSupplier(sourceConfig,
shardMetricGroupMap);
+ case EFO:
+ return getFanOutKinesisShardSplitReaderSupplier(sourceConfig,
shardMetricGroupMap);
+ default:
+ throw new IllegalArgumentException("Unsupported reader type: "
+ readerType);
+ }
+ }
+
+ private Supplier<SplitReader<Record, KinesisShardSplit>>
Review Comment:
Done
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
