@qingduyu 在 mongo-kaifa-同步失败 中说：

回复: mongo-kafka同步，没数据版本1.2.04

补充一下上下文

这是mongo-kafka的任务， 通过已有的，和自动创建的都是失败，

报错日志：

021-11-11 18:34:54.210 [main] ERROR c.c.c.mongodb.worker.position.MongoFullPosManager - Init table (riskmg.dtinDatasourceJnl)position failed with exception,msg:ClassCastException: Cannot cast java.lang.String to org.bson.Document
java.lang.ClassCastException: Cannot cast java.lang.String to org.bson.Document
at java.lang.Class.cast(Class.java:3369)
at org.bson.Document.get(Document.java:144)
at com.clougence.cloudcanal.mongodb.common.util.MongoMetaUtilImpl.queryMaxId(MongoMetaUtilImpl.java:108)
at com.clougence.cloudcanal.mongodb.common.util.MongoMetaUtilImpl.fetchNumericPkMinMaxVal(MongoMetaUtilImpl.java:82)
at com.clougence.cloudcanal.mongodb.common.util.MongoMetaUtilImpl.fetchNumericPkMinMaxVal(MongoMetaUtilImpl.java:23)
at com.clougence.cloudcanal.mongodb.worker.position.MongoFullPosManager.genInitialPos(MongoFullPosManager.java:84)
at com.clougence.cloudcanal.mongodb.worker.position.MongoFullPosManager.fillPositionMap(MongoFullPosManager.java:72)
at com.clougence.cloudcanal.mongodb.worker.position.MongoFullPosManager.initPositionMap(MongoFullPosManager.java:59)
at com.clougence.cloudcanal.ds.common.position.AbstractRdbFullPosManager.start(AbstractRdbFullPosManager.java:73)
at com.clougence.cloudcanal.task.service.impl.CanalFullServiceImpl.start(CanalFullServiceImpl.java:56)
at com.clougence.cloudcanal.task.DataTaskStarter.startDataTask(DataTaskStarter.java:246)
at com.clougence.cloudcanal.task.DataTaskStarter.start(DataTaskStarter.java:114)
at com.clougence.cloudcanal.task.TaskCoreApplication.main(TaskCoreApplication.java:49)
2021-11-11 18:34:54.217 [main] ERROR com.clougence.cloudcanal.task.DataTaskStarter - Start data task failed with exception. Root cause is ClassCastException: Cannot cast java.lang.String to org.bson.Document
com.clougence.cloudcanal.base.service.task.exception.runtime.PositionException: Init table (riskmg.dtinDatasourceJnl)position failed with exception,msg:ClassCastException: Cannot cast java.lang.String to org.bson.Document
at com.clougence.cloudcanal.mongodb.worker.position.MongoFullPosManager.genInitialPos(MongoFullPosManager.java:105)
at com.clougence.cloudcanal.mongodb.worker.position.MongoFullPosManager.fillPositionMap(MongoFullPosManager.java:72)
at com.clougence.cloudcanal.mongodb.worker.position.MongoFullPosManager.initPositionMap(MongoFullPosManager.java:59)
at com.clougence.cloudcanal.ds.common.position.AbstractRdbFullPosManager.start(AbstractRdbFullPosManager.java:73)
at com.clougence.cloudcanal.task.service.impl.CanalFullServiceImpl.start(CanalFullServiceImpl.java:56)
at com.clougence.cloudcanal.task.DataTaskStarter.startDataTask(DataTaskStarter.java:246)
at com.clougence.cloudcanal.task.DataTaskStarter.start(DataTaskStarter.java:114)
at com.clougence.cloudcanal.task.TaskCoreApplication.main(TaskCoreApplication.java:49)
Caused by: java.lang.ClassCastException: Cannot cast java.lang.String to org.bson.Document
at java.lang.Class.cast(Class.java:3369)
at org.bson.Document.get(Document.java:144)
at com.clougence.cloudcanal.mongodb.common.util.MongoMetaUtilImpl.queryMaxId(MongoMetaUtilImpl.java:108)
at com.clougence.cloudcanal.mongodb.common.util.MongoMetaUtilImpl.fetchNumericPkMinMaxVal(MongoMetaUtilImpl.java:82)
at com.clougence.cloudcanal.mongodb.common.util.MongoMetaUtilImpl.fetchNumericPkMinMaxVal(MongoMetaUtilImpl.java:23)
at com.clougence.cloudcanal.mongodb.worker.position.MongoFullPosManager.genInitialPos(MongoFullPosManager.java:84)
… 7 common frames omitted
2021-11-11 18:34:54.218 [main] ERROR com.clougence.cloudcanal.task.TaskCoreApplication - start task error.task id:10,msg:ClassCastException: Cannot cast java.lang.String to org.bson.Document
com.clougence.cloudcanal.base.service.task.exception.runtime.PositionException: Init table (riskmg.dtinDatasourceJnl)position failed with exception,msg:ClassCastException: Cannot cast java.lang.String to org.bson.Document
at com.clougence.cloudcanal.mongodb.worker.position.MongoFullPosManager.genInitialPos(MongoFullPosManager.java:105)
at com.clougence.cloudcanal.mongodb.worker.position.MongoFullPosManager.fillPositionMap(MongoFullPosManager.java:72)
at com.clougence.cloudcanal.mongodb.worker.position.MongoFullPosManager.initPositionMap(MongoFullPosManager.java:59)
at com.clougence.cloudcanal.ds.common.position.AbstractRdbFullPosManager.start(AbstractRdbFullPosManager.java:73)
at com.clougence.cloudcanal.task.service.impl.CanalFullServiceImpl.start(CanalFullServiceImpl.java:56)
at com.clougence.cloudcanal.task.DataTaskStarter.startDataTask(DataTaskStarter.java:246)
at com.clougence.cloudcanal.task.DataTaskStarter.start(DataTaskStarter.java:114)
at com.clougence.cloudcanal.task.TaskCoreApplication.main(TaskCoreApplication.java:49)
Caused by: java.lang.ClassCastException: Cannot cast java.lang.String to org.bson.Document
at java.lang.Class.cast(Class.java:3369)
at org.bson.Document.get(Document.java:144)
at com.clougence.cloudcanal.mongodb.common.util.MongoMetaUtilImpl.queryMaxId(MongoMetaUtilImpl.java:108)
at com.clougence.cloudcanal.mongodb.common.util.MongoMetaUtilImpl.fetchNumericPkMinMaxVal(MongoMetaUtilImpl.java:82)
at com.clougence.cloudcanal.mongodb.common.util.MongoMetaUtilImpl.fetchNumericPkMinMaxVal(MongoMetaUtilImpl.java:23)
at com.clougence.cloudcanal.mongodb.worker.position.MongoFullPosManager.genInitialPos(MongoFullPosManager.java:84)
… 7 common frames omitted
2021-11-11 18:34:54.220 [main] ERROR com.clougence.cloudcanal.task.TaskCoreApplication - try to exist with System.exit(-1)

