3.7 — 为您的 Kafka 集群配置安全组入站规则。

a. 打开 AWS Cloud9 控制台，然后针对您创建的环境选择 View details（查看详细信息）选项卡。

e. 在 Security Groupsp（安全组）页面上，点击 Inbound rules（入站规则）选项卡，然后点击 Edit inbound rules（编辑入站规则）。

f. 在 Edit inbound rules（编辑入站规则）页面上，使用端口 9094 和 2181 将复制的 security group ID（安全组 ID）添加为 Kafka 安全组的入站规则。

3.8 — 在 Kafka 集群中创建一个新主题 AWSKafkaTutorialTopic。生产者和消费者将使用此主题发布和订阅消息。在以下命令（如适用）中，更改部分 3.6.c 中设置为变量的主题的名称。

在 AWS Cloud9 终端运行以下命令：

这将创建一个新主题 AWSKafkaTutorialTopic。确认命令返回以下消息：已创建主题 AWSKafkaTutorialTopic。

注意：如果显示客户端会话超时错误，请检查集群的 Security Group Inbound Rules（安全组入站规则）是否配置正确。

4.1 — 在 AWS Cloud9 终端中运行以下命令，创建一个 client 文件夹，用于存储证书文件，并将 Java 密钥库复制到此文件夹。

4.7 — 运行以下命令，通过 AWS CLI 使用在上一步中收到的证书 ARN 获取证书。收到的响应为 JSON 格式，由 jq 解析，并保存到 signed-certificate-from-acm 文件中。

4.8 — 运行以下命令，将签名的证书导入密钥库。

a. 终端出现提示时，输入 Yes。  

b. client 目录下应有 4 个文件。

• client-cert-sign-request
• kafka.client.keystore.jks
• kafka.client.truststore.jks
• signed-certificate-from-acm
  

4.9 — 以前配置的 MSK 集群要求所有客户端通信都通过 TLS 以确保安全。文件 kafka.client.keystore.jks 必须以 SecretBinary 的形式上传到 Secrets Manager，以对 IoT Core 规则进行身份验证。

运行以下命令，在 Amazon Secrets Manager 中创建一个名为 Kafka_Keystore 的密钥。

可以在 Secrets Manager 控制台中验证新创建的密钥。

4.10 — 在新终端设置 Kafka 消费者，同时保持生产者会话处于活动状态。

注意：可以在 AWS Cloud9 控制台中选择 + 符号或 alt+T，打开新终端。

在新终端会话中，运行以下命令，设置初始变量并启动控制台生产者。

4.12 — 运行以下命令，启动 Kafka 消费者会话。

您现在应该启动了 Kafka 生产者会话和消费者会话。

4.13 — 在生产者终端中输入任意字符串，然后按 Enter（回车键）发布到 Kafka 流。

5.1 — 创建一个具有 AWS Secrets Manager 权限的 AWS IAM 角色，以允许 AWS IoT Core 规则访问存储在 Secrets Manager 中的 Kafka 密钥库。

a. 打开 IAM 控制台，从左侧导航窗格中点击 Roles（角色），然后点击 Create role（创建角色）。

b. 对于 Select type of trusted entity（选择可信实体的类型），选择 AWS service（AWS 服务）。对于 Choose a use case（选择使用场景），选择 EC2，然后点击 Next: Permissions（下一步：权限）。

5.2 — 在 Create role（创建角色）页面上的搜索框中，输入 SecretsManagerReadWrite，选择 SecretsManagerReadWrite 策略，然后点击 Next: Tags（下一步：标签）。

5.3 — 在 Create role（创建角色）页面上的 Add tags (optional)（添加标签（可选））部分中，添加任何必要标签。然后，点击 Next: Review（下一步：检查）。

5.4 — 在 Create role（创建角色）页面上的 Review（检查）部分中，对于 Role name（角色名称），输入 kafkaRole，并（可选）添加描述。然后，点击 Create role（创建角色）。

5.5 — 创建角色后，点击新创建的 kafkaRole。在角色的 Summary（摘要）页面上，点击 Trust relationships（信任关系）选项卡，点击 Edit trust relationship（编辑信任关系），然后复制并粘贴以下详细信息。

5.6 — 然后，点击 Update Trust Policy（更新信任策略）。

5.7 — 打开 AWS IoT Core 控制台，然后在左侧导航栏中，点击 ACT。然后，点击 Rules（规则）。点击 Create a rule（创建规则）。

5.8 — 在 Create a rule（创建规则）页面上，输入 Name（名称）和 Description（描述）。在 Role query statement（角色查询语句）部分中，选择 SQL 版本，并输入角色查询语句。然后，点击 Add action（添加操作）。

5.9 — 在 Select an action（选择操作）页面上，选中 Send a message to an Apache Kafka cluster（将消息发送到 Apache Kafka 集群），然后点击 Configure action（配置操作）。

5.10 — 在 Configure action（配置操作）页面上，点击 Create a VPC destination（创建 VPC 目标）。

