开源大数据平台E-MapReduce：使用SSL加密Kafka链接

前提条件

已在E-MapReduce控制台创建DataFlow集群（即后面说的Kafka集群），并选择了Kafka服务，详情请参见创建DataFlow Kafka集群。

配置SSL功能

E-MapReduce通过server.properties配置文件的kafka.ssl.config.type配置项来管理配置SSL的策略。

使用默认证书配置SSL

1. 进入服务的配置页面。
   1. 登录EMR on ECS控制台。
   2. 在顶部菜单栏处，根据实际情况选择地域和资源组。
   3. 单击目标集群操作列的集群服务。
   4. 在集群服务页面，单击Kafka服务区域的配置。
2. 修改配置项。
   1. 在配置页面，单击server.properties页签。
   2. 修改kafka.ssl.config.type的参数值为DEFAULT。
3. 保存配置。
   1. 单击保存。
   2. 在弹出的对话框中，输入执行原因。
   3. 单击确定。
4. 重启Kafka服务。
   1. 在Kafka服务的配置页面，选择更多操作 > 重启。
   2. 在弹出的对话框中，输入执行原因，单击确定。
   3. 在确认对话框中，单击确定。

自定义配置SSL

1. 进入服务的配置页面。
   1. 登录EMR on ECS控制台。
   2. 在顶部菜单栏处，根据实际情况选择地域和资源组。
   3. 单击目标集群操作列的集群服务。
   4. 在集群服务页面，单击Kafka服务区域的配置。
2. 修改配置项。
   1. 在配置页面，单击server.properties页签。
   2. 修改kafka.ssl.config.type的参数值为CUSTOM。
3. 保存配置。
   1. 单击保存。
   2. 在弹出的对话框中，输入执行原因。
      弹出的对话框中为您展示了本次配置联动修改的配置项及参数值。
   3. 单击确定。
4. 修改SSL其他配置。
   您需要根据业务需求，自行配置除listeners之外的SSL相关配置。例如，ssl.keystore.location，ssl.keystore.password，ssl.truststore.location，ssl.truststore.password，ssl.key.password，ssl.keystore.type和ssl.truststore.type等。
5. 重启Kafka服务。
   1. 在Kafka服务的配置页面，选择更多操作 > 重启。
   2. 在弹出的对话框中，输入执行原因，单击确定。
   3. 在确认对话框中，单击确定。

使用SSL连接Kafka

使用SSL连接Kafka时，需要客户端配置参数security.protocol、ssl.truststore.password和ssl.truststore.location。

例如，在已开启SSL的Kafka集群中，使用Kafka自带的Producer和Consumer执行作业，操作步骤如下：

1. 使用SSH方式登录Kafka集群，详情请参见登录集群。
2. 创建配置文件。
   1. 执行以下命令，创建配置文件ssl.properties。

      vim ssl.properties

   2. 添加以下内容至配置文件ssl.properties中。

      security.protocol=SSL
      ssl.truststore.location=/var/taihao-security/ssl/ssl/truststore
      ssl.truststore.password=${password}
      ssl.keystore.location=/var/taihao-security/ssl/ssl/keystore
      ssl.keystore.password=${password}
      ssl.endpoint.identification.algorithm=

      上面参数的值，您可以在EMR控制台Kafka服务的配置页面查看。如果是在Kafka集群以外的环境执行作业，您可以将Kafka集群中任意节点相应目录下的truststore和keystore文件，拷贝至运行环境进行相应配置。

3. 执行以下命令，创建Topic。

   kafka-topics.sh --partitions 10 --replication-factor 2 --bootstrap-server core-1-1:9092 --topic test --create

4. 执行以下命令，使用SSL配置文件产生数据。

   export IP=<your_InnerIP>
   kafka-producer-perf-test.sh --topic test --num-records 123456 --throughput 10000 --record-size 1024 --producer-props bootstrap.servers=${IP}:9092 --producer.config ssl.properties

   说明 本文代码示例中的your_InnerIP为master-1-1节点的内网IP地址。
5. 执行以下命令，使用SSL配置文件消费数据。

   export IP=<your_InnerIP>
   kafka-consumer-perf-test.sh --broker-list ${IP}:9092 --messages 100000000 --topic test --consumer.config ssl.properties