Flink CDC1.4及以上，这个问题有那位知道怎么解决吗？

Flink CDC1.4及以上，flink cdc 同步oracle数据库的时候做snapshot，会报Call snapshotState() on closed source, checkpoint failed。这个问题有那位知道怎么解决吗？

展开

收起

cuicuicuic 2023-11-16 07:54:04 70 0

2 条回答

写回答

取消提交回答

小周sir

面对过去，不要迷离；面对未来，不必彷徨；活在今天，你只要把自己完全展示给别人看。
这个问题可能是由于 Flink CDC 在同步 Oracle 数据库时在完成快照后关闭了源导致的。您可以在 Flink CDC 的官方文档中找到相关的解决方案：
```
```yaml
mySqlSource:
  type: mysql-source
  database: testDB
  table.whitelist: myTable
  parallelism: 1
  snapshot.incremental.snapshot.isolation.level: read-committed
  snapshot.new.snapshot.select.sql: SELECT * FROM myTable
  snapshot.mode: when_needed # 开启实时模式

# 解决 Call snapshotState() on closed source, checkpoint failed 错误
mySqlSource:
  snapshot.fetch-size: 5000
```
其中 snapshot.incremental.snapshot.isolation.level: read-committed 和 snapshot.new.snapshot.select.sql 设置是为了开启实时模式，避免在快照完成后关闭源；snapshot.fetch-size 设置则可以限制每次快照获取的数据量。
同时，您还可以尝试降低 checkpoint 间隔，以及检查网络环境、硬件资源等其他可能导致性能瓶颈的因素。
2023-11-16 13:29:05

赞同展开评论打赏
sunrr
这个问题可能是由于在关闭源表后尝试调用snapshotState()方法导致的。为了解决这个问题，你可以尝试以下步骤：
1. 确保在关闭源表之前已经成功创建了快照。你可以使用FlinkCDCSource的getChangelogStream()方法来获取一个ChangelogStream对象，然后使用createSnapshot()方法来创建快照。
```
FlinkCDCSource<Row> source = ...;
ChangelogStream changelogStream = source.getChangelogStream();
changelogStream.createSnapshot();
```
1. 在关闭源表之前，确保已经提交了所有挂起的操作。你可以使用FlinkCDCSource的getChangelogStream()方法来获取一个ChangelogStream对象，然后使用commit()方法来提交所有挂起的操作。
```
FlinkCDCSource<Row> source = ...;
ChangelogStream changelogStream = source.getChangelogStream();
changelogStream.commit();
```
1. 如果问题仍然存在，你可以尝试在关闭源表之前等待一段时间，以确保所有挂起的操作都已经提交。你可以使用Thread.sleep()方法来实现这个功能。
```
FlinkCDCSource<Row> source = ...;
ChangelogStream changelogStream = source.getChangelogStream();
changelogStream.commit();
Thread.sleep(5000); // 等待5秒
```
2023-11-16 10:22:42

赞同展开评论打赏