You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by lz...@apache.org on 2022/12/30 02:08:19 UTC
[flink-table-store] branch master updated: [hotfix] Remove incorrect check in FullChangelogStoreSinkWrite
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-table-store.git
The following commit(s) were added to refs/heads/master by this push:
new dea8ac80 [hotfix] Remove incorrect check in FullChangelogStoreSinkWrite
dea8ac80 is described below
commit dea8ac80a043728e6ae50984b315b78f8aea6628
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Fri Dec 30 10:08:14 2022 +0800
[hotfix] Remove incorrect check in FullChangelogStoreSinkWrite
This closes #448
---
.../store/connector/sink/FullChangelogStoreSinkWrite.java | 15 +--------------
1 file changed, 1 insertion(+), 14 deletions(-)
diff --git a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/FullChangelogStoreSinkWrite.java b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/FullChangelogStoreSinkWrite.java
index e7e9a148..3d7b0395 100644
--- a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/FullChangelogStoreSinkWrite.java
+++ b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/FullChangelogStoreSinkWrite.java
@@ -196,6 +196,7 @@ public class FullChangelogStoreSinkWrite extends StoreSinkWriteImpl {
}
private Optional<Snapshot> findSnapshot(long identifierToCheck) {
+ // TODO We need a mechanism to do timeout recovery in case of snapshot expiration.
SnapshotManager snapshotManager = table.snapshotManager();
Long latestId = snapshotManager.latestSnapshotId();
if (latestId == null) {
@@ -207,20 +208,6 @@ public class FullChangelogStoreSinkWrite extends StoreSinkWriteImpl {
return Optional.empty();
}
- if (earliestId > identifierToCheck) {
- throw new RuntimeException(
- String.format(
- "Can't find snapshot with identifier %d from user %s. "
- + "Earliest snapshot from all users has identifier %d. "
- + "This is rare but it might be that snapshots are expiring too fast.\n"
- + "If this exception happens continuously, please consider increasing %s or %s.",
- identifierToCheck,
- commitUser,
- earliestId,
- CoreOptions.SNAPSHOT_TIME_RETAINED.key(),
- CoreOptions.SNAPSHOT_NUM_RETAINED_MIN.key()));
- }
-
// We must find the snapshot whose identifier is exactly `identifierToCheck`.
// We can't just compare with the latest snapshot identifier by this user (like what we do
// in `AbstractFileStoreWrite`), because even if the latest snapshot identifier is newer,