本地测试中如何为Flink的flink-s3-fs-hadoop插件传递AWS配置？

问题根源 #

flink-s3-fs-hadoop底层依赖Hadoop S3A文件系统客户端，而非直接使用AWS SDK原生客户端，因此仅配置Flink参数无法生效，必须针对Hadoop的S3A配置项进行设置。

可行解决方案 #

方案一：测试代码中直接注入Hadoop配置 #

在启动MiniCluster前，显式设置Hadoop Configuration的S3A参数，确保底层客户端能获取到LocalStack的连接信息：

import org.apache.hadoop.conf.Configuration;
import org.apache.flink.configuration.Configuration as FlinkConf;
import org.apache.flink.runtime.minicluster.MiniCluster;
import org.apache.flink.runtime.minicluster.MiniClusterResourceConfiguration;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;

public class FlinkS3ParquetTest {
    private static MiniCluster miniCluster;

    @BeforeAll
    static void setupMiniCluster() throws Exception {
        // 初始化Hadoop配置，设置S3A核心参数
        Configuration hadoopConf = new Configuration();
        hadoopConf.set("fs.s3a.access.key", "test");
        hadoopConf.set("fs.s3a.secret.key", "test");
        hadoopConf.set("fs.s3a.endpoint", "http://localhost:4566"); // LocalStack默认端口
        hadoopConf.set("fs.s3a.region", "us-east-1");
        hadoopConf.set("fs.s3a.path.style.access", "true");
        hadoopConf.set("fs.s3a.connection.ssl.enabled", "false"); // LocalStack默认关闭SSL

        // 将Hadoop配置传递给Flink
        FlinkConf flinkConf = new FlinkConf();
        flinkConf.setString("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem");
        // 把Hadoop配置导入Flink配置
        hadoopConf.forEach(entry -> flinkConf.setString(entry.getKey(), entry.getValue()));

        // 启动MiniCluster
        miniCluster = MiniClusterFactory.createMiniCluster(flinkConf,
                new MiniClusterResourceConfiguration.Builder()
                        .setNumberSlotsPerTaskManager(2)
                        .setNumberTaskManagers(1)
                        .build());
        miniCluster.start();
    }

    // 测试逻辑...
}

方案二：通过测试资源目录的Hadoop配置文件注入 #

在src/test/resources下创建core-site.xml，配置S3A参数：

<?xml version="1.0" encoding="UTF-8"?>
<configuration>
    <property>
        <name>fs.s3a.access.key</name>
        <value>test</value>
    </property>
    <property>
        <name>fs.s3a.secret.key</name>
        <value>test</value>
    </property>
    <property>
        <name>fs.s3a.endpoint</name>
        <value>http://localhost:4566</value>
    </property>
    <property>
        <name>fs.s3a.region</name>
        <value>us-east-1</value>
    </property>
    <property>
        <name>fs.s3a.path.style.access</name>
        <value>true</value>
    </property>
    <property>
        <name>fs.s3a.connection.ssl.enabled</name>
        <value>false</value>
    </property>
</configuration>

然后在测试启动前指定Hadoop配置目录：

@BeforeAll
static void setupHadoopConf() {
    // 指向测试资源目录作为Hadoop配置源
    System.setProperty("HADOOP_CONF_DIR", "src/test/resources");
}

方案三：结合Testcontainers自动获取LocalStack配置 #

如果使用Testcontainers管理LocalStack实例，可直接通过容器对象获取动态生成的密钥和端点，避免硬编码：

import org.testcontainers.containers.LocalStackContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.utility.DockerImageName;

@Testcontainers
public class FlinkS3TestWithTestcontainers {
    @Container
    private static final LocalStackContainer localStack = new LocalStackContainer(DockerImageName.parse("localstack/localstack:latest"))
            .withServices(LocalStackContainer.Service.S3);

    @BeforeAll
    static void setupS3Config() {
        Configuration hadoopConf = new Configuration();
        hadoopConf.set("fs.s3a.access.key", localStack.getAccessKey());
        hadoopConf.set("fs.s3a.secret.key", localStack.getSecretKey());
        hadoopConf.set("fs.s3a.endpoint", localStack.getEndpointOverride(LocalStackContainer.Service.S3).toString());
        hadoopConf.set("fs.s3a.region", localStack.getRegion());
        hadoopConf.set("fs.s3a.path.style.access", "true");
        hadoopConf.set("fs.s3a.connection.ssl.enabled", "false");

        // 后续传递配置给MiniCluster...
    }
}

为什么之前的尝试未生效 #

• 环境变量方式：Hadoop S3A客户端不会自动映射AWS_ACCESS_KEY_ID等环境变量到自身配置项，仅部分参数能被识别。
• flink-conf.yaml：MiniCluster默认不会自动加载外部配置文件，且即使加载，S3相关参数需要设置到Hadoop配置而非Flink配置。
• 仅设置Flink执行环境配置：未同步修改底层Hadoop的Configuration，导致S3客户端无法获取正确的连接参数。

内容的提问来源于stack exchange，提问作者r_g_s_