MySQL binlog 表名大小写 产生的kafka动态匹配规则错误

环境信息

canal version 1.1.3 mysql version 5.7

问题描述

在集成kafka的情况下，canal在拉取到binlog之后，匹配topic的时候，如果topic字母相同但大小写不同会引发kafka的日志恢复并resize，报错如下：（之前已有一个topic：redis_cachecloud_QRTZ_FIRED_TRIGGERS） ERROR Error while creating log for redis_cachecloud_qrtz_fired_triggers-0 in dir E:\kafka\logs\kafka2 (kafka.server.LogDirFailureChannel) java.io.IOException: 请求的操作无法在使用用户映射区域打开的文件上执行 at java.io.RandomAccessFile.setLength(Native Method) at kafka.log.AbstractIndex$$anonfun$resize$1.apply$mcZ$sp(AbstractIndex.scala:186) at kafka.log.AbstractIndex$$anonfun$resize$1.apply(AbstractIndex.scala:173) at kafka.log.AbstractIndex$$anonfun$resize$1.apply(AbstractIndex.scala:173) at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:251) at kafka.log.AbstractIndex.resize(AbstractIndex.scala:173) at kafka.log.AbstractIndex$$anonfun$trimToValidSize$1.apply$mcZ$sp(AbstractIndex.scala:242) at kafka.log.AbstractIndex$$anonfun$trimToValidSize$1.apply(AbstractIndex.scala:242) at kafka.log.AbstractIndex$$anonfun$trimToValidSize$1.apply(AbstractIndex.scala:242) at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:251) at kafka.log.AbstractIndex.trimToValidSize(AbstractIndex.scala:241) at kafka.log.LogSegment.recover(LogSegment.scala:377) at kafka.log.Log.kafka$log$Log$$recoverSegment(Log.scala:467) at kafka.log.Log.kafka$log$Log$$recoverLog(Log.scala:581) at kafka.log.Log$$anonfun$2.apply$mcJ$sp(Log.scala:552) at kafka.log.Log$$anonfun$2.apply(Log.scala:552) at kafka.log.Log$$anonfun$2.apply(Log.scala:552) at kafka.log.Log.retryOnOffsetOverflow(Log.scala:1938) at kafka.log.Log.loadSegments(Log.scala:551) at kafka.log.Log.(Log.scala:276) at kafka.log.Log$.apply(Log.scala:2071) at kafka.log.LogManager$$anonfun$getOrCreateLog$1.apply(LogManager.scala:691) at kafka.log.LogManager$$anonfun$getOrCreateLog$1.apply(LogManager.scala:659) at scala.Option.getOrElse(Option.scala:121) at kafka.log.LogManager.getOrCreateLog(LogManager.scala:659) at kafka.cluster.Partition$$anonfun$getOrCreateReplica$1.apply(Partition.scala:199) at kafka.cluster.Partition$$anonfun$getOrCreateReplica$1.apply(Partition.scala:195) at kafka.utils.Pool$$anon$2.apply(Pool.scala:61) at java.util.concurrent.ConcurrentHashMap.computeIfAbsent(ConcurrentHashMap.java:1660) at kafka.utils.Pool.getAndMaybePut(Pool.scala:60) at kafka.cluster.Partition.getOrCreateReplica(Partition.scala:194) at kafka.cluster.Partition$$anonfun$makeFollower$1$$anonfun$apply$mcZ$sp$3.apply(Partition.scala:439) at kafka.cluster.Partition$$anonfun$makeFollower$1$$anonfun$apply$mcZ$sp$3.apply(Partition.scala:439) at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48) at kafka.cluster.Partition$$anonfun$makeFollower$1.apply$mcZ$sp(Partition.scala:439) at kafka.cluster.Partition$$anonfun$makeFollower$1.apply(Partition.scala:431) at kafka.cluster.Partition$$anonfun$makeFollower$1.apply(Partition.scala:431) at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:251) at kafka.utils.CoreUtils$.inWriteLock(CoreUtils.scala:259) at kafka.cluster.Partition.makeFollower(Partition.scala:431) at kafka.server.ReplicaManager$$anonfun$makeFollowers$3.apply(ReplicaManager.scala:1244) at kafka.server.ReplicaManager$$anonfun$makeFollowers$3.apply(ReplicaManager.scala:1238) at scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:130) at scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:130) at scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:236) at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:40) at scala.collection.mutable.HashMap.foreach(HashMap.scala:130) at kafka.server.ReplicaManager.makeFollowers(ReplicaManager.scala:1238) at kafka.server.ReplicaManager.becomeLeaderOrFollower(ReplicaManager.scala:1076) at kafka.server.KafkaApis.handleLeaderAndIsrRequest(KafkaApis.scala:185) at kafka.server.KafkaApis.handle(KafkaApis.scala:110) at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:69) at java.lang.Thread.run(Thread.java:748)

步骤重现

既然数据库不区分大小写，那为何动态匹配规则的时候要弄成两个topic？可否修改？

期待结果

正确匹配

现执行情况

broker报错

我查看了binlog，确实是同一张表，但是记录却是大小写不一致，结果导致canal认为是两张表的修改，发到kafka的时候就会出现错误，后来我在canal的MQMessageUtils类里动态匹配topic那里修改了，直接全部通过String的toLowerCase()改为小写并清除kafka log，重启broker之后就能正常运行了

原提问者GitHub用户finefuture