oakad opened a new issue, #23206:
URL: https://github.com/apache/beam/issues/23206
### What happened?
Beam v2.41.0 under Flink 1.13.
Possibly related to #22707.
`org.apache.beam.sdk.io.aws2.kinesis.KinesisIO` consumer has several issues
with its current implementation.
1. First, it's not handling the pipeline level `AwsOptions` correctly. Let's
say a pipeline is configured to use a profile to authenticate:
```
options.setAwsCredentialsProvider(
ProfileCredentialsProvider.create("some_profile")
);
Pipeline.create(options).apply(
"KinesisRead",
KinesisIO.read().withStreamName("stream_name")
);
```
When in "running" stage, the created pipeline step will use the "default'
credentials from the local credentials file or whatever other means to
authenticate, default AWS client factory can come with. It will not use
`some_profile` set in the `options`, even though debug printouts can be used to
confirm that `ProfileCredentialsProvider` in the `options` bean is correctly
initialized.
To make the situation worse, the `options` will be correctly used during
initialization, so the issue will be postponed until after the job start and
made even more cryptic.
2. The not unexpected outcome from the above issue is
`ResourceNotFoundException` from the various AWS SDK calls (if we are hitting
the wrong account, the expected stream will not be there). However, current
implementation of Kinesis consumer swallows those exceptions and proceeds as if
no error had happened, eventually failing with cryptic "NullPointerException"
errors (because consumer iterators can not be read/updated properly), like
shown below:
>> Caused by: java.lang.NullPointerException
at
org.apache.beam.sdk.io.aws2.kinesis.ShardReadersPool.getMinTimestamp(ShardReadersPool.java:257)
at
org.apache.beam.sdk.io.aws2.kinesis.ShardReadersPool.getWatermark(ShardReadersPool.java:249)
at
org.apache.beam.sdk.io.aws2.kinesis.KinesisReader.getWatermark(KinesisReader.java:133)
at
org.apache.beam.runners.flink.translation.wrappers.streaming.io.UnboundedSourceWrapper.onProcessingTime(UnboundedSourceWrapper.java:460)
at
org.apache.flink.streaming.runtime.tasks.StreamTask.invokeProcessingTimeCallback(StreamTask.java:1425)
### Issue Priority
Priority: 2
### Issue Component
Component: io-java-kinesis
--
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]
