问题

生产环境下发现某个数据库中的某张表同步。

由于我们使用的是canal进行数据库的同步。

canal中针对kafka的主要配置如下:

该配置定义了根据库及表名,发送数据到指定的分区。最大分区数为10。

经过研究发现,未同步的表会将数据发送至分区5。至此发现主要的队列的分区5已不存在。

排查

使用命令查看kafka的分区信息:

分区信息如结果如下:

5号分区的Leader未选举出来。原因是brokers为1006的机器,已禁用kafka服务导致。

解决

  1. 分区信息修改:在zookeeper中查看分区信息,并修改分区至指定的broker,例如1002。
  2. 数据手工迁移:查看1002的机器,并将原1006机器下的分区文件夹拷贝至1002机器下
  3. 重启kafka服务
  4. 重分配分区

分区信息修改

使用zookeeper的命令行zkCli.sh查看分区信息:

结果如下:

修改其中的leader1003,并将leader_epoch+1。再设置回zookeeper:

数据手工迁移

在原1006机器上面的kafka配置中存储的日志目录,查找该分区的5号分区,并使用scp命令拷贝至1003机器上指定的位置。

一般数据的存储都在配置log.dirs配置。

至此重启服务。

重分配分区

编辑文件reassign.json

使用以下命令进行分区重分配:

使用以下命令查看进度:

成功后结果如下:

如果数据较多,同步时间过长,可以通过以下命令清除一定时间范围以前的数据,以下命令指示清除1天前的数据:

在增加该配置后,可以使用kafka-topics.sh --describe命令查看topic时,会展示在Configs处。

在迁移完成后,再使用以下命令,清除上述命令:

retention.ms该配置是topic级别的配置。在默认情况下,主题的数据有效期由kafka配置中的log.retention.hours控制。