教程:将自定义信任商店与亚马逊 MSK 结合使用 - Amazon Kinesis Data Analytics
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅中国的 Amazon Web Services 服务入门

本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。

教程:将自定义信任商店与亚马逊 MSK 结合使用

以下教程演示了如何安全地连接(传输中的加密)到使用自定义证书、私有甚至自托管证书颁发机构 (CA) 颁发的服务器证书的 Kafka Cluster。

要通过 TLS 将任何 Kafka 客户端安全地连接到 Kafka 集群,Kafka 客户端(如示例 Flink 应用程序)必须信任 Kafka 集群的服务器证书(从颁发 CA 到根级 CA)提供的完整信任链。作为自定义信任商店的示例,我们将使用启用了双向 TLS (MTLS) 身份验证的 Amazon MSK 集群。这意味着 MSK 群集节点使用由AmazonCertificate Manager 私有证书颁发机构 (ACM Private CA),它对您的账户和区域是私有的,因此不受执行 Flink 应用程序的 Java 虚拟机 (JVM) 默认信任库的信任库的信任。

注意
  • 一个密钥库用于存储应用程序应向服务器或客户端出示的私钥和身份证书以供验证。

  • 一个信任商店用于存储来自认证颁发机构 (CA) 的证书,用于验证服务器在 SSL 连接中提供的证书。

您还可以使用本教程中的技术进行 Kinesis Data Analytics 应用程序与其他 Apache Kafka 源之间的交互,例如:

您的应用程序将使用自定义消费者(CustomFlinkKafkaConsumer) 这将覆盖open方法来加载自定义信任库。这使得信任库在应用程序重新启动或替换线程后可供应用程序使用。

使用以下代码检索和存储自定义信任库,从CustomFlinkKafkaConsumer.javafile:

