adapter1.1.3，syncHbase，在RDS批量删除数据时只删掉了第一条，后面的未删除

[jinguangyu]

environment

• canal version 1.1.3
• mysql version rds5.7

Issue Description

执行下面的删除语句，
DELETE FROM test_bin_log WHERE id >= 20

打印的DML日志如下，发现Hbase中只删掉了id=20的记录，之后的记录未被删除！
2019-06-19 11:56:15.991 [pool-7-thread-1] DEBUG c.a.o.c.client.adapter.hbase.service.HbaseSyncService - DML: {"data":[{"id":20,"name":"q1","birthday":1560916508000},{"id":21,"name":"q2","birthday":1560916518000},{"id":22,"name":"q3","birthday":1560916526000},{"id":24,"name":"q4","birthday":1560916536000}],"database":"metrics","destination":"example","es":1560916575000,"groupId":null,"isDdl":false,"old":null,"pkNames":["id"],"sql":"","table":"test_bin_log","ts":1560916575986,"type":"DELETE"}

@agapple @rewerma

[yuanx1]

@jinguangyu
/**
* 批量删除数据
*
* @param tableName 表名
* @param rowKeys rowKey集合
* @return 是否成功
*/
public Boolean deletes(String tableName, Set<byte[]> rowKeys) {
boolean flag = false;
try {
HTable table = (HTable) getConnection().getTable(TableName.valueOf(tableName));
List deletes = new ArrayList<>();
for (byte[] rowKey : rowKeys) {
Delete delete = new Delete(rowKey);
deletes.add(delete);
}
if (!deletes.isEmpty()) {
table.delete(deletes);
}
flag = true;
} catch (Exception e) {
logger.error(e.getMessage(), e);
throw new RuntimeException(e);
}
return flag;
}

[yuanx1]

@jinguangyu 用的是这个，如果是的话，从代码上看，没有问题啊～！

[yuanx1]

@jinguangyu
/**

• HBase操作模板
• @author machengyuan 2018-8-21 下午10:12:34
• @Version 1.0.0
  */
  public class HbaseTemplate {
  代码位置

[jinguangyu]

@agapple ,@rewerma,@yuanx1
发现update存在同样问题，
UPDATE test_bin_log SET name = 2 WHERE id >= 16

2019-06-19 20:36:03.330 [pool-7-thread-1] DEBUG c.a.o.c.client.adapter.hbase.service.HbaseSyncService - DML: {"data":[{"id":16,"name":"2","birthday":null},{"id":17,"name":"2","birthday":null},{"id":18,"name":"2","birthday":null},{"id":19,"name":"2","birthday":null}],"database":"metrics","destination":"example","es":1560947762000,"groupId":null,"isDdl":false,"old":[{"name":"1"},{"name":"1"},{"name":"1"},{"name":"1"}],"pkNames":["id"],"sql":"","table":"test_bin_log","ts":1560947763322,"type":"UPDATE"}

Hbase中只有id=16的记录被更新成功！

[yuanx1]

commitBatch = 5000; // etl等批量提交大小
5000判断
if (i % config.getHbaseMapping().getCommitBatch() == 0 && !rows.isEmpty()) {
hbaseTemplate.deletes(hbaseMapping.getHbaseTable(), rowKeys);

            hbaseTemplate.puts(hbaseMapping.getHbaseTable(), rows);
            rowKeys.clear();
            rows.clear();
            complete = true;

}
不是5000的倍数但是小于5000的情况
if (!complete && !rows.isEmpty()) {
hbaseTemplate.deletes(hbaseMapping.getHbaseTable(), rowKeys);
hbaseTemplate.puts(hbaseMapping.getHbaseTable(), rows);
}
@jinguangyu 当old和data的数量相等的并且又小于5000 时候代码又bug，只会删除或者更新第一条记录

[yuanx1]

@agapple @rewerma @jinguangyu 建议合并两个判断，就可以修复此bug

[yuanx1]

@agapple @rewerma @jinguangyu 其实可以简化操作，在循环里踢出或者更新数据，看了代码逻辑还是有点乱的～！可以优化～！

[yuanx1]

得到新的数据集合就可以了～！

[jinguangyu]

#1936

[yuanx1]

@jinguangyu @agapple 已经修复还是什么的，没有看懂，就关闭了～！

[jinguangyu]

@jinguangyu @agapple 已经修复还是什么的，没有看懂，就关闭了～！

代码有bug，1936我修复了