# 配置Kafka数据安全传输协议
#### 操作场景
在企业级Kafka集群部署中，数据传输安全是核心关注点。默认情况下，Kafka使用明文传输数据，存在数据被窃听、篡改的风险。特别是在跨网络、跨数据中心传输敏感数据时，需要启用加密传输协议来保障数据安全。
本文介绍如何配置Kafka的安全传输协议，实现数据的加密传输和身份认证。
- **SSL（Secure Sockets Layer）**：安全套接字层，一种加密传输协议，用于在网络上传输数据时进行加密，防止数据被窃听和篡改。
- **SASL（Simple Authentication and Security Layer）**：简单认证和安全层，一种认证框架，支持多种认证机制（如Kerberos），用于验证客户端和服务端的身份。
- **ACL（Access Control List）**：访问控制列表，用于控制用户对Kafka Topic的访问权限，包括读写权限、创建删除权限等。
- **Producer API**：Kafka生产者API，定义在org.apache.kafka.clients.producer.KafkaProducer中，用于向Kafka Topic发送消息。使用kafka-console-producer.sh时默认使用此API。
- **Consumer API**：Kafka消费者API，定义在org.apache.kafka.clients.consumer.KafkaConsumer中，用于从Kafka Topic消费消息。使用kafka-console-consumer.sh时默认使用此API。
 
#### 约束与限制
- MRS 3.x及之后版本，Kafka不支持旧Producer API和旧Consumer API。关于新、旧API介绍请参见[Kafka安全使用说明](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_300744.html)。
- 启用SSL协议需要配置SSL证书。
- 启用SASL认证需要配置Kerberos认证。
 
#### 前提条件
- 已安装Kafka客户端。
- Kafka服务正常运行。
 
#### 配置Kafka访问协议
Kafka支持四种协议类型的访问，可以根据安全需求选择合适的协议：
表1Kafka访问协议类型 
| 协议类型           | 说明                    | 默认端口                            | 适用场景                   |
|:---|:---|:---|:---|
| PLAINTEXT      | 支持无认证的明文访问。           | 获取参数"port"的值，默认为9092。           | 内部网络，对安全性要求不高的场景。      |
| SASL_PLAINTEXT | 支持Kerberos认证的明文访问。    | 获取参数"sasl.port"的值，默认为21007。     | 需要身份认证但对数据加密要求不高的场景。   |
| SSL            | 支持无认证的SSL加密访问。        | 获取参数"ssl.port"的值，默认为9093。       | 需要数据加密但不需要身份认证的场景。     |
| SASL_SSL       | 支持Kerberos认证的SSL加密访问。 | 获取参数"sasl-ssl.port"的值，默认为21009。 | 需要身份认证和数据加密的高安全场景（推荐）。 |
   
- Kafka服务启动时，默认会启动PLAINTEXT和SASL_PLAINTEXT两种协议类型的访问监测。 可通过设置Kafka服务配置"ssl.mode.enable"为"true"，来启动SSL和SASL_SSL两种协议类型的访问监测。
  
