1.17.1检查点保存超时。拜托拜托

[banyue0618]

DeserializationSchema代码：
`public class CustomerDeserializationSchema implements DebeziumDeserializationSchema {

public CustomerDeserializationSchema() {
}

@Override
public void deserialize(SourceRecord sourceRecord, Collector<String> out) throws Exception {
    Envelope.Operation operation = Envelope.operationFor(sourceRecord);

    Struct value = (Struct) sourceRecord.value();

    Struct after = value.getStruct("after");

    JSONObject afterJson = struct2Json(after);

    // 如果是更新操作，则需要判断公证flag字段发生变化
    if(operation.name().equals("UPDATE")){
        Struct before = value.getStruct("before");
        JSONObject beforeJson = struct2Json(before);
        //        result.put("before", beforeJson);
        if(!beforeJson.getString("notarial").equals(afterJson.getString("notarial"))){
            // 四个指标全部设置为负数，是当前值 * -1
            beforeJson.put("data_count", -1);
            beforeJson.put("cargo_count", beforeJson.getIntValue("cargo_count") * -1);
            beforeJson.put("cargo_value", beforeJson.getBigDecimal("cargo_value").multiply(new java.math.BigDecimal(-1)));
            beforeJson.put("item_count", beforeJson.getIntValue("item_count") * -1);
            out.collect(beforeJson.toJSONString());
        }
    }
    afterJson.put("data_count", 1);
    out.collect(afterJson.toJSONString());
}

private JSONObject struct2Json(Struct struct) {
    JSONObject jsonObject = new JSONObject();
    if (struct != null){
        Schema schema = struct.schema();
        List<Field> fieldsList = schema.fields();

        for (Field field : fieldsList) {
            jsonObject.put(field.name(), struct.get(field));
        }
    }
    return jsonObject;
}

@Override
public TypeInformation<String> getProducedType() {
    return BasicTypeInfo.STRING_TYPE_INFO;
}

}`

sink端
`public class DorisSink extends RichSinkFunction {

private static final Logger logger = org.slf4j.LoggerFactory.getLogger(DorisSink.class);

private static final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");

ParameterTool parameterTool;

PreparedStatement ps;
private Connection connection;

public DorisSink(ParameterTool parameterTool) {
    this.parameterTool = parameterTool;
}

@Override
public void open(Configuration parameters) throws Exception {
    super.open(parameters);
    connection = getConnection();
    String sql = "INSERT INTO bi_data_subject (cbm_id, create_time, notarial, data_count, item_count, cargo_count, cargo_value, production_country, production_cbm) VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?);";
    if (connection != null) {
        ps = this.connection.prepareStatement(sql);
    }
}

@Override
public void close() throws Exception {
    super.close();
    //关闭连接和释放资源
    if (connection != null) {
        connection.close();
    }
    if (ps != null) {
        ps.close();
    }
}

@Override
public void invoke(String value, Context context) throws Exception {
    if (ps == null) {
        return;
    }
    JSONObject object = JSONObject.parseObject(value);
    //组装数据，执行插入操作
    ps.setInt(1, object.getIntValue("req_cbm_id"));
    LocalDateTime dateTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(object.getLongValue("create_time")), ZoneId.systemDefault());
    ps.setObject(2, dateTime.format(formatter));
    ps.setString(3, object.getString("notarial"));
    ps.setInt(4, object.getIntValue("data_count"));
    ps.setInt(5, object.getIntValue("item_count"));
    ps.setInt(6, object.getIntValue("cargo_count"));
    ps.setBigDecimal(7, object.getBigDecimal("cargo_value"));
    ps.setString(8, object.getString("production_country"));
    ps.setString(9, object.getString("production_cbm"));
    ps.executeUpdate();
}

private Connection getConnection() {
    Connection con = null;
    try {
        Class.forName("com.mysql.cj.jdbc.Driver");
        con = DriverManager.getConnection(
                parameterTool.get(PropertiesConstants.DORIS_URI),
                parameterTool.get(PropertiesConstants.DORIS_USERNAME),
                parameterTool.get(PropertiesConstants.DORIS_PASSWORD));
    } catch (Exception e) {
        logger.error("获取数据库连接失败", e);
    }
    return con;
}

}`