============mongo 中的 范例 collection的某一值====================
{
“_id”: “600102S08022019060614030226137600102S080220190606140302261370000”,
“dsId”: “OGZT”,
“entNo”: “1”,
“intfId”: “OGZT004”,
“channel”: “0101”,
“channelDate”: “20190606”,
“sourceSystemId”: “6001”,
“cacheKey”: “FKCACHE:OGZT004:巫ZXXX,44088119822356910”,
“transStatus”: “4”,
“responseCode”: “DT01009”,
“responseMsg”: “数据源调用错误——原始返回报文转换出现错误”,
“oriRequestTime”: “2019-06-06 14:03:02”,
“oriResponseTime”: “2019-06-06 14:03:16”,
“convertStandMapTime”: “2019-06-06 14:03:16”,
“oriRequestMap”: {
“idCard”: “44088119860905234910”,
“customerName”: “巫查”
},
“oriResponseMsg”: “”,
“oriResponseMsgProvenance”: “1”,
“_class”: “com.veredholdings.dtin.ds.proxy.mongo.dto.DtinDatasourceJnl”
}

好的，mongodb 源端实际上还处于测试阶段–，不过我们跟踪下。

简述

CloudCanal 社区版为了体验便利，默认设置了 777777 验证码，但是有些用户希望更加正式地使用，这个就有安全风险。所以 CloudCanal 提供了修改默认验证码的配置，通过以下步骤即可。

步骤

• 进入 CloudCanal社区版解压目录/console_data/cloudcanal/console/conf 目录
• 编辑 business-output.properties 文件和以下参数

  console.config.product.trial.verify_code=777777
  

• 重启 console 容器

