配置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安全使用说明。
• 启用SSL协议需要配置SSL证书。
• 启用SASL认证需要配置Kerberos认证。

前提条件

• 已安装Kafka客户端。
• Kafka服务正常运行。

配置Kafka访问协议

• Kafka服务启动时，默认会启动PLAINTEXT和SASL_PLAINTEXT两种协议类型的访问监测。

  可通过设置Kafka服务配置“ssl.mode.enable”为“true”，来启动SSL和SASL_SSL两种协议类型的访问监测。

• 可以参考以下步骤进入Kafka全部配置页面，查看或修改参数。
  1. 登录FusionInsight Manager。

     登录集群Manager具体操作，请参考访问MRS集群Manager。

  2. 选择“集群 > 服务 > Kafka > 配置 > 全部配置”。
  3. 搜索参数名称，并根据实际需要修改参数值。
  4. 修改完成后重启配置过期的实例生效。
  5. 配置Topic的ACL权限（可选）。

     如果需要对Topic进行访问控制，需要在Linux客户端上使用“kafka-acls.sh”脚本进行查看和设置。

     了解如何设置Topic权限，请参考Kafka用户权限说明。

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）
  	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组		“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组		“allow.everyone.if.no.acl.found”配置为“true” “ssl.mode.enable”配置为“true”	sasl-ssl.port（默认21009）
  -	security.protocol=PLAINTEXT	“allow.everyone.if.no.acl.found”配置为“true”	port（默认9092）
  -	security.protocol=SSL	“allow.everyone.if.no.acl.found”配置为“true” “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安全使用说明。
• 新、旧版API获取Kafka Consumer Offset信息相关操作，请参考如何获取Kafka Consumer Offset信息。
• 如果需要设置Topic的权限信息，可参考Kafka用户权限说明章节进行操作。
• 创建用户相关操作请参考创建Kafka用户并绑定角色。