如何处理非static的KafkaPartitioner类对象构造FlinkKafkaProducer010运行时报错

问题

Flink内核升级到1.3.0之后，当Kafka调用带有非static的KafkaPartitioner类对象为参数的FlinkKafkaProducer010去构造函数时，运行时会报错。

报错内容如下：

org.apache.flink.api.common.InvalidProgramException: The implementation of the FlinkKafkaPartitioner is not serializable. The object probably contains or references non serializable fields.

回答

Flink的1.3.0版本，为了兼容原有那些使用KafkaPartitioner的API接口，如FlinkKafkaProducer010带KafkaPartitioner对象的构造函数，增加了FlinkKafkaDelegatePartitioner类。

该类定义了一个成员变量，即kafkaPartitioner：

private final KafkaPartitioner<T> kafkaPartitioner;

当Flink传入参数是KafkaPartitioner去构造FlinkKafkaProducer010时，调用栈如下：

FlinkKafkaProducer010(String topicId, KeyedSerializationSchema<T> serializationSchema, Properties producerConfig, KafkaPartitioner<T> customPartitioner)
->  FlinkKafkaProducer09(String topicId, KeyedSerializationSchema<IN> serializationSchema, Properties producerConfig, FlinkKafkaPartitioner<IN> customPartitioner)
---->  FlinkKafkaProducerBase(String defaultTopicId, KeyedSerializationSchema<IN> serializationSchema, Properties producerConfig, FlinkKafkaPartitioner<IN> customPartitioner)
------>  ClosureCleaner::clean(Object func, boolean checkSerializable)