# 配置Kafka数据安全传输协议
本章节内容适用于MRS 3.x及后续版本。
#### Kafka API简单说明
- Producer API 指**org.apache.kafka.clients.producer.KafkaProducer**中定义的接口，在使用"kafka-console-producer.sh"时，默认使用此API。
  
- Consumer API 指**org.apache.kafka.clients.consumer.KafkaConsumer**中定义的接口，在使用"kafka-console-consumer.sh"时，默认会调用此API。
  
![](https://support.huaweicloud.com/cmpntguide-mrs/public_sys-resources/note_3.0-zh-cn.png)
MRS 3.x后，Kafka不支持旧Producer API和旧Consumer API。
#### Kafka访问协议说明
请参考[修改集群服务配置参数](https://support.huaweicloud.com/cmpntguide-mrs/mrs_01_2125.html)查看或配置参数。
Kafka当前支持四种协议类型的访问：PLAINTEXT、SSL、SASL_PLAINTEXT、SASL_SSL。
Kafka服务启动时，默认会启动PLAINTEXT和SASL_PLAINTEXT两种协议类型的安全认证。可通过设置Kafka服务配置"ssl.mode.enable"为"true"，来启动SSL和SASL_SSL两种协议类型的安全认证。下表是四种协议类型的简单说明：
| 协议类型           | 说明                   | 默认端口                           |
|:---|:---|:---|
| PLAINTEXT      | 支持无认证的明文访问           | 获取参数"port"的值，默认为9092           |
| SASL_PLAINTEXT | 支持Kerberos认证的明文访问    | 获取参数"sasl.port"的值，默认为21007     |
| SSL            | 支持无认证的SSL加密访问        | 获取参数"ssl.port"的值，默认为9093       |
| SASL_SSL       | 支持Kerberos认证的SSL加密访问 | 获取参数"sasl-ssl.port"的值，默认为21009 |
   
#### Topic的ACL设置
Topic的权限信息，需要在Linux客户端上，使用"kafka-acls.sh"脚本进行查看和设置，具体可参考[Kafka用户权限说明](https://support.huaweicloud.com/cmpntguide-mrs/mrs_01_0378.html)。
#### 针对不同的Topic访问场景，Kafka中API使用说明
- 场景一：访问设置了ACL的Topic
  
  | 使用的API                                                      | 用户属组                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                         | 客户端参数                                                                            | 服务端参数                    | 访问的端口                  |
  |:---|:---|:---|:---|:---|
  | API  | 用户需满足以下条件之一即可： - 加入System_administrator角色  - 属于kafkaadmin组  - 属于kafkasuperuser组  - 被授权的kafka组的用户   | security.inter.broker.protocol=SASL_PLAINTEXT sasl.kerberos.service.name = kafka | -                        | sasl.port（默认21007）     |
  | API  | 用户需满足以下条件之一即可： - 加入System_administrator角色  - 属于kafkaadmin组  - 属于kafkasuperuser组  - 被授权的kafka组的用户   | security.protocol=SASL_SSL sasl.kerberos.service.name = kafka                    | "ssl.mode.enable"配置为true | sasl-ssl.port（默认21009） |
     
  
- 场景二：访问未设置ACL的Topic
  
  | 使用的API | 用户属组                                                                                                                                                                                                                                                                                                                                                                                                                                                                                     | 客户端参数                                                               | 服务端参数                                                                                                                                                                                                                                                                                                                              | 访问的端口                  |
  |:---|:---|:---|:---|:---|
  | API    | 用户需满足以下条件之一： - 加入System_administrator角色    - 属于kafkaadmin组  - 属于kafkasuperuser组   | security.protocol=SASL_PLAINTEXT sasl.kerberos.service.name = kafka | -                                                                                                                                                                                                                                                                                                                                  | sasl.port（默认21007）     |
  | API    | 用户属于kafka组                                                                                                                                                                                                                                                                                                                                                                                                                                                                               | security.protocol=SASL_PLAINTEXT sasl.kerberos.service.name = kafka | "allow.everyone.if.no.acl.found"配置为true 说明： 普通集群下不涉及服务端参数"allow.everyone.if.no.acl.found"的修改                                                                                                                                                       | sasl.port（默认21007）     |
  | API    | 用户需满足以下条件之一： - 加入System_administrator角色  - 属于kafkaadmin组  - kafkasuperuser组用户    | security.protocol=SASL_SSL sasl.kerberos.service.name = kafka       | "ssl.mode.enable"配置为"true"                                                                                                                                                                                                                                                                                                         | sasl-ssl.port（默认21009） |
  | API    | 用户属于kafka组                                                                                                                                                                                                                                                                                                                                                                                                                                                                               | security.protocol=SASL_SSL sasl.kerberos.service.name = kafka       | 1. "allow.everyone.if.no.acl.found"配置为"true"  2. "ssl.mode.enable"配置为"true"   | sasl-ssl.port（默认21009） |
  | API    | -                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        | security.protocol=PLAINTEXT                                         | "allow.everyone.if.no.acl.found"配置为"true"                                                                                                                                                                                                                                                                                          | port（默认9092）           |
  | API    | -                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        | security.protocol=SSL                                               | 1. "allow.everyone.if.no.acl.found"配置为"true"  2. "ssl.mode.enable"配置为"true"   | ssl.port（默认9063）       |