@navyaijm2017 在 canal无法写入kafka多分区 中说：

已解决，正确的配置

mq config

canal.mq.dynamicTopic=db_chj_test.sbtest2
#canal.mq.partition=3
canal.mq.partitionsNum=3
canal.mq.partitionHash=db_chj_test.sbtest2:id

补充下。

canal.mq.partition 这个参数实际上表示的是在 canal.mq.partitionsNum 和 canal.mq.partitionHash 未设置的情况下，默认写入的partition。以下是参考源码

 for (FlatMessage flatMessage : flatMessages) {
                /**canal.mq.partitionsNum和canal.mq.partitionHash **/
                if (mqDestination.getPartitionHash() != null && !mqDestination.getPartitionHash().isEmpty()) {
                    FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage,
                        mqDestination.getPartitionsNum(),
                        mqDestination.getPartitionHash(),
                        this.mqProperties.isDatabaseHash());
                    int length = partitionFlatMessage.length;
                    for (int i = 0; i < length; i++) {
                        FlatMessage flatMessagePart = partitionFlatMessage[i];
                        if (flatMessagePart != null) {
                            records.add(new ProducerRecord<>(topicName, i, null, JSON.toJSONBytes(flatMessagePart,
                                SerializerFeature.WriteMapNullValue)));
                        }
                    }
                } else {
                    /**canal.mq.partition**/
                    final int partition = mqDestination.getPartition() != null ? mqDestination.getPartition() : 0;
                    records.add(new ProducerRecord<>(topicName, partition, null, JSON.toJSONBytes(flatMessage,
                        SerializerFeature.WriteMapNullValue)));
                }
            }

功能需求 : 源端没有任何变更事件时，需要有心跳支持，让任务保持追上状态，而不是延迟
应用场景 : 很少有变更的源端库普遍存在这类问题
推荐理由 :
状况：1.0.3 对于 mysql 官方 heartbeat 包进行了响应，可让任务保持追上状态(空闲时)

@hufengjie 在 Channel被挂起，查看日志是”this rowdata has no pks“ 中说：

