Hey Avi, I have reproduced and found a solution. The issue is not MFA, it is the BASIC credential provider is not using the token: https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/AWSUtil.java#L181
If you want to supply AK/SK/Token then you will have to use another CredentialProviderType, below is an example using SYS_PROP. We could improve the Kinesis connector to detect the session token and construct a BasicSessionCredentials: https://github.com/aws/aws-sdk-java/blob/master/aws-java-sdk-core/src/main/java/com/amazonaws/auth/BasicSessionCredentials.java Properties systemProperties = System.getProperties(); systemProperties.setProperty("aws.accessKeyId", accessKey); systemProperties.setProperty("aws.secretKey", secretKey); systemProperties.setProperty("aws.sessionToken", seesionToken); Properties producerConfig = new Properties(); producerConfig.setProperty(AWSConfigConstants.AWS_REGION, REGION); producerConfig.setProperty(AWSConfigConstants.AWS_CREDENTIALS_PROVIDER, "SYS_PROP"); I will add this to the Jira also. Let me know if you have any issues. Thanks, Danny From: Avi Levi <a...@neosec.com> Date: Wednesday, 16 December 2020 at 08:09 To: Robert Metzger <rmetz...@apache.org> Cc: user <user@flink.apache.org> Subject: RE: [EXTERNAL] Connecting to kinesis with mfa CAUTION: This email originated from outside of the organization. Do not click links or open attachments unless you can confirm the sender and know the content is safe. Thanks Robert, I actually tried all of the above but got to the same unfortunate result On Wed, Dec 16, 2020 at 8:24 AM Robert Metzger <rmetz...@apache.org<mailto:rmetz...@apache.org>> wrote: Hey Avi, Maybe providing secret/access key + session token doesn't work, and you need to provide either one of them? https://docs.aws.amazon.com/credref/latest/refdocs/setting-global-aws_session_token.html I'll also ping some AWS contributors active in Flink to take a look at this. Best, Robert On Tue, Dec 15, 2020 at 10:07 AM Avi Levi <a...@neosec.com<mailto:a...@neosec.com>> wrote: Hi guys, we are struggling to connect to kinesis when mfa is activated. I did configured everything according to the documentation but still getting exception : val producerConfig = new Properties() producerConfig.put(AWSConfigConstants.AWS_REGION, awsRegion) producerConfig.put(AWSConfigConstants.AWS_ACCESS_KEY_ID, awsAccessKey) producerConfig.put(AWSConfigConstants.AWS_SECRET_ACCESS_KEY, awsSecretAccessKey) producerConfig.put(com.amazonaws.auth.profile.internal.ProfileKeyConstants.AWS_SESSION_TOKEN, awsSessionToken) with a very simple pipeline : val producer = new FlinkKinesisProducer(new SimpleStringSchema(), producerConfig) producer.setFailOnError(true) producer.setDefaultStream(outputStreamName) producer.setDefaultPartition("0") env.fromElements("a", "b", "c").addSink(producer) env.execute() the results with: 15:30:44,292 WARN org.apache.flink.kinesis.shaded.com.amazonaws.services.kinesis.producer.LogInputStreamReader - [2020-12-14 15:30:44.292188] [0x0000cb5f][0x000070000512c000] [warning] [AWS Log: WARN](AWSClient)If the signature check failed. This could be because of a time skew. Attempting to adjust the signer. 15:30:44,378 INFO org.apache.flink.kinesis.shaded.com.amazonaws.services.kinesis.producer.LogInputStreamReader - [2020-12-14 15:30:44.377865] [0x0000cb5b][0x00007000082c1000] [info] [shard_map.cc:87] Updating shard map for stream "ExampleOutputStream" 15:30:44,396 WARN org.apache.flink.kinesis.shaded.com.amazonaws.services.kinesis.producer.LogInputStreamReader - [2020-12-14 15:30:44.396208] [0x0000cb55][0x0000700002a3e000] [warning] [AWS Log: WARN](AWSErrorMarshaller)Encountered AWSError 'UnrecognizedClientException': The security token included in the request is invalid. 15:30:44,396 ERROR org.apache.flink.kinesis.shaded.com.amazonaws.services.kinesis.producer.LogInputStreamReader - [2020-12-14 15:30:44.396256] [0x0000cb55][0x0000700002a3e000] [error] [AWS Log: ERROR](AWSClient)HTTP response code: 400 Exception name: UnrecognizedClientException Error message: The security token included in the request is invalid. 6 response headers: connection : close I double check that all keys are correct using the same keys that work perfectly when I execute commands from the cli. also when removing the mfa from kinesis the pipeline works as expected. finally i did open a ticket<https://issues.apache.org/jira/browse/FLINK-20602> for that also .
