一、前言
Apache Kafka作为常用的开源分布式流媒体平台，可以实时发布、订阅、存储和处理数据流,多用于作为消息队列获取实时数据，构建对数据流的变化进行实时反应的应用程序，已被数千家公司用于高性能数据管道、流分析、数据集成和任务关键型应用程序。而其中Apache Kafka Connect 作为 Kafka 中用于和其他数据系统流式传输数据的服务，其独立运行版本可以在 Kafka 发布包中通过 bin/connect-standalone.sh 启动，默认会在 8083 端口开启 HTTP REST API 服务，攻击者可以利用基于SASLJAAS 配置和SASL 协议的任意Kafka客户端，对可对连接器（Connector）的配置进行操作，将连接器中的 Kafka 客户端 sasl.jaas.config 属性值设置为 com.sun.security.auth.module.JndiLoginModule（通过 producer.override.sasl.jaas.config, consumer.override.sasl.jaas.config 或 admin.override.sasl.jaas.config 属性进行配置）时，如果此时连接器连接到攻击者可控的 LDAP 服务器时容易受到反序列化攻击，也称JNDI 注入来实现远程任意代码执行。云平台中，Kafka Connect 服务通常用于提供 Kafka 数据迁移、数据同步的管道能力，其默认 HTTP API 开放于 8083 端口。
因现场kafka选用版本较低，安全扫描时触发安全风险告警，低于 Kafka 升级3.4.0版本，涉及【Apache Kafka JNDI注入漏洞(CVE-2023-25194)】漏洞，该漏洞可允许远程代码执行，当攻击者可控制kafka-clients连接时的属性，可通过设置 ’ sasl.jaas.config ’ 属性为 ’ com.sun.security.auth.module.JndiLoginModule ’ 进行JNDI注入或反序列化利用，当JDK版本过低或者存在Gadgets时可导致远程代码执行。现场版本kafka_2.13-2.8.0，java version “1.8.0_361”，sasl.jaas.config 配置采用：
☬ 漏洞复现： 执行创建文件，相关安全软件会报：JNDI注入的告警；影响版本： 2.3.0 至 3.3.2 版本Kafka Connect，原则上不影响 Kafka server (broker)，但是会级联影响，最好还是升级到3.4.0及以上版本，升级JDK版本，可采用OpenJDK替换，相关经验已验证：OpenJDK1.8.0_362 + Zookeeper3.6.3 + Kafka3.4.0。
关联资源：官网升级指导、kafka部署快走
二、软件升级
本次软件要升级到3.4.0版本，版本说明见Kafka - Version 3.4.0，升级步骤也可参考官网升级指导。注意这里咱们是从2.1.x升级到3.x，不同于3.x升级，需注意存储 consumer offsets的schema和inter.broker.protocol.version里的版本，一旦升级后不支持降级。Apache Kafka 3.4.0以来，新增了一个系统属性：org.apache.kafka.disallowed.login.modules，用来在SASL JAAS配置中禁用有问题的登录模块，另外默认com.sun.security.auth.module.JndiLoginModule 在该版本中被禁用；另外需注意的是， Kafka 3.0中 Java 8 已注明废弃， 在Apache Kafka 4.0将不再支持；如果启用TLS，Java 11及更高版本的性能会明显更好，因此强烈建议使用它们。对应的zk版本稳定版为 3.5，注意zk需要足够的堆空间（3-5G，看数据量大小）；另zk集群不宜过大，尤其是在写操作频繁的使用模式中，意味着会造成大量的集群内通信（写操作和随后的集群成员更新的配额），尽量让ZooKeeper系统尽可能小，并尽可能保持其独立性，以处理负载。
1）滚动升级步骤
2）关停替换升级
即： 如果可接受停机，可将所有broker关闭后，更新版本替换后重启启动。
3）升级计划或流程
测试方案：
2.1、软件下载
#MD5： CF 6B 8B 1C A1 12 9E 69 41 39 92 99 B6 CC 47 8C
wget https://archive.apache.org/dist/kafka/3.4.0/kafka_2.12-3.4.0.tgz
md5sum kafka_2.12-3.4.0.tgz #输出
cf6b8b1ca1129e6941399299b6cc478c ./kafka_2.12-3.4.0.tgz
2.2、单节点/实例kafka升级
注意： 停止 Kafka 集群时，一定要等 Kafka 所有节点进程全部停止后再停止 Zookeeper 集群。因为 Zookeeper 集群当中记录着 Kafka 集群相关信息，Zookeeper 集群一旦先停止，Kafka 集群就没有办法再获取停止进程的信息，只能手动杀死 Kafka 进程。特别注意的是，broker之间的通讯协议和message的传输协议要与旧版本的一致，否则升级完的broker会因为通讯协议版本不一致导致节点一直报错（Connection to “broker id” was disconnected before the response was read），即升级新版本需现将旧版本的信息写入新版的配置文件中，以兼容当前（旧版）环境适配。从2.6.0版开始，对于Java 11或更新版本，TLSv1.3是默认启用的。客户机和服务器将协商是否支持TLSv1.3，否则将退回到TLSv1.2。
#1、解压缩新版本kafka
tar -xzf kafka_2.12-3.4.0.tgz
cd ./kafka_2.12-3.4.0/config
#2、修改新旧版配置文件，添加版本参数
vim config/server.properties #新增
inter.broker.protocol.version=0.10.1 #旧版本号
#3、旧版查看topic
bin/kafka-topics.sh --bootstrap-server 10.100.1.94:9092 --list
2.3、剩余2节点kafka升级
2.4、集群状态确认
2.5、kafka性能测试
#使用 8 个线程向名为 test-update-perf 的主题发送 500000 条大小为 50000 字节的消息，并将性能统计信息写入位于 ./perf-test 目录中的 CSV 文件中。性能统计信息将每 3000 毫秒报告一次。
./bin/kafka-producer-perf-test.sh --messages 500000 --message-size 50000 --topic test-update-perf --threads 8 --broker-list * —show-detailed-stats --csv-reporter-enabled --metrics-dir ./perf-test --reporting-interval 3000
#参数说明
--messages 500000：指定要发送的消息总数。
--message-size 50000：指定要发送的每条消息的大小（以字节为单位）。
--topic test-update-perf：指定要发送消息的目标主题。
--threads 8：指定用于发送消息的线程数。
--broker-list *：指定 Kafka 集群中所有代理的列表。星号* 表示所有代理。
--show-detailed-stats：启用详细统计信息的显示，包括每个线程的统计信息。
--csv-reporter-enabled：启用 CSV 报告程序，它将性能统计信息写入 CSV 文件。
--metrics-dir ./perf-test：指定用于存储 CSV 报告程序输出的目录。
--reporting-interval 3000：指定性能统计信息报告的间隔（以毫秒为单位）
#示例2：perf-consumer-t4单线程向test-update-perf主题请求消费500000 条大小为 50000 字节的消息，测试性能
/bin/kafka-consumer-perf-test.sh --topic test-update-perf --zookeeper 10.100.1.94:2183 --threads 1 --group perf-consumer-t4 --message-size 50000 --messages 10
三、附录
3.1、Kafka消息发送流程
kafka在消息发送的过程中，涉及到两个线程：main 线程和 Sender 线程。其中，main 线程中会创建了一个队列 RecordAccumulator，main 线程将消息发送给 RecordAccumulator；Sender 线程则不断从 RecordAccumulator 中拉取消息发送到 Kafka Broker。示意如下：
3.2、kafka应用场景
3.3、kafka架构回顾
3.4、zookeeper中kafka信息结构
3.5、Kafka Broker 总工作流程
3.6、消费者组消费流程