错误信息
2024-08-29 15:35:54,798 DEBUG org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Received heartbeat request from a308816b3413a14efb329fdda338c135. 2024-08-29 15:35:54,925 DEBUG org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Received heartbeat request from f80fa117ee2f48201f2098a66a43a313. 2024-08-29 15:36:04,798 DEBUG org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Received heartbeat request from a308816b3413a14efb329fdda338c135. 2024-08-29 15:36:04,926 DEBUG org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Received heartbeat request from f80fa117ee2f48201f2098a66a43a313. 2024-08-29 15:36:14,798 DEBUG org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Received heartbeat request from a308816b3413a14efb329fdda338c135. 2024-08-29 15:36:14,926 DEBUG org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Received heartbeat request from f80fa117ee2f48201f2098a66a43a313. 2024-08-29 15:36:22,517 DEBUG org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Abort checkpoint 2@1724916382513 for 411b05fdee584c9ae572b2d3a2ac10b8_cbc357ccb763df2852fee8c4fc7d55f2_0_0. 2024-08-29 15:36:22,522 INFO org.apache.flink.runtime.taskmanager.Task [] - Attempting to cancel task Source: Custom Source -> Sink: Unnamed (1/1)#0 (411b05fdee584c9ae572b2d3a2ac10b8_cbc357ccb763df2852fee8c4fc7d55f2_0_0). 2024-08-29 15:36:22,522 INFO org.apache.flink.runtime.taskmanager.Task [] - Source: Custom Source -> Sink: Unnamed (1/1)#0 (411b05fdee584c9ae572b2d3a2ac10b8_cbc357ccb763df2852fee8c4fc7d55f2_0_0) switched from RUNNING to CANCELING. 2024-08-29 15:36:22,522 INFO org.apache.flink.runtime.taskmanager.Task [] - Triggering cancellation of task code Source: Custom Source -> Sink: Unnamed (1/1)#0 (411b05fdee584c9ae572b2d3a2ac10b8_cbc357ccb763df2852fee8c4fc7d55f2_0_0). 2024-08-29 15:36:22,524 INFO io.debezium.embedded.EmbeddedEngine [] - Stopping the embedded engine 2024-08-29 15:36:22,524 INFO io.debezium.embedded.EmbeddedEngine [] - Waiting for PT5M for connector to stop 2024-08-29 15:36:24,798 DEBUG org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Received heartbeat request from a308816b3413a14efb329fdda338c135. 2024-08-29 15:36:24,927 DEBUG org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Received heartbeat request from f80fa117ee2f48201f2098a66a43a313. 2024-08-29 15:36:34,799 DEBUG org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Received heartbeat request from a308816b3413a14efb329fdda338c135. 2024-08-29 15:36:34,927 DEBUG org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Received heartbeat request from f80fa117ee2f48201f2098a66a43a313. 2024-08-29 15:36:44,799 DEBUG org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Received heartbeat request from a308816b3413a14efb329fdda338c135. 2024-08-29 15:36:44,927 DEBUG org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Received heartbeat request from f80fa117ee2f48201f2098a66a43a313. 2024-08-29 15:36:52,525 WARN org.apache.flink.runtime.taskmanager.Task [] - Task 'Source: Custom Source -> Sink: Unnamed (1/1)#0' did not react to cancelling signal - interrupting; it is stuck for 30 seconds in method: org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:92) org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:90) org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMail(MailboxProcessor.java:398) org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsWhenDefaultActionUnavailable(MailboxProcessor.java:367) org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:352) org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:229) org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:839) org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:788) org.apache.flink.runtime.taskmanager.Task$$Lambda$853/672909390.run(Unknown Source) org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:952) org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:931) org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:745) org.apache.flink.runtime.taskmanager.Task.run(Task.java:562) java.lang.Thread.run(Thread.java:748)

不知道为什么保存检查点会超时失败。

任务自动重启。
表里面数据是同步了一部分。但是web上的数量一直显示0 ，这又是为啥 ？？？