- 可以参考以下步骤进入Kafka全部配置页面，查看或修改参数。
  1. 登录FusionInsight Manager。 登录集群Manager具体操作，请参考[访问MRS集群Manager](https://support.huaweicloud.com/usermanual-mrs/mrs_01_0128.html)。
     
  
  2. 选择"集群 \> 服务 \> Kafka \> 配置 \> 全部配置"。
  
  3. 搜索参数名称，并根据实际需要修改参数值。
  
  4. 修改完成后重启配置过期的实例生效。
  
  5. 配置Topic的ACL权限（可选）。 如果需要对Topic进行访问控制，需要在Linux客户端上使用"kafka-acls.sh"脚本进行查看和设置。
     了解如何设置Topic权限，请参考[Kafka用户权限说明](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_0378.html)。
     
   
 
#### Kafka中API使用说明
- 场景一：访问设置了ACL的Topic
  表2访问设置了ACL的Topic 
  | 用户属组                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                         | 客户端参数                                                                            | 服务端参数                      | 访问的端口                  |
  |:---|:---|:---|:---|
  | 用户需满足以下条件之一即可： - 加入System_administrator角色  - 属于kafkaadmin组  - 属于kafkasuperuser组  - 被授权的kafka组的用户   | security.inter.broker.protocol=SASL_PLAINTEXT sasl.kerberos.service.name = kafka | -                          | sasl.port（默认21007）     |
  | 用户需满足以下条件之一即可： - 加入System_administrator角色  - 属于kafkaadmin组  - 属于kafkasuperuser组  - 被授权的kafka组的用户   | security.protocol=SASL_SSL sasl.kerberos.service.name = kafka                    | "ssl.mode.enable"配置为"true" | sasl-ssl.port（默认21009） |
     
  
- 场景二：访问未设置ACL的Topic
  表3访问未设置ACL的Topic 
  | 用户属组                                                                                                                                                                                                                                                                                                                                                                                                                                                                                     | 客户端参数                                                               | 服务端参数                                                                                                                                                                                                                                                                                                                              | 访问的端口                  |
  |:---|:---|:---|:---|
  | 用户需满足以下条件之一： - 加入System_administrator角色    - 属于kafkaadmin组  - 属于kafkasuperuser组   | security.protocol=SASL_PLAINTEXT sasl.kerberos.service.name = kafka | -                                                                                                                                                                                                                                                                                                                                  | sasl.port（默认21007）     |
  | 用户属于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）     |
  | 用户需满足以下条件之一： - 加入System_administrator角色  - 属于kafkaadmin组  - kafkasuperuser组用户    | security.protocol=SASL_SSL sasl.kerberos.service.name = kafka       | "ssl.mode.enable"配置为"true"                                                                                                                                                                                                                                                                                                         | sasl-ssl.port（默认21009） |
  | 用户属于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） |
  | -                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        | security.protocol=PLAINTEXT                                         | "allow.everyone.if.no.acl.found"配置为"true"                                                                                                                                                                                                                                                                                          | port（默认9092）           |
  | -                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        | security.protocol=SSL                                               | 1. "allow.everyone.if.no.acl.found"配置为"true"  2. "ssl.mode.enable"配置为"true"   | ssl.port（默认9063）       |
     
  **验证操作：**
  配置完成后，可通过以下方式验证安全传输协议是否生效：
  1. 使用kafka-console-producer.sh发送消息，指定security.protocol参数为SASL_SSL。
  
  2. 使用kafka-console-consumer.sh消费消息，指定security.protocol参数为SASL_SSL。
  
  3. 如果消息能够正常发送和消费，则表示安全传输协议配置成功。
   
 
#### 常见问题
- **SSL连接失败**
  SSL连接失败的常见原因包括：SSL证书配置错误、证书过期、客户端不信任服务端证书等。
  请检查SSL证书是否正确配置，确认证书未过期。如果是自签名证书，需要在客户端信任该证书。查看Kafka服务端日志定位具体错误信息。
  
- **SASL认证失败**
  SASL认证失败的常见原因包括：Kerberos配置错误、Principal名称错误、Keytab文件权限不正确等。
  请检查Kerberos配置是否正确，确认Principal名称和Keytab文件路径正确。使用**kinit**命令测试Kerberos认证是否正常。
  
 
#### 相关文档
- 新、旧版API介绍请参考[Kafka安全使用说明](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_300744.html)。
- 新、旧版API获取Kafka Consumer Offset信息相关操作，请参考[如何获取Kafka Consumer Offset信息](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_300743.html)。
- 如果需要设置Topic的权限信息，可参考[Kafka用户权限说明](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_0378.html)章节进行操作。
- 创建用户相关操作请参考[创建Kafka用户并绑定角色](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_1032.html#mrs_01_1032__section15172119027)。