5.11 — 在 Create a VPC destination（创建 VPC 目标）页面上，选择创建 Kafka 集群的 VPC ID。

5.12 — 选择 Subnet IDs（子网 ID），然后选择 Security group（安全组）。

注意：请确保选择可以访问 Kafka 集群的安全组。

选择之前创建的 EC2 实例的安全组设置和/或 Kafka 集群本身的安全组。

5.13 — 点击 Create Role（创建角色），然后在 Create a new role（创建新角色）窗口中，将角色命名为 AWSKafkaTutorialTopic。然后，点击 Create role（创建角色）。

5.14 — 点击 Create Destination（创建目标）。

注意：目标需要 5-10 分钟才能变为已启用状态。状态变为 Enabled（已启用）后，返回规则创建页面。

5.15 — 在 Web 浏览器中，导航至先前打开的 AWS IoT - Rules - Configuration action（AWS IoT - 规则 - 配置操作）页面。然后，从下拉列表中选择新创建的 VPC Destination（VPC 目标），并对 Kafka topic（Kafka 主题）输入 AWSKafkaTutorialTopic。保留 Key（键）和 Partition（分区）为空。

5.16 — 在 Configure action（配置操作）页面上的 Client properties（客户端属性）部分中，输入以下详细信息。

• Bootstrap.server = Kafka 集群的 TLS 引导字符串
• security.protocol = SSL
• ssl.truststore = 保留为空
  
• ss.truststore.password = 保留为空
• ssl.keystore = ${get_secret('Kafka_Keystore','SecretBinary','arn:aws:iam::[YOUR ROLE ARN FOR IAM]:role/kafkaRole')}
• ssl.keystore.password = myPassword
  
  注意：密码是您为 myPassword 输入的变量，该变量在前面的步骤中进行了配置。
  
  

5.17 — 然后，点击 Add action（添加操作）。

5.18 — 在 Create a rule（创建规则）页面上，点击 Create rule（创建规则）。

5.19 — 使用以下步骤测试 MQTT 流。

a. 在 AWS IoT Core 的左侧导航栏中，点击 Test（测试），然后点击 MQTT test client（MQTT 测试客户端）。

b. 在 MQTT test client（MQTT 测试客户端）页面上，点击 publish to a topic（发布到主题）选项卡。

c. 在 Topic name（主题名称）下，输入名称。在 Message payload（消息负载）下，输入消息。

d. 在活动的 AWS Cloud9 消费者会话（在上一步中创建/打开）中，您应该会看到从 MQTT 主题发布并流式传输到 Kafka 消费者的数据。

6.1 — 打开 AWS IoT 控制台，然后在左侧导航栏中，点击 Settings（设置）。然后，在 Logs（日志）部分中，点击 Manage Logs（管理日志）。

a. 在 Log level（日志级别）部分中，从下拉列表中选择 Debug (most verbosity)（调试（最详细）），然后点击 Update（更新）。 

6.2 — 要创建 IoT 日志级别，请打开 IoT 控制台，并点击 Rules（规则）。然后，选择 kafkaRole。

6.3 — 在 Error action（错误操作）部分中，点击 Add action（添加操作）。然后，选择 Republish message to an AWS IoT topic（将消息重新发布到 AWS IoT 主题），并点击 Configure action（配置操作）。

现在可以在 Amazon Cloudwatch、日志组和 AWS IoT 日志中查看您的 IoT 日志。

7.10 — 打开 Secrets Manager 控制台，然后在左侧栏中，点击 Secrets（密钥）。

7.11 — 选择 kafka_keystore 密钥。

7.12 — 在 Secret details（密钥详细信息）部分中，点击 Delete secret（删除密钥）。

7.13 — 打开 ACM Private CA 控制台。

7.14 — 点击 Private CAs（私有 CA）。

7.15 — 从列表中选择您的私有 CA。

7.16 — 如果您的 CA 处于 ACTIVE 状态，则必须将其禁用。在 Actions（操作）菜单上，点击 Disable（禁用）。

7.17 — 在 Actions（操作）菜单上，点击 Delete（删除）。

7.18 — 如果您的 CA 处于 PENDING_CERTIFICATE、EXPIRED 或 DISABLED 状态，请指定 7 到 30 天的恢复期。然后，点击 Delete（删除）。

注意：如果您的私有 CA 不处于其中一种状态，后续将无法恢复。

7.19 — 如果您确定要删除私有 CA，请在出现提示时选择 Permanently delete（永久删除）。私有 CA 的状态将更改为 DELETED。

7.20 — 打开 AWS Cloud9 控制台。

7.21 — 在顶部导航栏中，选择环境所在的 AWS 区域。

7.22 — 在环境列表中，对于要删除的环境，选择该环境对应的卡式窗口标题。

7.23 — 点击 Delete（删除）。

7.24 — 在 Delete（删除）对话框中，输入 Delete，然后点击 Delete（删除）。