Programmatically configuring S3 options in Flink

user_483876 0
Viewed 528

Apparently Flink 1.14.0 doesn't correctly translate S3 options when they are set programmatically. I'm creating a local environment like this to connect to local MinIO instance:

  val flinkConf = new Configuration()
  flinkConf.setString("s3.endpoint", "http://127.0.0.1:9000")
  flinkConf.setString("s3.aws.credentials.provider", "org.apache.hadoop.fs.s3a.AnonymousAWSCredentialsProvider")

  val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(flinkConf)

Then StreamingFileSink fails with a huge stack trace with most relevant messages being Caused by: org.apache.hadoop.fs.s3a.auth.NoAuthWithAWSException: No AWS Credentials provided by SimpleAWSCredentialsProvider EnvironmentVariableCredentialsProvider InstanceProfileCredentialsProvider : com.amazonaws.SdkClientException: Failed to connect to service endpoint: which means that Hadoop tried to enumerate all of the credential providers instead of using the one set in configuration. What am I doing wrong?

flink-streaming apache-flink hadoop amazon-s3
1 Answers

I've spent ages trying to figure out this one too. I could not find a way to set it programmatically, but adding the following to src/main/resources/core-site.xml in my Flink java project root worked in the end:

<?xml version="1.0"?>
<configuration>
    <property>
        <name>fs.s3a.aws.credentials.provider</name>
        <value>com.amazonaws.auth.profile.ProfileCredentialsProvider</value>
    </property>
</configuration>

Then I could use AWS_PROFILE env var to select stored credentials. This was for Flink with flink-s3-fs-hadoop 1.13.2

user_46489690
Related