public void start()

in eventmesh-connectors/eventmesh-connector-canal/src/main/java/org/apache/eventmesh/connector/canal/sink/connector/CanalCheckConsumer.java [77:165]
84 lines of code
20 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>> sourceRows = JsonUtils.parseObject(dataMap.get("data").toString(), List.class);

            if (sourceRows == null || sourceRows.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;
            }

            String sql = genTargetPkInSql(tableDefinition, sourceRows.size(), Constants.MySQLQuot, Constants.MySQLQuot, "*");
            DruidPooledConnection connection = null;
            PreparedStatement statement = null;
            try {
                connection = DatabaseConnection.sinkDataSource.getConnection();
                statement =
                        connection.prepareStatement(sql);
                setPrepareParams(statement, sourceRows, tableDefinition);
                log.debug("select sql {}", statement.toString());
                ResultSet resultSet = statement.executeQuery();
                List<Map<String, Object>> targetRows = new LinkedList<>();
                while (resultSet.next()) {
                    Map<String, Object> columnValues = new LinkedHashMap<>();
                    for (Map.Entry<String, MySQLColumnDef> col :
                            tableDefinition.getColumnDefinitions().entrySet()) {
                        columnValues.put(col.getKey(), readColumn(resultSet, col.getKey(),
                                col.getValue().getType()));
                    }
                    targetRows.add(columnValues);
                }
                compareData(sourceRows, targetRows, tableDefinition);
                record.getCallback().onSuccess(convertToSendResult(record));
            } catch (SQLException e) {
                log.warn("check sink process schema [{}] table [{}] connector check fail", tableDefinition.getSchemaName(),
                        tableDefinition.getTableName(),
                        e);
                LockSupport.parkNanos(3000 * 1000L);
                record.getCallback().onException(buildSendExceptionContext(record, e));
            } catch (Exception e) {
                log.error("check sink process schema [{}] table [{}] catch unknown exception", tableDefinition.getSchemaName(),
                        tableDefinition.getTableName(), e);
                record.getCallback().onException(buildSendExceptionContext(record, 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);
                    }
                }
            }
        }
    }