Flink消费Kafka报错：ByteArrayDeserializer非Deserializer实例排查

问题根源 #

这个错误是类加载冲突或版本不兼容导致的：Flink Kafka连接器依赖的Kafka客户端版本，与Docker环境中运行的Kafka版本不匹配，使得ByteArrayDeserializer被不同类加载器加载，无法被识别为Deserializer接口的实现类。

解决方案 #

1. 对齐Kafka与Flink连接器版本 #

确保Flink使用的Kafka连接器版本和Kafka服务版本严格匹配：

• 例如Kafka用2.8.x，对应Flink 1.15.x的连接器为flink-connector-kafka-1.15.2_2.12，同时搭配kafka-clients-2.8.1.jar。

修改docker-compose.yml，给Flink的jobmanager和taskmanager挂载对应版本的依赖包：

services:
  jobmanager:
    image: flink:1.15.2-scala_2.12
    volumes:
      - ./flink-connector-kafka-1.15.2_2.12.jar:/opt/flink/lib/flink-connector-kafka-1.15.2_2.12.jar
      - ./kafka-clients-2.8.1.jar:/opt/flink/lib/kafka-clients-2.8.1.jar
    # 其他原有配置...
  taskmanager:
    image: flink:1.15.2-scala_2.12
    volumes:
      - ./flink-connector-kafka-1.15.2_2.12.jar:/opt/flink/lib/flink-connector-kafka-1.15.2_2.12.jar
      - ./kafka-clients-2.8.1.jar:/opt/flink/lib/kafka-clients-2.8.1.jar
    # 其他原有配置...

2. 修正PyFlink消费者的反序列化配置 #

不要依赖默认的Kafka反序列化器，改用Flink提供的Schema类，避免类加载问题：

• 如果生产者发送的是字符串/JSON数据，用SimpleStringSchema：

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors import FlinkKafkaConsumer
from pyflink.common.serialization import SimpleStringSchema

env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)

consumer = FlinkKafkaConsumer(
    'topic-4',
    SimpleStringSchema(),
    {
        'bootstrap.servers': 'kafka:9092',
        'group.id': 'flink-consumer-group'
    }
)

ds = env.add_source(consumer)
ds.print()
env.execute('Kafka Consumer Job')

• 如果发送的是二进制数据，用ByteArraySchema替代手动指定Kafka的ByteArrayDeserializer：

from pyflink.common.serialization import ByteArraySchema

consumer = FlinkKafkaConsumer(
    'topic-4',
    ByteArraySchema(),
    {'bootstrap.servers': 'kafka:9092', 'group.id': 'flink-consumer-group'}
)

3. 清理容器内冲突依赖 #

进入Flink TaskManager容器，删除旧版本的Kafka客户端jar包：

docker exec -it <taskmanager-container-id> bash
rm /opt/flink/lib/kafka-clients-*.jar
exit
# 重启容器生效
docker restart <taskmanager-container-id> <jobmanager-container-id>

4. 验证生产者数据格式 #

确保Python生产者发送的数据格式与消费者的Schema匹配，比如用kafka-python发送JSON字符串：

from kafka import KafkaProducer
from faker import Faker
import json

fake = Faker()
producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

for _ in range(10):
    producer.send('topic-4', {
        'name': fake.name(),
        'email': fake.email(),
        'address': fake.address()
    })
producer.flush()

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