flinkcdc heartbeat.timeout 如何配置？

flinkcdc heartbeat.timeout 如何配置我使用的是 flink-clients模式使用flink-connector-mysql-cdc 在Java编码上在哪里配置超时时间呀 flink默认是50秒？

展开

收起

真的很搞笑 2023-11-22 08:14:59 228 0

3 条回答

写回答

取消提交回答

小周sir

面对过去，不要迷离；面对未来，不必彷徨；活在今天，你只要把自己完全展示给别人看。
在Apache Flink的flink-connector-mysql-cdc中，可以设置heartbeat.timeout属性来控制连接MySQL的心跳超时时长，以确保连接的有效性。以下是详细步骤：
1. 在项目根目录下的conf/flink-conf.yaml文件中设置heartbeat超时时间：
```
execution.checkpointing.mode: exactly_once_at_checkpoint outlets: - name: result path: sftp://localhost:2222/my_output topic: output-topic compression.type: gzip checkpoint-interval: 60000 flush-on-checkpoint: true heartbeat.timeout: 600000 # 默认50s
```
这里，heartbeat.timeout代表心跳超时的时间长度，单位为毫秒，默认为50秒。
1. 在代码中设置heartbeat.timeout参数：
```
Properties properties = new Properties();
properties.setProperty("heartbeat.timeout", "600000");
```
2023-11-29 14:20:19

赞同展开评论打赏
芯在这

不推荐client 一个jar包运行呀，此回答整理自钉群“Flink CDC 社区”

2023-11-22 11:41:18

赞同展开评论打赏
sunrr
是的，Flink CDC提供了一些策略来处理同步过程中的错误。你可以在Flink SQL中设置semantic_only参数为true，这样当遇到某些特定类型的错误时（例如唯一性约束冲突），Flink CDC只会记录这些错误，但不会停止同步。

以下是一个示例：
```
CREATE TABLE my_table (
  id INT,
  name STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH ('connector' = 'kafka',
'topic' = 'my_topic',
'properties.bootstrap.servers' = 'localhost:9092',
'scan.startup.mode' = 'latest-offset',
'semantic_only' = 'true');
```
在这个示例中，我们创建了一个名为my_table的表，并使用Kafka connector从my_topic主题中读取数据。我们还设置了semantic_only参数为true，这样当遇到唯一性约束冲突等错误时，Flink CDC只会记录这些错误，但不会停止同步。
2023-11-22 08:55:26

赞同展开评论打赏