pid:1 nid:1 exception:setl:com.alibaba.otter.node.etl.select.exceptions.SelectException: com.alibaba.otter.node.etl.select.exceptions.SelectException: this rowdata has no pks , entry: header {
version: 1
logfileName: “mysql-bin.000001”
logfileOffset: 2424093
serverId: 1
serverenCode: “UTF-8”
executeTime: 1615806816000
sourceType: MYSQL

从 otter 代码来看，不支持无主键同步。如果表是有主键的，可能需要debug 下在哪一步把主键给丢了。

  private EventData internParse(Pipeline pipeline, Entry entry, RowChange rowChange, RowData rowData) {
        EventData eventData = new EventData();
        eventData.setTableName(entry.getHeader().getTableName());
        eventData.setSchemaName(entry.getHeader().getSchemaName());
        eventData.setEventType(EventType.valueOf(rowChange.getEventType().name()));
        eventData.setExecuteTime(entry.getHeader().getExecuteTime());
        EventType eventType = eventData.getEventType();
        TableInfoHolder tableHolder = null;

        if (!StringUtils.equalsIgnoreCase(pipeline.getParameters().getSystemSchema(), eventData.getSchemaName())) {
            boolean useTableTransform = pipeline.getParameters().getUseTableTransform();
            Table table = null;
            DataMediaPair dataMediaPair = ConfigHelper.findDataMediaPairBySourceName(pipeline,
                eventData.getSchemaName(),
                eventData.getTableName());
            DataMedia dataMedia = dataMediaPair.getSource();
            eventData.setTableId(dataMedia.getId());
            // 获取目标表
            DataMedia targetDataMedia = dataMediaPair.getTarget();
            if (useTableTransform || dataMedia.getSource().getType().isOracle()) {// oracle需要反查一次meta
                // 如果设置了需要进行table meta转化，则反查一下table信息
                // 比如oracle erosa解析时可能使用了非物理主键，需要直接使用，信任erosa的信息
                DbDialect dbDialect = dbDialectFactory.getDbDialect(pipeline.getId(),
                    (DbMediaSource) dataMedia.getSource());
                table = dbDialect.findTable(eventData.getSchemaName(), eventData.getTableName());// 查询一下meta信息
                if (table == null) {
                    logger.warn("find table[{}.{}] is null , may be drop table.",
                        eventData.getSchemaName(),
                        eventData.getTableName());
                }
                // 获取一下目标库的拆分字段,设置源表为主键
                // 首先要求源和目标的库名表名是一致的
                DataMediaSource targetSource = targetDataMedia.getSource();
                if (targetSource instanceof DbMediaSource
                    && StringUtils.containsIgnoreCase(((DbMediaSource) targetSource).getUrl(), "drds")) {
                    // 优先判断是否为drds
                    DbDialect targetDbDialect = dbDialectFactory.getDbDialect(pipeline.getId(),
                        (DbMediaSource) targetDataMedia.getSource());
                    if (targetDbDialect.isDRDS()) {
                        String schemaName = buildName(eventData.getSchemaName(),
                            dataMedia.getNamespaceMode(),
                            targetDataMedia.getNamespaceMode());
                        String tableName = buildName(eventData.getSchemaName(),
                            dataMedia.getNameMode(),
                            targetDataMedia.getNameMode());
                        String shardColumns = targetDbDialect.getShardColumns(schemaName, tableName);
                        if (StringUtils.isNotEmpty(shardColumns)) {
                            String columns[] = StringUtils.split(shardColumns, ',');
                            for (String key : columns) {
                                org.apache.ddlutils.model.Column col = table.findColumn(key, false);
                                if (col != null) {
                                    col.setPrimaryKey(true);
                                } else {
                                    logger.warn(String.format("shardColumn %s in table[%s.%s] is not found",
                                        key,
                                        eventData.getSchemaName(),
                                        eventData.getTableName()));
                                }
                            }
                        }
                    }
                }
                tableHolder = new TableInfoHolder(dbDialect, table, useTableTransform);
            }
        }

        List<Column> beforeColumns = rowData.getBeforeColumnsList();
        List<Column> afterColumns = rowData.getAfterColumnsList();
        String tableName = eventData.getSchemaName() + "." + eventData.getTableName();

        // 判断一下是否需要all columns
        boolean isRowMode = pipeline.getParameters().getSyncMode().isRow(); // 如果是rowMode模式，所有字段都需要标记为updated
        boolean needAllColumns = isRowMode || checkNeedAllColumns(pipeline);

        // 变更后的主键
        Map<String, EventColumn> keyColumns = new LinkedHashMap<String, EventColumn>();
        // 变更前的主键
        Map<String, EventColumn> oldKeyColumns = new LinkedHashMap<String, EventColumn>();
        // 有变化的非主键
        Map<String, EventColumn> notKeyColumns = new LinkedHashMap<String, EventColumn>();

        if (eventType.isInsert()) {
            for (Column column : afterColumns) {
                if (isKey(tableHolder, tableName, column)) {
                    keyColumns.put(column.getName(), copyEventColumn(column, true, tableHolder));
                } else {
                    // mysql 有效
                    notKeyColumns.put(column.getName(), copyEventColumn(column, true, tableHolder));
                }
            }
        } else if (eventType.isDelete()) {
            for (Column column : beforeColumns) {
                if (isKey(tableHolder, tableName, column)) {
                    keyColumns.put(column.getName(), copyEventColumn(column, true, tableHolder));
                } else {
                    // mysql 有效
                    notKeyColumns.put(column.getName(), copyEventColumn(column, true, tableHolder));
                }
            }
        } else if (eventType.isUpdate()) {
            // 获取变更前的主键.
            for (Column column : beforeColumns) {
                if (isKey(tableHolder, tableName, column)) {
                    oldKeyColumns.put(column.getName(), copyEventColumn(column, true, tableHolder));
                    // 同时记录一下new
                    // key,因为mysql5.6之后出现了minimal模式,after里会没有主键信息,需要在before记录中找
                    keyColumns.put(column.getName(), copyEventColumn(column, true, tableHolder));
                } else {
                    if (needAllColumns && entry.getHeader().getSourceType() == CanalEntry.Type.ORACLE) {
                        // 针对行记录同步时，针对oracle记录一下非主键的字段，因为update时针对未变更的字段在aftercolume里没有
                        notKeyColumns.put(column.getName(), copyEventColumn(column, isRowMode, tableHolder));
                    }
                }
            }
            for (Column column : afterColumns) {
                if (isKey(tableHolder, tableName, column)) {
                    // 获取变更后的主键
                    keyColumns.put(column.getName(), copyEventColumn(column, true, tableHolder));
                } else if (needAllColumns || entry.getHeader().getSourceType() == CanalEntry.Type.ORACLE
                           || column.getUpdated()) {
                    // 在update操作时，oracle和mysql存放变更的非主键值的方式不同,oracle只有变更的字段;
                    // mysql会把变更前和变更后的字段都发出来，只需要取有变更的字段.
                    // 如果是oracle库，after里一定为对应的变更字段

                    boolean isUpdate = true;
                    if (entry.getHeader().getSourceType() == CanalEntry.Type.MYSQL) { // mysql的after里部分数据为未变更,oracle里after里为变更字段
                        isUpdate = column.getUpdated();
                    }

                    notKeyColumns.put(column.getName(), copyEventColumn(column, isRowMode || isUpdate, tableHolder));// 如果是rowMode，所有字段都为updated
                }
            }

            if (entry.getHeader().getSourceType() == CanalEntry.Type.ORACLE) { // 针对oracle进行特殊处理
                checkUpdateKeyColumns(oldKeyColumns, keyColumns);
            }
        }

        List<EventColumn> keys = new ArrayList<EventColumn>(keyColumns.values());
        List<EventColumn> oldKeys = new ArrayList<EventColumn>(oldKeyColumns.values());
        List<EventColumn> columns = new ArrayList<EventColumn>(notKeyColumns.values());

        Collections.sort(keys, new EventColumnIndexComparable());
        Collections.sort(oldKeys, new EventColumnIndexComparable());
        Collections.sort(columns, new EventColumnIndexComparable());
        if (!keyColumns.isEmpty()) {
            eventData.setKeys(keys);
            if (eventData.getEventType().isUpdate() && !oldKeys.equals(keys)) { // update类型，如果存在主键不同,则记录下old
                                                                                // keys为变更前的主键
                eventData.setOldKeys(oldKeys);
            }
            eventData.setColumns(columns);
            // } else if (CanalEntry.Type.MYSQL ==
            // entry.getHeader().getSourceType()) {
            // // 只支持mysql无主键同步
            // if (eventType.isUpdate()) {
            // List<EventColumn> oldColumns = new ArrayList<EventColumn>();
            // List<EventColumn> newColumns = new ArrayList<EventColumn>();
            // for (Column column : beforeColumns) {
            // oldColumns.add(copyEventColumn(column, true, tableHolder));
            // }
            //
            // for (Column column : afterColumns) {
            // newColumns.add(copyEventColumn(column, true, tableHolder));
            // }
            // Collections.sort(oldColumns, new EventColumnIndexComparable());
            // Collections.sort(newColumns, new EventColumnIndexComparable());
            // eventData.setOldKeys(oldColumns);// 做为老主键
            // eventData.setKeys(newColumns);// 做为新主键，需要保证新老主键字段数量一致
            // } else {
            // // 针对无主键，等同为所有都是主键进行处理
            // eventData.setKeys(columns);
            // }
        } else {
            throw new SelectException("this rowdata has no pks , entry: " + entry.toString() + " and rowData: "
                                      + rowData);
        }

        return eventData;
    }

@奔牛 目前不行。

@九霄道长 控制台->数据源管理 ->添加数据库 就行了。

或者说你是希望换cloudcanal元数据库？ 参考https://www.clougence.com/cc-doc/operation/change_mysql

@微希夷 不好意思，回复有点晚。

private CustomFieldV2 isDel_0 = CustomFieldV2.buildField(“is_del”, 0, Types.TINYINT, false, true, false);

把最后一个字段isUpdated 改成true.

@扯蛋的人生 我们确认下，之前应该有修复过类似问题，请问你的版本是什么？

@wx-ou3ig6q_w9jt8apglcags-ilql7g 增量任务有正常跑着么？

@wx-ou3ig6q_w9jt8apglcags-ilql7g 是否在全量迁移之后写入的这个数据？

@wx-ou3ig6q_w9jt8apglcags-ilql7g

https://www.clougence.com/cc-doc/faq/如何debug自定义代码

参考下这个？

@奔牛 目前不行。

@qq-ac8aa848cb9824046e3bcb01b80bece6 默认365天，所有增量任务共享。需要定期更新下时间。社区群可以找我们下。

@wx-ou3ig6q_w9jt8apglcags-ilql7g select * from xxx limit 1 有数据么？

@呆头鹅 额，这个不太正常。