flink 1.13.1 Kafka producer报错ByteArraySerializer is not an instance of org.apache.kafka.common.serialization.Serializer答案

【问题标题】：flink 1.13.1 Kafka producer error ByteArraySerializer is not an instance of org.apache.kafka.common.serialization.Serializerflink 1.13.1 Kafka producer报错ByteArraySerializer is not an instance of org.apache.kafka.common.serialization.Serializer
【发布时间】：2021-09-12 17:28:12
【问题描述】：

我正在尝试像下面这样使用 flink kafka 生产者

import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.ByteArraySerializer;

public static FlinkKafkaProducer<SelfDescribingMessageDO> createProducer() {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "<Server details>");
        props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
        props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());

        return new FlinkKafkaProducer<>(
                "FlinkSdmKafkaTopic",                  
                new SerializationSchema("FlinkSdmKafkaTopic", 8),
                props,                  
                FlinkKafkaProducer.Semantic.AT_LEAST_ONCE);
    }

private static class SerializationSchema implements KafkaSerializationSchema<SelfDescribingMessageDO> {
    final String topic;
    final int numPartitions;

    public SerializationSchema(final String topic, final int numPartitions) {
        this.topic = topic;
        this.numPartitions = numPartitions;
    }

    @Override
    public ProducerRecord<byte[], byte[]> serialize(SelfDescribingMessageDO sdm, @Nullable Long aLong) {
        return new ProducerRecord<>(topic,
                KafkaPublisher.getPartitionId(sdm.getHashKey(), numPartitions),
                sdm.getHashKey().getBytes(StandardCharsets.UTF_8),
                sdm.toByteArray());
    }
}

尝试部署 flink 作业时出现以下异常。在单元测试期间，我没有收到此错误。

2021-08-23T14:47:55.504Z WARN runtime.taskmanager.Task Source: MetricSource -> Filter -> MetricStoreMapper -> (Filter -> Timestamps/Watermarks -> Map -> Flat Map, Sink: FlinkKafkaProducer11, Sink: TSDBSink14) (5/8)#0 transitionState:1069 Source: MetricSource -> Filter -> MetricStoreMapper -> (Filter -> Timestamps/Watermarks -> Map -> Flat Map, Sink: FlinkKafkaProducer11, Sink: TSDBSink14) (5/8)#0 (5764a387ede7d6710bcf3ad4e2222248) switched from INITIALIZING to FAILED with failure cause: org.apache.kafka.common.KafkaException: Failed to construct kafka producer
        at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:432)
        at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:298)
        at org.apache.flink.streaming.connectors.kafka.internals.FlinkKafkaInternalProducer.<init>(FlinkKafkaInternalProducer.java:77)
        at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.createProducer(FlinkKafkaProducer.java:1230)
        at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.initProducer(FlinkKafkaProducer.java:1346)
        at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.initNonTransactionalProducer(FlinkKafkaProducer.java:1342)
        at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.beginTransaction(FlinkKafkaProducer.java:990)
        at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.beginTransaction(FlinkKafkaProducer.java:99)
        at org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction.beginTransactionInternal(TwoPhaseCommitSinkFunction.java:403)
        at org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction.initializeState(TwoPhaseCommitSinkFunction.java:394)
        at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.initializeState(FlinkKafkaProducer.java:1195)
        at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.tryRestoreFunction(StreamingFunctionUtils.java:189)
        at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.restoreFunctionState(StreamingFunctionUtils.java:171)
        at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.initializeState(AbstractUdfStreamOperator.java:96)
        at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.initializeOperatorState(StreamOperatorStateHandler.java:118)
        at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:290)
        at org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:436)
        at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:582)
        at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.call(StreamTaskActionExecutor.java:100)
        at org.apache.flink.streaming.runtime.tasks.StreamTask.executeRestore(StreamTask.java:562)
        at org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:647)
        at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:537)
        at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:759)
        at org.apache.flink.runtime.taskmanager.Task.run(Task.java:566)
        at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: org.apache.kafka.common.KafkaException: class org.apache.kafka.common.serialization.ByteArraySerializer is not an instance of org.apache.kafka.common.serialization.Serializer
        at org.apache.kafka.common.config.AbstractConfig.getConfiguredInstance(AbstractConfig.java:374)
        at org.apache.kafka.common.config.AbstractConfig.getConfiguredInstance(AbstractConfig.java:392)
        at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:359)

我在应用程序代码中添加了 flink 连接器 kafka 作为我的依赖项。

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka_2.12</artifactId>
    <version>1.13.1</version>
</dependency>

我的 pom 中只有 flink-connector-kafka 作为非测试依赖项（对于 kafka）。

ubuntu@vrni-platform:~$ mvn dependency:tree | grep -i kafka

[INFO] +- org.apache.flink:flink-connector-kafka_2.12:jar:1.13.1:compile
[INFO] |  +- org.apache.kafka:kafka-clients:jar:2.4.1:compile
[INFO] +- org.apache.kafka:kafka_2.12:jar:2.4.1:test
[INFO] +- org.apache.flink:flink-connector-kafka_2.12:test-jar:tests:1.13.1:test
[INFO] +- net.mguenther.kafka:kafka-junit:jar:2.4.0:test
[INFO] |  +- org.apache.kafka:kafka_2.11:jar:2.4.0:test
[INFO] |  +- org.apache.kafka:kafka_2.11:jar:test:2.4.0:test
[INFO] |  +- org.apache.kafka:kafka-clients:jar:test:2.4.0:test
[INFO] |  +- org.apache.kafka:connect-api:jar:2.4.0:test
[INFO] |  +- org.apache.kafka:connect-json:jar:2.4.0:test
[INFO] |  \- org.apache.kafka:connect-runtime:jar:2.4.0:test
[INFO] |     +- org.apache.kafka:kafka-tools:jar:2.4.0:test
[INFO] |     |  +- org.apache.kafka:kafka-log4j-appender:jar:2.4.0:test
[INFO] |     +- org.apache.kafka:connect-transforms:jar:2.4.0:test

有人可以建议可能出了什么问题吗？

【问题讨论】：

你正在混合kafka_2.12和kafka_2.11
Kafka 2.11 处于测试依赖关系中，我在测试期间没有收到此错误。

标签： java apache-kafka apache-flink

【解决方案1】：

两个思路可以参考

想法 1 可能是依赖问题，如果集群flink/lib下已经有flink-connector-kafka.jar，则将pom中kafka的connector依赖改为provider

想法 2 和flink的类加载方式有关。 flink-conf.yml 中 classloader.resolve-order 的默认值孩子优先，你可以试试改成父母优先

【讨论】：

不适合我。