增加错误消息日志打印

duxinglangzi

2022-09-06 ddae546da9487622631d47133c962e7d870982de

 src/main/java/com/duxinglangzi/canal/starter/container/DmlMessageTransponderContainer.java

@@ -5,6 +5,7 @@
import com.alibaba.otter.canal.protocol.Message;

import com.duxinglangzi.canal.starter.configuration.CanalAutoConfigurationProperties;

import com.duxinglangzi.canal.starter.configuration.CanalListenerEndpointRegistrar;

import com.duxinglangzi.canal.starter.mode.CanalMessage;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;



@@ -50,42 +51,49 @@




    public void doStart() {

        Message message = null;

        try {

            message = connector.getWithoutAck(endpointInstance.getBatchSize()); // 获取指定数量的数据

        } catch (Exception clientException) {

            logger.error("[DmlMessageTransponderContainer] error msg : ", clientException);

            if (local_retry_count > 0) {

                // 重试次数减一

                local_retry_count = local_retry_count - 1;

                sleep(endpointInstance.getAcquireInterval());

            } else {

                // 重试次数 <= 0 时,直接终止线程

                logger.error("[DmlMessageTransponderContainer] retry count is zero ,  " +

                                "thread interrupt , current connector host: {} , port: {} ",

                        endpointInstance.getHost(), endpointInstance.getPort());

                Thread.currentThread().interrupt();

            }

            return;

        }

        // 如果重试次数小于设置的,则修改

        if (local_retry_count < endpointInstance.getRetryCount()) local_retry_count = endpointInstance.getRetryCount();

        List<CanalEntry.Entry> entries = message.getEntries();

        if (message.getId() == -1 || entries.isEmpty()) {

            sleep(endpointInstance.getAcquireInterval());

            return;

        }

        for (CanalEntry.Entry entry : entries) {

            Message message = null;

            try {

                consumer(entry);

            } catch (Exception e) {

                e.printStackTrace();

                logger.error("[DmlMessageTransponderContainer_doStart] CanalEntry.Entry consumer error ", e);

                // connector.rollback(message.getId()); // 目前先不处理失败, 无需回滚数据

                // return;

                message = connector.getWithoutAck(endpointInstance.getBatchSize()); // 获取指定数量的数据

            } catch (Exception clientException) {

                logger.error("[DmlMessageTransponderContainer] error msg : ", clientException);

                if (local_retry_count > 0) {

                    // 重试次数减一

                    local_retry_count = local_retry_count - 1;

                    sleep(endpointInstance.getAcquireInterval());

                } else {

                    // 重试次数 <= 0 时,直接终止线程

                    logger.error("[DmlMessageTransponderContainer] retry count is zero ,  " +

                                    "thread interrupt , current connector host: {} , port: {} ",

                            endpointInstance.getHost(), endpointInstance.getPort());

                    Thread.currentThread().interrupt();

                }

                return;

            }

            // 如果重试次数小于设置的,则修改

            if (local_retry_count < endpointInstance.getRetryCount())

                local_retry_count = endpointInstance.getRetryCount();

            List<CanalEntry.Entry> entries = message.getEntries();

            if (message.getId() == -1 || entries.isEmpty()) {

                sleep(endpointInstance.getAcquireInterval());

                return;

            }

            for (CanalEntry.Entry entry : entries) {

                try {

                    consumer(entry);

                } catch (Exception e) {

                    e.printStackTrace();

                    logger.error("[DmlMessageTransponderContainer_doStart] CanalEntry.Entry consumer error ", e);

                    // connector.rollback(message.getId()); // 目前先不处理失败, 无需回滚数据

                    // return;

                }

            }

            connector.ack(message.getId()); // 提交确认

        } catch (Exception exc) {

            logger.error("[DmlMessageTransponderContainer_doStart] Pull data message exception,errorMessage:{}", exc.getLocalizedMessage());

            // 防止删除消息时发生错误,或者拉取消息失败等情况

            exc.printStackTrace();

        }

        connector.ack(message.getId()); // 提交确认

    }



    public DmlMessageTransponderContainer(

@@ -119,7 +127,7 @@
                            eventType))

                    .forEach(element -> {

                        try {

                            element.getMethod().invoke(element.getBean(), eventType, rowData);

                            element.getMethod().invoke(element.getBean(), new CanalMessage(entry.getHeader(), eventType, rowData));

                        } catch (IllegalAccessException | InvocationTargetException e) {

                            logger.error("[DmlMessageTransponderContainer_consumer] RowData Callback Method invoke error message", e);

                            throw new RuntimeException("RowData Callback Method invoke error message： " + e.getMessage(), e);