@Override public void open(Configuration configuration) throws Exception { // write truststore to /tmp // NOTE: make sure that truststore is in JKS format for KDA/Flink. See README for details dropFile("/tmp"); super.open(configuration); } private void dropFile(String destFolder) throws Exception { ClassLoader classLoader = Thread.currentThread().getContextClassLoader(); InputStream input = classLoader.getResourceAsStream("kafka.client.truststore.jks"); byte[] buffer = new byte[input.available()]; input.read(buffer); File destDir = new File(destFolder); File targetFile = new File(destDir, "kafka.client.truststore.jks"); OutputStream outStream = new FileOutputStream(targetFile); outStream.write(buffer); outStream.flush(); }
注意

Apache Flink 要求信任库存在.

注意

要为本练习设置所需的先决条件,请先完成入门 (DataStreamAPI)练习。

使用 Amazon MSK 集群创建 VPC

要创建示例 VPC 和 Amazon MSK 集群以从 Kinesis Data Analytics 应用程序中访问,请按照开始使用亚马逊 MSK教程。

在完成本教程时,您还可以执行以下操作:

  • In第 5 步:创建主题,重复kafka-topics.sh --create命令以创建名为的目标主题AmazonKafkaTutorialTopicDestination

    bin/kafka-topics.sh --create --zookeeper ZooKeeperConnectionString --replication-factor 3 --partitions 1 --topic AmazonKafkaTutorialTopicDestination
    注意

    如果kafka-topics.sh命令返回ZooKeeperClientTimeoutException中,验证 Kafka 集群的安全组是否具有入站规则,以允许来自客户端实例的私有 IP 地址的所有流量。

  • 记录集群的引导服务器列表。您可以使用以下命令获取引导服务器列表(将 ClusterArn 替换为 MSK 集群的 ARN):

    aws kafka get-bootstrap-brokers --region us-west-2 --cluster-arn ClusterArn {... "BootstrapBrokerStringTls": "b-2.awskafkatutorialcluste.t79r6y.c4.kafka.us-west-2.amazonaws.com:9094,b-1.awskafkatutorialcluste.t79r6y.c4.kafka.us-west-2.amazonaws.com:9094,b-3.awskafkatutorialcluste.t79r6y.c4.kafka.us-west-2.amazonaws.com:9094" }
  • 在执行本教程中的步骤和先决条件教程中的步骤时,请务必使用所选的Amazon代码、命令和控制台条目中的区域。

创建自定义信任库并将其应用到您的集群

在本节中,您将创建自定义证书颁发机构 (CA),使用它生成自定义信任库,然后将其应用于 MSK 集群。

要创建和应用自定义信任库,请遵循客户端身份验证中的教程Amazon Managed Streaming for Apache Kafka 开发人员指南.

创建应用程序代码

在本节中,您下载并编译应用程序 JAR 文件。

在 GitHub 中提供了该示例的 Java 应用程序代码。要下载应用程序代码,请执行以下操作:

  1. 如果尚未安装 Git 客户端,请安装它。有关更多信息,请参阅安装 Git

  2. 使用以下命令克隆远程存储库:

    git clone https://github.com/aws-samples/amazon-kinesis-data-analytics-java-examples
  3. 应用程序代码位于 amazon-kinesis-data-analytics-java-examples/CustomKeystore/KDAFlinkStreamingJob.javaCustomFlinkKafkaConsumer.java 文件中。您可以检查代码以熟悉适用于 Apache Flink 的 Kinesis Data Analytics 应用程序代码结构。

  4. 使用命令行 Maven 工具或首选的开发环境以创建 JAR 文件。要使用命令行 Maven 工具编译 JAR 文件,请输入以下内容:

    mvn package -Dflink.version=1.13.2

    如果构建成功,则会创建以下文件:

    target/flink-app-1.0-SNAPSHOT.jar
    注意

    提供的源代码依赖于 Java 11 中的库。如果你使用的是开发环境,

上传 Apache Flink 流式处理 Java 代码

在本节中,您将应用程序代码上传到在中创建的 Amazon S3 存储桶。入门 (DataStreamAPI)教程。

注意

如果您从入门教程中删除了 Amazon S3 存储桶,请执行上传 Apache Flink 流式处理 Java 代码再次步骤。

  1. 在 Amazon S3 控制台中,选择ka-app-code-<username>存储桶,然后选择上传.

  2. 选择文件步骤中,选择添加文件。导航到您在上一步中创建的 KafkaGettingStartedJob-1.0.jar 文件。

  3. 您无需更改该对象的任何设置,因此,请选择 Upload (上传)

您的应用程序代码现在存储在 Amazon S3 存储桶中,应用程序可以在其中访问代码。

创建 应用程序

  1. 打开 Kinesis Data Analytics 控制台https://console.aws.amazon.com/kinesisanalytics.

  2. 在 Amazon Kinesis Data Analytics 控制面板上,选择创建分析应用.

  3. Kinesis Analytics – 创建应用程序页面上,提供应用程序详细信息,如下所示:

    • 对于 Application name (应用程序名称),输入 MyApplication

    • 适用于运行时,选择Apache Flink 版本 1.13.2.

  4. 对于访问权限,请选择 Create / update IAM role (创建/更新 IAM 角色) kinesis-analytics-MyApplication-us-west-2

  5. 选择 Create application(创建应用程序)。

注意

在使用控制台创建适用于 Apache Flink 的 Kinesis Data Analytics 应用程序时,您可以选择为应用程序创建 IAM 角色和策略。您的应用程序使用此角色和策略访问其从属资源。这些 IAM 资源使用您的应用程序名称和区域命名,如下所示:

  • 策略:kinesis-analytics-service-MyApplication-us-west-2

  • 角色:kinesis-analytics-MyApplication-us-west-2

配置应用程序

  1. MyApplication (我的应用程序) 页面上,选择 Configure (配置)

  2. Configure application (配置应用程序) 页面上,提供 Code location (代码位置)

    • 适用于Amazon S3 存储桶输入,ka-app-code-<username>.

    • 适用于Amazon S3 对象的路径输入,flink-app-1.0-SNAPSHOT.jar.

  3. Access to application resources (对应用程序的访问权限) 下,对于 Access permissions (访问权限),选择 Create / update IAM role (创建/更新 IAM 角色) kinesis-analytics-MyApplication-us-west-2

    注意

    在使用控制台指定应用程序资源(例如日志或 VPC)时,控制台修改应用程序执行角色以授予访问这些资源的权限。

  4. Properties (属性) 下面,选择 Add Group (添加组)。使用以下属性创建一个名为 KafkaSource 的属性组:

    密钥
    topic AmazonKafKat教程主题
    bootstrap.servers 您以前保存的引导服务器列表
    security.protocol SSL
    ssl.truststore.location /usr/lib/jvm/java-11-Amazon corret/lib/security/cacerts
    ssl.truststore.password changeit
    注意

    这些区域有:ssl.truststore.password对于默认证书为 “changit”,如果使用默认证书,则无需更改此值。

    再次选择 Add Group (添加组)。使用以下属性创建一个名为 KafkaSink 的属性组:

    密钥
    topic AmazonKafkatTORY 主题目的地
    bootstrap.servers 您以前保存的引导服务器列表
    security.protocol SSL
    ssl.truststore.location /usr/lib/jvm/java-11-Amazon corret/lib/security/cacerts
    ssl.truststore.password changeit
    交易 .timeout.ms 1000

    应用程序代码读取上述应用程序属性,以配置用于与 VPC 和 Amazon MSK 集群交互的源和接收器。有关使用属性的更多信息,请参阅运行时属性

  5. Snapshots (快照) 下面,选择 Disable (禁用)。这样,就可以轻松更新应用程序,而无需加载无效的应用程序状态数据。

  6. Monitoring (监控) 下,确保 Monitoring metrics level (监控指标级别) 设置为 Application (应用程序)

  7. 对于 CloudWatch logging (CloudWatch 日志记录),请选中 Enable (启用) 复选框。

  8. Virtual Private Cloud (VPC) 部分中,选择要与应用程序关联的 VPC。选择与您的 VPC 关联的子网和安全组,您希望应用程序使用它们访问 VPC 资源。

  9. 选择 Update(更新)。

注意

当你选择启用时 CloudWatch 日志记录时,Kinesis Data Analytics 将为您创建日志组和日志流。这些资源的名称如下所示:

  • 日志组:/aws/kinesis-analytics/MyApplication

  • 日志流:kinesis-analytics-log-stream

该日志流用于监控应用程序。

运行应用程序

可以通过运行应用程序、打开 Apache Flink 控制面板并选择所需的 Flink 作业来查看 Flink 作业图。

测试应用程序

在本节中,您将记录写入到源主题。应用程序从源主题中读取记录,并将其写入到目标主题中。您可以将记录写入到源主题以及从目标主题中读取记录,以验证应用程序是否正常工作。

要从主题中写入和读取记录,请按照中的步骤操作步骤 6:生成和使用数据中的开始使用亚马逊 MSK教程。

要从目标主题中读取,请在到集群的第二个连接中使用目标主题名称,而不是源主题:

bin/kafka-console-consumer.sh --bootstrap-server BootstrapBrokerString --consumer.config client.properties --topic AmazonKafkaTutorialTopicDestination --from-beginning

如果在目标主题中没有任何记录,请参阅故障排除主题中的无法访问 VPC 中的资源一节。