You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by yu...@apache.org on 2026/02/26 14:00:10 UTC
(fluss) branch release-0.9 updated: [hotfix][flink] Fix release kv snapshot lease for log table (#2719) (#2721)
This is an automated email from the ASF dual-hosted git repository.
yunhong pushed a commit to branch release-0.9
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/release-0.9 by this push:
new abf4c9149 [hotfix][flink] Fix release kv snapshot lease for log table (#2719) (#2721)
abf4c9149 is described below
commit abf4c91495d4eaae48ec3512005e350a93fc1e03
Author: Liebing <44...@users.noreply.github.com>
AuthorDate: Thu Feb 26 22:00:06 2026 +0800
[hotfix][flink] Fix release kv snapshot lease for log table (#2719) (#2721)
(cherry picked from commit 4e54416986b31bd75900e5ba4cedd02e63003579)
---
.../source/enumerator/FlinkSourceEnumerator.java | 63 ++++++++++++----------
1 file changed, 34 insertions(+), 29 deletions(-)
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java
index e4e800c67..b537a8122 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java
@@ -1054,37 +1054,42 @@ public class FlinkSourceEnumerator
public void notifyCheckpointComplete(long checkpointId) throws Exception {
checkpointTriggeredBefore = true;
- // lower than this checkpoint id.
- Set<TableBucket> consumedKvSnapshots = getAndRemoveConsumedBucketsUpTo(checkpointId);
+ if (hasPrimaryKey) {
+ // lower than this checkpoint id.
+ Set<TableBucket> consumedKvSnapshots = getAndRemoveConsumedBucketsUpTo(checkpointId);
- LOG.info(
- "kv snapshot has already consumed and try to release kv snapshot lease for: {}, checkpoint id: {}",
- consumedKvSnapshots,
- checkpointId);
+ if (!consumedKvSnapshots.isEmpty()) {
+ LOG.info(
+ "kv snapshot has already consumed and try to release kv snapshot lease for: {}, checkpoint id: {}",
+ consumedKvSnapshots,
+ checkpointId);
- // send request to fluss to unregister the kv snapshot lease.
- try {
- flussAdmin
- .createKvSnapshotLease(
- leaseContext.getKvSnapshotLeaseId(),
- leaseContext.getKvSnapshotLeaseDurationMs())
- .releaseSnapshots(consumedKvSnapshots)
- .get();
- } catch (Exception e) {
- if (ExceptionUtils.findThrowable(e, UnsupportedVersionException.class).isPresent()) {
- LOG.warn(
- "Failed to release kv snapshot lease because the server does not support "
- + "kv snapshot lease API. Snapshots may remain in storage longer "
- + "than necessary. Please upgrade the Fluss server to version 0.9 "
- + "or later.",
- e);
- } else {
- LOG.error(
- "Failed to release kv snapshot lease. These snapshots need to re-enqueue",
- e);
- // use the current checkpoint id to re-enqueue the buckets
- consumedKvSnapshots.forEach(
- tableBucket -> addConsumedBucket(checkpointId, tableBucket));
+ // send request to fluss to unregister the kv snapshot lease.
+ try {
+ flussAdmin
+ .createKvSnapshotLease(
+ leaseContext.getKvSnapshotLeaseId(),
+ leaseContext.getKvSnapshotLeaseDurationMs())
+ .releaseSnapshots(consumedKvSnapshots)
+ .get();
+ } catch (Exception e) {
+ if (ExceptionUtils.findThrowable(e, UnsupportedVersionException.class)
+ .isPresent()) {
+ LOG.warn(
+ "Failed to release kv snapshot lease because the server does not support "
+ + "kv snapshot lease API. Snapshots may remain in storage longer "
+ + "than necessary. Please upgrade the Fluss server to version 0.9 "
+ + "or later.",
+ e);
+ } else {
+ LOG.error(
+ "Failed to release kv snapshot lease. These snapshots need to re-enqueue",
+ e);
+ // use the current checkpoint id to re-enqueue the buckets
+ consumedKvSnapshots.forEach(
+ tableBucket -> addConsumedBucket(checkpointId, tableBucket));
+ }
+ }
}
}
}