hlteoh37 commented on code in PR #76:
URL:
https://github.com/apache/flink-connector-aws/pull/76#discussion_r1825693268
##########
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>>
+ getPollingKinesisShardSplitReaderSupplier(
+ Configuration sourceConfig,
+ Map<String, KinesisShardMetrics> shardMetricGroupMap) {
+ // We create a new stream proxy for each split reader since they have
their own independent
+ // lifecycle.
+ return () ->
+ new PollingKinesisShardSplitReader(
+ createKinesisStreamProxy(sourceConfig),
shardMetricGroupMap);
+ }
+
+ private Supplier<SplitReader<Record, KinesisShardSplit>>
+ getFanOutKinesisShardSplitReaderSupplier(
+ Configuration sourceConfig,
+ Map<String, KinesisShardMetrics> shardMetricGroupMap) {
+ String consumerArn = getConsumerArn(streamArn,
sourceConfig.get(EFO_CONSUMER_NAME));
+
+ // We create a new stream proxy for each split reader since they have
their own independent
+ // lifecycle.
+ return () ->
+ new FanOutKinesisShardSplitReader(
+ createKinesisAsyncStreamProxy(sourceConfig),
+ consumerArn,
+ shardMetricGroupMap);
+ }
+
+ private String getConsumerArn(final String streamArn, final String
consumerName) {
+ try (StreamProxy streamProxy = createKinesisStreamProxy(sourceConfig))
{
+ DescribeStreamConsumerResponse response =
+ streamProxy.describeStreamConsumer(streamArn,
consumerName);
+ return response.consumerDescription().consumerARN();
+ } catch (IOException e) {
+ throw new KinesisStreamsSourceException(
+ "Unable to lookup consumer ARN for stream "
+ + streamArn
+ + " and consumer "
+ + consumerName,
+ e);
+ }
+ }
+
+ private KinesisAsyncStreamProxy
createKinesisAsyncStreamProxy(Configuration consumerConfig) {
+ SdkAsyncHttpClient asyncHttpClient =
Review Comment:
Rejiggled to make this close to the end
--
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
