Canal写kafka的逻辑没有考虑kafka的message size限制?
canal/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java
Lines 104 to 113 in e4b6385
if (canalDestination.getPartition() != null) { try { ProducerRecord<String, String> record = new ProducerRecord<String, String>(canalDestination .getTopic(), canalDestination.getPartition(), null, JSON.toJSONString(flatMessage)); producer2.send(record); } catch (Exception e) { logger.error(e.getMessage(), e); // producer.abortTransaction(); callback.rollback(); }
如果需要发送的record大小超过kafka限制(默认1MB), 是不是无法继续处理?
原提问者GitHub用户firnsan
写回答
-
canal batchSize配置不要超过1M
servers: localhost:9092 #for rocketmq: means the nameserver retries: 0 batchSize: 16384 lingerMs: 1 bufferMemory: 33554432
Canal的batch size, 默认50K, 由于kafka最大消息体限制请勿超过1M(900K以下)
canalBatchSize: 50
原回答者GitHub用户rewerma
2023-05-09 19:05:07赞同 展开评论 打赏
版权声明：本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
