Apache Flink - s3文件夹监控-许多文件丢失

<dependencies>
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-s3_2.11</artifactId>
    <version>${flink.version}</version>
  </dependency>
  <dependency>
    <groupId>com.amazonaws</groupId>
    <artifactId>aws-java-sdk-s3</artifactId>
    <version>${aws.sdk.version}</version>
  </dependency>
</dependencies>

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.s3.FlinkS3FileSystem;
import org.apache.flink.streaming.connectors.s3.S3StreamInputFormat;

public class S3Monitor {

    public static void main(String[] args) throws Exception {
        // 设置 Flink 执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 配置 S3 连接参数
        env.getConfig().setGlobalJobParameters(new Configuration());
        Configuration config = new Configuration();
        config.setString("fs.s3a.access.key", "your-access-key");
        config.setString("fs.s3a.secret.key", "your-secret-key");
        config.setString("fs.s3a.endpoint", "s3.amazonaws.com");

        // 创建 S3 数据流
        DataStream<String> s3Stream = env.createInput(
                new S3StreamInputFormat<>(
                        new Path("s3://your-bucket/your-folder"),
                        new SimpleStringSchema()
                )
        );

        // 处理 S3 文件流
        s3Stream.map(new MapFunction<String, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> map(String value) throws Exception {
                // 解析 S3 文件内容并处理丢失文件
                return new Tuple2<>(fileName, fileContent);
            }
        }).print();

        // 执行 Flink 程序
        env.execute("S3 Monitor");
    }
}