PyFlink 1.15.2对接RabbitMQ Sink出现序列化注册异常求助

问题 #

使用Python开发PyFlink应用，版本为Flink 1.15.2，通过flink-sql-connector-rabbitmq-1.15.2.jar连接RabbitMQ读取数据，编写的Sink代码如下：

stream.add_sink(RMQSink(
    connection_config,      # RabbitMQ连接配置
    'queueName',            # 目标RabbitMQ队列名称
    SimpleStringSchema()))

运行时触发gRPC错误栈：

File "/home/ali/.virtualenvs/LAB_920_log_parser_more_investigation-DQLOhTET/lib/python3.8/site-packages/grpc/_channel.py", line 826, in _next
    raise self
grpc._channel._MultiThreadedRendezvous: <_MultiThreadedRendezvous of RPC that terminated with:
    status = StatusCode.CANCELLED
    details = "Multiplexer hanging up"
    debug_error_string = "{"created":"@1662371359.807069114","description":"Error received from peer ipv6:[::1]:44295","file":"src/core/lib/surface/call.cc","file_line":966,"grpc_message":"Multiplexer hanging up","grpc_status":1}"
>
Traceback (most recent call last):
  File "/home/ali/.virtualenvs/LAB_920_log_parser_more_investigation-DQLOhTET/lib/python3.8/site-packages/apache_beam/runners/worker/sdk_worker.py", line 289, in _execute
    response = task()

核心Java层错误为：

RuntimeError: java.lang.UnsupportedOperationException: A serializer has already been registered for the state; re-registration is not allowed.
    at org.apache.flink.runtime.state.StateSerializerProvider$EagerlyRegisteredStateSerializerProvider.registerNewSerializerForRestoredState(StateSerializerProvider.java:344)
    at org.apache.flink.runtime.state.RegisteredKeyValueStateBackendMetaInfo.updateNamespaceSerializer(RegisteredKeyValueStateBackendMetaInfo.java:132)

解决方案 #

• 替换RabbitMQ连接器依赖
  当前使用的flink-sql-connector-rabbitmq是面向SQL API的连接器，而DataStream API的RMQSink需要使用专门的flink-connector-rabbitmq依赖。两者实现逻辑不同，混用会引发状态序列化冲突，请替换为对应版本的flink-connector-rabbitmq-1.15.2.jar。

• 确保状态序列化器一致性
  作业重启或恢复时，状态的序列化器配置必须和之前完全一致。如果混用Table API与DataStream API，避免重复定义状态相关的序列化逻辑；检查算子中是否重复注册了相同的状态描述符（StateDescriptor）。

• 清理旧状态后重启作业
  若作业遗留了旧状态数据，新的序列化配置与旧状态不兼容会触发该错误。清理作业的状态存储（如RocksDB本地文件或分布式存储中的状态目录），以全新状态启动作业。

内容的提问来源于stack exchange，提问作者Ali Ait-Bachir