1. I have 2 AWS accounts, Account A and Account B.
2. Account A has a EKS cluster running with a flink cluster running on it. To manage the IAM roles, we use Kube2iam.
3. All the pods on cluster have specific roles assigned to them. For simplicity lets say the role for one of the pods is Pod-Role
4. The K8s worker nodes have the role Worker-Node-role
5. Kube2iam is correctly configured to make proper EC2 metadata calls when required.
6. Account B has a S3 bucket, which the Pod hosted in Account A worked node need to read.

Possible Solution:

1. Create a role in Account B, let's say, AccountB_Bucket_access_role with a policy that allows reading the bucket. Add Pod-Role as a trusted entity to it.
2. Add a policy in Pod-role which allows switching to AccountB_Bucket_access_role, basically the STS AssumeRole action.
3. Create a AWS profile in Pod, let's say, custom_profile, with role_arn set to AccountB_Bucket_access_role role's arn.
4. While deploying the flink pod, set AWS_PROFILE=AccountB_Bucket_access_role.

QUESTION: Given above whenever the flink app needs to talk to S3 bucket, it first assumes the AccountB_Bucket_access_role role and is able to read the S3 bucket. But setting AWS_PROFILE actually switches the role for flink app, hence all the POD-ROLE permissions are lost, and they are required for proper functioning of flink app.

Is there a way, that this AWS custom_profile could only be used when reading S3 bucket and it switches to POD-ROLE after that.

val flinkEnv: StreamExecutionEnvironment = AppUtils.setUpAndGetFlinkEnvRef(config.flink)
val textInputFormat = new TextInputFormat(new Path(config.path))
env
 .readFile(
    textInputFormat,
    config.path,
    FileProcessingMode.PROCESS_CONTINUOUSLY,
    config.refreshDurationMs
 )

This is what I use in flink job to read S3 file.