Debezium/Kafka Connect能否忽略或调整过大的binlog消息？

问题场景 #

使用AWS MSK Serverless + MSK Connect + Debezium监控RDS Aurora MySQL的binlog时，偶尔出现以下错误：

[Worker-0ee2268e7e753a81a] org.apache.kafka.common.errors.RecordTooLargeException: The message is 12472229 bytes when serialized which is larger than 8388608, which is the value of the max.request.size configuration.

MSK Serverless的最大消息大小固定为8MB，无法通过调整Broker配置解决，当前只能手动更新Debezium偏移量跳过事件，效率低下且有风险。已尝试配置event.processing.failure.handling.mode=warn、connector.client.config.override.policy=ALL、producer.override.max.request.size=8388608，但未解决问题。

可行解决方案 #

1. 启用消息压缩减小事件体积 #

在Debezium连接器配置中添加Producer端压缩配置，通过压缩将大事件体积压缩至8MB以内：

producer.override.compression.type=lz4
# 可选压缩算法：snappy（平衡性能与压缩率）、gzip（高压缩率但性能稍低）

Debezium事件多为JSON格式文本，压缩率通常较高，能有效降低大事件的体积。

2. 使用Kafka Connect转换器过滤超大事件 #

通过Filter转换器拦截并跳过序列化后超过8MB的事件，无需自定义代码：

transforms=filterLargeEvents
transforms.filterLargeEvents.type=org.apache.kafka.connect.transforms.Filter
transforms.filterLargeEvents.condition=size($value) < 8388608

该配置利用Kafka Connect表达式语言判断消息值的字节大小，自动过滤超出限制的事件。

3. 缩小事件捕获范围减少单事件体积 #

• 排除大字段：若事件过大是因表中包含TEXT/BLOB等超大字段，通过column.exclude.list排除无需监控的大字段：

  column.exclude.list=db_name.table_name.large_blob_col,db_name.table_name.large_text_col
  

• 拆分大事务：若为大事务导致批量事件总大小超标，调整Debezium的批次参数拆分事务：

  max.batch.size=1000
  max.queue.size=2000
  

  注：此方法仅针对批量事件总大小，无法解决单个事件本身超标的情况。

4. 自定义错误处理逻辑跳过失败事件 #

若上述方法无法覆盖场景，可自定义Kafka Connect错误处理类，当捕获到RecordTooLargeException时自动跳过对应事件。需实现ErrorHandlingLogic接口或扩展Debezium的故障处理策略，适合有开发能力的场景。

为何原有配置未生效 #

• event.processing.failure.handling.mode=warn仅处理Debezium解析binlog阶段的错误（如格式异常），无法覆盖Producer发送到Broker时触发的RecordTooLargeException。
• producer.override.max.request.size=8388608仅设置Producer允许发送的最大消息大小，与MSK Serverless限制一致，无法解决事件本身超标的问题。

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