FlinkKafkaProducer09未按指定属性实现Kafka分区选择问题

针对你遇到的FlinkKafkaProducer09发送消息全部进入同一Kafka分区的问题，结合Flink 1.1.4版本的特性，可从以下几个方向排查解决：

1. 确认Kafka主题分区数是否符合预期 #

首先检查目标主题的分区数，若分区数为1，无论key如何哈希，所有消息都会进入同一分区。使用Kafka命令行工具查看：

kafka-topics.sh --describe --topic myTopic --zookeeper <zk-host>:2181

若分区数不足，需扩容主题分区：

kafka-topics.sh --alter --topic myTopic --partitions <目标分区数> --zookeeper <zk-host>:2181

2. 验证Key的生成逻辑是否正确 #

检查myImplementationOfKeyedSerializationSchema中serializeKey方法生成的key是否确实唯一且不同：

• 在serializeKey中添加日志，打印每个事件的key值，确认50种不同属性类型的事件生成的key存在差异：

@Override
public byte[] serializeKey(JsonObject event) {
    String keyStr = event.get(messageKey).toString();
    System.out.println("Generated key: " + keyStr); // 或使用日志框架输出
    return keyStr.getBytes();
}

若key存在重复或异常（如所有key为空字符串），需修正属性获取逻辑。

3. 显式指定基于Key的分区策略 #

Flink 1.1.4的FlinkKafkaProducer09默认虽调用Kafka的DefaultPartitioner，但可能存在版本兼容问题。可手动实现FlinkKafkaPartitioner，强制基于key哈希分配分区：

FlinkKafkaProducer09<JsonObject> myProducer = new FlinkKafkaProducer09<>(
    myTopic,
    new myImplementationOfKeyedSerializationSchema("attributeNameToUseForPartition"),
    kafkaproperties,
    new FlinkKafkaPartitioner<JsonObject>() {
        @Override
        public int partition(JsonObject record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
            if (key == null) {
                // 无key时随机分配
                return ThreadLocalRandom.current().nextInt(partitions.length);
            }
            // 复用Kafka DefaultPartitioner的哈希逻辑
            return Math.abs(org.apache.kafka.common.utils.Utils.murmur2(key)) % partitions.length;
        }
    }
);

4. 检查Sink并行度与分区绑定逻辑 #

你设置了sink并行度为1，这本身不会导致所有消息进入同一分区（Kafka生产者会基于key哈希选分区），但需注意：若Flink版本存在bug，可能强制将单个并行实例绑定到固定分区。若上述方案无效，可尝试将sink并行度调整为与Kafka主题分区数一致，再观察分区分配情况。

5. 版本兼容性考量 #

Flink 1.1.4是非常老旧的版本（发布于2017年），存在较多已知问题。若业务允许，建议升级到Flink 1.13+或更高的稳定版本，新版本对Kafka生产者的分区逻辑支持更完善。

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