如何开始使用加密？ - Amazon Managed Streaming for Apache Kafka
AWS 文档中描述的 AWS 服务或功能可能因区域而异。要查看适用于中国区域的差异，请参阅中国的 AWS 服务入门。

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

如何开始使用加密？

创建 MSK 集群时，您可以使用 JSON 格式指定加密设置。以下是示例。

{
   "EncryptionAtRest": {
       "DataVolumeKMSKeyId": "arn:aws:kms:us-east-1:123456789012:key/abcdabcd-1234-abcd-1234-abcd123e8e8e"
    },
   "EncryptionInTransit": {
        "InCluster": true,
        "ClientBroker": "TLS"
    }
}

对于 DataVolumeKMSKeyId，您可以为账户中的 MSK 指定客户托管 CMK 或 AWS 托管 CMK (alias/aws/kafka)。如果您没有指定 EncryptionAtRest，Amazon MSK 仍会在 AWS 托管 CMK 下对数据进行静态加密。要确定您的集群使用的 CMK，请发送 GET 请求或调用 DescribeCluster API 操作。

对于 EncryptionInTransit，InCluster 的默认值为 true，但是如果您不想在代理之间传递数据时让 Amazon MSK 加密数据，则可以将此项设置为 false。

要为客户端和代理之间传输的数据指定加密模式，请将 ClientBroker 设置为以下三个值之一：TLS、TLS_PLAINTEXT 或 PLAINTEXT。

创建集群时指定加密设置

  1. 将上一示例的内容保存在文件中，并为该文件指定所需的任何名称。例如，将其命名为 encryption-settings.json。

  2. 运行 create-cluster 命令并使用 encryption-info 选项指向您保存配置 JSON 的文件。以下是示例。 

    aws kafka create-cluster --cluster-name "ExampleClusterName" --broker-node-group-info file://brokernodegroupinfo.json --encryption-info file://encryptioninfo.json --kafka-version "2.2.1" --number-of-broker-nodes 3

    以下是运行此命令后的成功响应示例。

    {
    "ClusterArn": "arn:aws:kafka:us-east-1:123456789012:cluster/SecondTLSTest/abcdabcd-1234-abcd-1234-abcd123e8e8e",
    "ClusterName": "ExampleClusterName",
    "State": "CREATING"
}

测试 TLS 加密

  1. 按照第4步: 创建客户机中的指导创建客户端计算机。

  2. 在客户端计算机上安装 Apache Kafka。

  3. 在已安装 AWS CLI 的计算机上运行以下命令，并将 ClussterARN 替换为集群 ARN（如前面过程中的示例所示，将 ClientBroker 设置为 TLS 创建的集群）。 

    aws kafka describe-cluster --cluster-arn clusterARN

    在结果中，查找 ZookeeperConnectString 的值并保存它，因为您需要在下一步中使用该值。

  4. 转到客户端计算机上 Apache Kafka 安装的 bin 文件夹。要创建主题，请运行以下命令，将 ZookeeperConnectString 替换为您在上一步中获取的 ZookeeperConnectString 的值。 

    kafka-topics.sh --create --zookeeper ZookeeperConnectString --replication-factor 3 --partitions 1 --topic TLSTestTopic

  5. 在本示例中，我们使用 JVM 信任存储与 MSK 集群通信。为此，请首先在客户端计算机上创建一个名为 /tmp 的文件夹。然后，转到 Apache Kafka 安装的 bin 文件夹，并运行以下命令。（您的 JVM 路径可能不相同。） 

    cp /usr/lib/jvm/java-1.8.0-openjdk-1.8.0.201.b09-0.amzn2.x86_64/jre/lib/security/cacerts /tmp/kafka.client.truststore.jks

  6. 仍在客户端计算机上的 Apache Kafka 安装的 bin 文件夹中，创建一个名为 client.properties 的文本文件，该文件包含以下内容。 

    security.protocol=SSL
ssl.truststore.location=/tmp/kafka.client.truststore.jks

  7. 在已安装 AWS CLI 的计算机上运行以下命令，将 clusterARN 替换为集群 ARN。 

    aws kafka get-bootstrap-brokers --cluster-arn clusterARN

    成功结果如下所示。保存此结果，因为您需要在下一步中使用它。

    {
    "BootstrapBrokerStringTls": "a-1.example.g7oein.c2.kafka.us-east-1.amazonaws.com:0123,a-3.example.g7oein.c2.kafka.us-east-1.amazonaws.com:0123,a-2.example.g7oein.c2.kafka.us-east-1.amazonaws.com:0123"
}

  8. 在客户端计算机上的 Apache Kafka 安装的 bin 文件夹中，运行以下命令，将 BootstrapBrokerStringTls 替换为您在上一步中获得的值。保持运行此生成器命令。 

    kafka-console-producer.sh --broker-list BootstrapBrokerStringTls --producer.config client.properties --topic TLSTestTopic

  9. 在同一客户端计算机上打开一个新的命令窗口，转到 Apache Kafka 安装的 bin 文件夹，然后运行以下命令创建使用器。 

    kafka-console-consumer.sh --bootstrap-server BootstrapBrokerStringTls --consumer.config client.properties --topic TLSTestTopic

  10. 在生成器窗口中，键入文本消息后键入 return，并在使用器窗口中查找相同消息。Amazon MSK 会在传输过程中对此消息进行加密。

有关配置 Apache Kafka 客户端以使用加密数据的更多信息，请参阅配置 Kafka 客户端。