自定义Kafka Partitioner处理JSON消息时出现ClassCastException异常

报错java.lang.ClassCastException: [B cannot be cast to java.lang.String是因为Kafka生产者的序列化器已经将消息value序列化为字节数组，传入partition方法的Object value实际为byte[]类型，无法直接强转为String。
你无需对value对象做类型转换，partition方法已经提供了序列化后的字节数组参数byte[] valueBytes，直接使用该参数转字符串即可。

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;
import org.json.JSONObject;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;

public class JsonPartitioner implements Partitioner {

    @Override
    public void configure(Map<String, ?> configs) {
        // 无自定义配置需求可留空
    }

    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        int defaultPartition = 0;
        try {
            // 直接通过valueBytes转UTF-8字符串，避免强转value的异常
            String jsonValueStr = new String(valueBytes, StandardCharsets.UTF_8).toLowerCase();
            JSONObject jsonObj = new JSONObject(jsonValueStr);

            // 以下为自定义分区逻辑示例，可根据业务调整
            List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
            int partitionCount = availablePartitions.size();
            if (partitionCount <= 0) {
                partitionCount = cluster.partitionsForTopic(topic).size();
            }
            // 例：按age字段分区
            int age = jsonObj.getInt("age");
            return Math.abs(age % partitionCount);

        } catch (Exception e) {
            // 格式解析异常时返回默认分区，避免生产者崩溃
            e.printStackTrace();
            return defaultPartition;
        }
    }

    @Override
    public void close() {
        // 无资源释放需求可留空
    }
}

• 确保生产者发送的value序列化编码为UTF-8，和代码中解码编码保持一致，避免JSON解析乱码失败
• 原代码存在变量名笔误：定义的字符串变量为jsonStringValue，调用new JSONObject()时错误传入了未定义的jsonString变量，已在修正代码中处理
• 必须添加全局异常捕获，避免非法格式消息导致生产者发送流程崩溃

内容的提问来源于stack exchange，提问作者Lav