本文为您介绍如何配置E-MapReduce Kafka集群的SASL（Simple Authentication and Security Layer）功能，并通过SASL功能登录Kafka集群。

前提条件

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

配置SASL功能

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

Kafka集群的SASL功能默认关闭，您可以执行以下步骤快速开启SASL功能。本文以配置SASL/SCRAM-SHA-512认证机制为例。

  1. 进入服务的配置页面。
    1. 登录EMR on ECS控制台。
    2. 在顶部菜单栏处，根据实际情况选择地域和资源组。
    3. 单击目标集群操作列的集群服务。
    4. 在集群服务页面，单击Kafka服务区域的配置。
  2. 修改listener配置。
    1. 在配置页面，单击server.properties页签。
    2. 修改配置项kafka.sasl.config.type的值为CUSTOM，单击保存。
    3. 在弹出的对话框中，输入执行原因，打开自动配置更新开关，单击确定。
  3. 修改SASL认证机制相关的配置。
    1. 使用SSH方式登录Kafka集群，详情请参见登录集群。
    2. 执行以下命令，创建admin用户。
      kafka-configs.sh --zookeeper master-1-1:2181/emr-kafka --alter --add-config 'SCRAM-SHA-256=[password=admin-secret],SCRAM-SHA-512=[password=admin-secret]' --entity-type users --entity-name admin
    3. 设置SASL配置项。
      在Kafka服务配置页面的server.properties页签，添加配置项。
      1. 单击新增配置项。
      2. 在新增配置项对话框中，添加以下配置，单击确定。
        参数参数值
        sasl.mechanism.inter.broker.protocolSCRAM-SHA-512
        sasl.enabled.mechanismsSCRAM-SHA-512
      3. 在弹出对话框中，输入执行原因，打开自动配置更新开关，单击确定。
    4. 配置服务端JAAS。
      • 方式一：通过自定义配置项配置服务端JAAS。
        1. 在Kafka服务配置页面，单击server.properties页签。
        2. 单击新增配置项，新增以下配置项，单击确定。
          参数参数值
          listener.name.sasl_plaintext.sasl.enabled.mechanismsSCRAM-SHA-512
          listener.name.sasl_plaintext.scram-sha-512.sasl.jaas.configorg.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin-secret" ;
        3. 在弹出的对话框中，输入执行原因，打开自动配置更新开关，单击确定。
      • 方式二：通过文本文件配置JAAS。
        1. 在Kafka服务配置页面，修改以下配置项，单击保存。
          页签参数参数值
          kafka_server_jaas.confkafka.server.jass.content
          KafkaServer {
org.apache.kafka.common.security.scram.ScramLoginModule required
username="admin"
password="admin-secret";
};
          server.propertieskafka_opts-Djava.security.auth.login.config=/etc/taihao-apps/kafka-conf/kafka-conf/kafka_server_jaas.conf
        2. 在弹出的对话框中，输入执行原因，打开自动配置更新开关，单击确定。
    5. 配置客户端JAAS。

      通过kafka_client_jaas.conf配置文件的kafka.client.jass.content配置项，配置Kafka客户端JAAS，该配置将会用于启动Kafka Schema Registry以及Kafka Rest Proxy组件。

      1. 在Kafka服务配置页面，修改以下配置项，单击保存。
        页签参数参数值
        kafka_client_jaas.confkafka.client.jaas.content
        KafkaClient {
org.apache.kafka.common.security.scram.ScramLoginModule required
username="admin"
password="admin-secret";
};
        schema-registry.propertiesschema_registry_opts-Djava.security.auth.login.config=/etc/taihao-apps/kafka-conf/kafka-conf/kafka_client_jaas.conf
        kafka-rest.propertieskafkarest_opts-Djava.security.auth.login.config=/etc/taihao-apps/kafka-conf/kafka-conf/kafka_client_jaas.conf
      2. 在弹出的对话框中，输入执行原因，打开自动配置更新开关，单击确定。
  4. 重启Kafka服务。
    1. 在Kafka服务的配置页面，选择更多操作 > 重启。
    2. 在弹出的对话框中，输入执行原因，单击确定。
    3. 在确认对话框中，单击确定。

使用SASL登录Kafka集群

本文以SASL/SCRAM-SHA-512机制为例，使用Kafka自带的Producer和Consumer执行作业，介绍客户端如何认证登录Kafka服务。

  1. 使用SSH方式登录Kafka集群，详情请参见登录集群。
  2. 执行以下命令，创建用于登录的用户名和密码。
    kafka-configs.sh --zookeeper master-1-1:2181/emr-kafka --alter --add-config 'SCRAM-SHA-256=[password=client-secret],SCRAM-SHA-512=[password=client-secret]' --entity-type users --entity-name client-user
  3. 创建配置文件。
    1. 执行以下命令，创建配置文件sasl.properties。
      vim sasl.properties
    2. 添加以下内容至配置文件sasl.properties中。
      security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="client-user" password="client-secret";
  4. 执行以下命令，创建Topic。
    kafka-topics.sh --partitions 10 --replication-factor 2 --zookeeper master-1-1:2181/emr-kafka --topic test --create
  5. 执行以下命令，使用SASL配置文件产生数据。
    kafka-producer-perf-test.sh --topic test --num-records 123456 --throughput 10000 --record-size 1024 --producer-props bootstrap.servers=core-1-1:9092 --producer.config sasl.properties
  6. 执行以下命令，使用SASL配置文件消费数据。
    kafka-consumer-perf-test.sh --broker-list core-1-1:9092 --messages 100000000 --topic test --consumer.config sasl.properties