canal binlog 丢失

现网数据变更较大，发现数据存在遗漏的情况，定位在binlog未接收到事件信息。

对比发现binlog 29827255 未知的信息丢失

dump binlog

canal log

日志丢失的问题看源码找到原因了，但是binlog接收处确实未接收到，log每次会将接收到的binlog打印出来，29827255未被接收，处理代码如下

void execute() { long batchId; LOGGER.debug("execute destination : " + destination); while (true) { try { connector.connect(); connector.subscribe(filter); while (true) { Message message = connector.getWithoutAck(BATCH_SIZE); batchId = message.getId(); int size = message.getEntries().size(); if (batchId == -1 || size == 0) { try { Thread.sleep(SLEEP_TIME); } catch (InterruptedException ignored) { // ignored } } else { process(message.getEntries()); } connector.ack(batchId); } } catch (Exception e) { LOGGER.error("canal connect error, destination : " + destination, e); AlarmUtil.dcAlarm(App.DC_ID, "canal_connect_error", "canal connect error, destination : " + destination); } finally { connector.disconnect(); }

        try {
            Thread.sleep(1000 * 60);
        } catch (InterruptedException e) {
            // ignored
        }
    }
}


private void process(List<CanalEntry.Entry> entryList) {
    for (CanalEntry.Entry entry : entryList) {
        if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN
                || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
            continue;
        }

        if (entry.getEntryType() != CanalEntry.EntryType.ROWDATA) {
            continue;
        }

        CanalEntry.RowChange rowChange;
        try {
            rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
        } catch (Exception e) {
            throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
        }

        CanalEntry.EventType eventType = rowChange.getEventType();
        if (eventType == CanalEntry.EventType.DELETE) {
            return;
        }

        if (eventType != CanalEntry.EventType.INSERT && eventType != CanalEntry.EventType.UPDATE) {
            return;
        }

        List<CanalEntry.RowData> rowDataList = rowChange.getRowDatasList();
        if (rowDataList == null || rowDataList.size() == 0) {
            continue;
        }

        List<BinlogColumnDTO> columns = new ArrayList<>();
        try {
            for (CanalEntry.RowData rowData : rowDataList) {
                columns = convertColumnList(rowData.getAfterColumnsList());
                long updatedCount = columns.stream().filter(BinlogColumnDTO::getUpdated).count();
                if (updatedCount < 1) {
                    return;
                }

                String binlogFileOffset = entry.getHeader().getLogfileName() + ":"
                        + entry.getHeader().getLogfileOffset() + ":"
                        + DateUtil.timestamp2DateTime(entry.getHeader().getExecuteTime());

                LOGGER.log(ACCESS, binlogFileOffset + ", eventType : {}, data : {}",
                        eventType, toJsonString(columns));

                if (eventType == CanalEntry.EventType.INSERT) {
                    syncService.insert(columns, binlogFileOffset);
                }
                else if (eventType == CanalEntry.EventType.UPDATE){
                    syncService.update(columns, binlogFileOffset);
                }
            }
        }
        catch (Exception e) {
            LOGGER.error("binlog handle error, data : " + JSONArray.toJSONString(columns), e);
            AlarmUtil.dcAlarm(App.DC_ID, "binlog_handle_error", e.toString());
        }

    }
}

跟踪源码，canal server打印debug日志，发现client发请求过来的时候是有读取到对应postion的，但就是没有返回事件内容过去。

另一个案例的截图

binlog file

canal server log

原提问者GitHub用户xiaopan0513