public void start()

in eventmesh-connectors/eventmesh-connector-canal/src/main/java/org/apache/eventmesh/connector/canal/sink/connector/CanalFullConsumer.java [70:161]
88 lines of code
22 McCabe index (conditional complexity)

    public void start(AtomicBoolean flag) {
        while (flag.get()) {
            List<ConnectRecord> sinkRecords = null;
            try {
                sinkRecords = queue.poll(2, TimeUnit.SECONDS);
                if (sinkRecords == null || sinkRecords.isEmpty()) {
                    continue;
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            ConnectRecord record = sinkRecords.get(0);
            Map<String, Object> dataMap =
                JsonUtils.parseTypeReferenceObject((byte[]) record.getData(), new TypeReference<Map<String, Object>>() {
                });

            List<Map<String, Object>> rows = JsonUtils.parseObject(dataMap.get("data").toString(), List.class);

            if (rows == null || rows.isEmpty()) {
                if (log.isDebugEnabled()) {
                    log.debug("[{}] got rows data is none", this.getClass());
                }
                return;
            }
            CanalFullRecordOffset offset = JsonUtils.parseObject(dataMap.get("offset").toString(), CanalFullRecordOffset.class);
            if (offset == null || offset.getPosition() == null) {
                if (log.isDebugEnabled()) {
                    log.debug("[{}] got canal full offset is none", this.getClass());
                }
                return;
            }

            MySQLTableDef tableDefinition = (MySQLTableDef) tableMgr.getTable(offset.getPosition().getSchema(), offset.getPosition().getTableName());
            if (tableDefinition == null) {
                log.warn("target schema [{}] table [{}] is not exists", offset.getPosition().getSchema(), offset.getPosition().getTableName());
                return;
            }
            List<MySQLColumnDef> cols = new ArrayList<>(tableDefinition.getColumnDefinitions().values());
            String sql = generateInsertPrepareSql(offset.getPosition().getSchema(), offset.getPosition().getTableName(),
                cols);
            DruidPooledConnection connection = null;
            PreparedStatement statement = null;
            try {
                connection = DatabaseConnection.sinkDataSource.getConnection();
                statement =
                    connection.prepareStatement(sql);
                for (Map<String, Object> col : rows) {
                    setPrepareParams(statement, col, cols);
                    log.debug("insert sql {}", statement.toString());
                    statement.addBatch();
                }
                statement.executeBatch();
                connection.commit();
                log.info("execute batch insert sql size: {}", rows.size());
                record.getCallback().onSuccess(convertToSendResult(record));
            } catch (SQLException e) {
                log.warn("full sink process schema [{}] table [{}] connector write fail", tableDefinition.getSchemaName(),
                    tableDefinition.getTableName(),
                    e);
                LockSupport.parkNanos(3000 * 1000L);
                record.getCallback().onException(buildSendExceptionContext(record, e));
            } catch (Exception e) {
                log.error("full sink process schema [{}] table [{}] catch unknown exception", tableDefinition.getSchemaName(),
                    tableDefinition.getTableName(), e);
                record.getCallback().onException(buildSendExceptionContext(record, e));
                try {
                    if (connection != null && !connection.isClosed()) {
                        connection.rollback();
                    }
                } catch (SQLException rollback) {
                    log.warn("full sink process schema [{}] table [{}] rollback fail", tableDefinition.getSchemaName(),
                        tableDefinition.getTableName(), e);
                }
            } finally {
                if (statement != null) {
                    try {
                        statement.close();
                    } catch (SQLException e) {
                        log.error("close prepare statement fail", e);
                    }
                }

                if (connection != null) {
                    try {
                        connection.close();
                    } catch (SQLException e) {
                        log.error("close db connection fail", e);
                    }
                }
            }
        }
    }