如何在写入S3的Apache Flink应用程序中为StreamingFileSink配置KMS加密

如何解决如何在写入S3的Apache Flink应用程序中为StreamingFileSink配置KMS加密

我有一个在Kinesis Data Analytics托管环境上运行的Flink 1.8.2应用程序。该应用程序从Kinesis Data Stream读取数据，并将聚合的数据写入S3。我收到“访问被拒绝”异常。已验证IAM角色权限。似乎还不错-它对所涉及的S3存储桶和KMS密钥具有必要的权限。

存储桶策略强制使用PutObject API上传的数据使用KMS密钥加密。如何在Flink应用程序中配置StreamingFileSink连接器以设置KMS加密？

StreamingFileSink代码段

private static StreamingFileSink<String> createS3SinkFromStaticConfigtest() {
        return StreamingFileSink
                .forRowFormat(new Path(s3SinkPath),new SimpleStringEncoder<String>("UTF-8"))
                .withRollingPolicy(
                        DefaultRollingPolicy.create()
                                .withRolloverInterval(TimeUnit.MINUTES.toMillis(1))
                                .withInactivityInterval(TimeUnit.MINUTES.toMillis(1))
                                .withMaxPartSize(1024 * 1024)
                                .build()
                )
                .build();
    }

错误：

    "locationinformation": "org.apache.flink.runtime.executiongraph.ExecutionGraph.transitionState(ExecutionGraph.java:1497)","logger": "org.apache.flink.runtime.executiongraph.ExecutionGraph","message": "Job Flink S3 Streaming Sink Job (f8901746927663ecb23b562ed4d85e37) switched from state RUNNING to FAILING.","throwableinformation": [ "java.nio.file.AccessDeniedException: app/flink-data/2020-09-11--15/part-0-0: initiate MultiPartUpload on app/flink-data/2020-09-11--15/part-0-0: org.apache.flink.fs.s3base.shaded.com.amazonaws.services.s3.model.AmazonS3Exception: Access Denied (Service: Amazon S3; Status Code: 403; Error Code: AccessDenied