You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "XuQianJin-Stars (via GitHub)" <gi...@apache.org> on 2026/05/09 03:28:30 UTC
[PR] [flink] Support kv snapshot lease in tiering service (#2898) [fluss]
XuQianJin-Stars opened a new pull request, #3286:
URL: https://github.com/apache/fluss/pull/3286
TieringSourceEnumerator now acquires a KV snapshot lease for all TieringSnapshotSplits before they are assigned to readers, and releases the lease when the table finishes or fails tiering, or when a reader failover returns the splits. A best-effort `dropLease` is also performed on enumerator close. This prevents the Fluss server from cleaning up snapshots that the tiering job still depends on.
One lease id per tiering job (UUID-based) is reused across tables and persisted into `TieringSourceEnumeratorState` so that it survives enumerator restore instead of leaking orphan leases. The lease uses a fixed 1-day duration that is implicitly renewed by every `acquireSnapshots` call, and `UnsupportedVersionException` from older Fluss servers is downgraded to a warning to keep backward compatibility.
### Purpose
Linked issue: close #2898
Before this change, the tiering job only read Fluss KV snapshots without holding any lease on them. A long-running tiering job could therefore race with the server-side snapshot GC: the server may clean up a snapshot that is still being / about to be consumed by the tiering `SourceReader`, causing tiering failures or data loss on the lake side.
This PR makes `TieringSource` hold a KV snapshot lease for the full lifecycle of each snapshot split it hands out, so that the Fluss server will not reclaim those snapshots while tiering is in progress.
### Brief change log
- `TieringSourceEnumerator`
- Generate one `kvSnapshotLeaseId` per tiering job (UUID-based) and reuse it across all tables.
- Before assigning any `TieringSnapshotSplit` to a reader, call `acquireSnapshots(leaseId, snapshots, 1 day)` on the admin / gateway client to acquire a lease covering all snapshot splits of the table.
- Track in-flight leased snapshots per table; release the lease (`releaseSnapshots`) when the table finishes tiering, fails tiering, or when a reader failover returns the splits back to the enumerator.
- On enumerator `close()`, best-effort `dropLease(leaseId)` to release everything still held by this job.
- Downgrade `UnsupportedVersionException` (old server) to a warning log, so the tiering job keeps working against older Fluss servers without the lease API.
- `TieringSourceEnumeratorState` + `TieringSourceEnumeratorStateSerializer`
- Persist `kvSnapshotLeaseId` into the enumerator checkpoint state (new serializer version, backward compatible with the previous version).
- `TieringSource`
- On `restoreEnumerator`, reuse the persisted `kvSnapshotLeaseId` from the checkpoint so the recovered enumerator does not generate a new UUID and leak the previous lease.
### Tests
- `TieringSourceEnumeratorTest`
- New cases covering: lease is acquired before snapshot splits are assigned; lease is released on table finish / fail / reader failover; `dropLease` is invoked on enumerator close; enumerator works gracefully when the server returns `UnsupportedVersionException`.
- `TieringSourceEnumeratorStateSerializerTest`
- Round-trip tests for the new `kvSnapshotLeaseId` field, plus a backward-compatibility case that deserializes a state written by the previous serializer version.
- `mvn clean verify` passes locally for the affected modules.
### API and Format
- No public user-facing API change.
- `TieringSourceEnumeratorState` checkpoint format is extended with a new `kvSnapshotLeaseId` field. The serializer version is bumped and older checkpoints remain readable (the field defaults to a freshly generated UUID on restore from old state).
- No storage / wire format change on the Fluss server side; this PR only consumes the existing `acquireSnapshots` / `releaseSnapshots` / `dropLease` admin APIs.
### Documentation
- No new user-facing feature or configuration option is introduced, so no documentation update is required.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] Support kv snapshot lease in tiering service (#2898) [fluss]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3286:
URL: https://github.com/apache/fluss/pull/3286#discussion_r3212376980
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -515,6 +586,18 @@ public void close() throws IOException {
LOG.error("Failed to close Tiering Source enumerator.", e);
}
}
+ // NOTE: we intentionally do NOT drop the kv snapshot lease here. The lease id is
+ // persisted into the enumerator checkpoint state and will be reused by the restored
+ // enumerator after a JM failover. Dropping it on close would destroy the lease that
+ // the restored enumerator expects to reuse, potentially causing the referenced
+ // snapshots to be garbage-collected before tiering finishes. The lease will expire
Review Comment:
The PR description says the enumerator performs a best-effort `dropLease` on close, but the implementation explicitly avoids dropping the lease here. Please either align the implementation with the description (e.g., drop when it’s safe/known not to be a restart) or update the PR description so operators understand the actual lifecycle/cleanup behavior.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -533,6 +616,145 @@ public void close() throws IOException {
}
}
+ /**
+ * Acquire kv snapshot lease for all {@link TieringSnapshotSplit}s of the given table so that
+ * snapshots referenced by these splits will not be cleaned up by the Fluss server during
+ * tiering. Bucket-snapshot mappings are remembered in {@link #leasedBucketsByTable} so they can
+ * be released on finish/fail.
+ *
+ * <p>Falls back to a warning (no exception thrown) when the server does not support the kv
+ * snapshot lease API, to preserve compatibility with older Fluss clusters.
+ */
+ private void maybeAcquireKvSnapshotLease(long tableId, List<TieringSplit> tieringSplits) {
+ if (flussAdmin == null) {
+ return;
+ }
+ Map<TableBucket, Long> bucketsToLease = new HashMap<>();
+ for (TieringSplit split : tieringSplits) {
+ if (split.isTieringSnapshotSplit()) {
+ TieringSnapshotSplit snapshotSplit = split.asTieringSnapshotSplit();
+ bucketsToLease.put(snapshotSplit.getTableBucket(), snapshotSplit.getSnapshotId());
+ }
+ }
+ if (bucketsToLease.isEmpty()) {
+ return;
+ }
+ LOG.info(
+ "Try to acquire kv snapshot lease {} for tiering table {} with {} buckets.",
+ kvSnapshotLeaseId,
+ tableId,
+ bucketsToLease.size());
+ try {
+ flussAdmin
+ .createKvSnapshotLease(kvSnapshotLeaseId, KV_SNAPSHOT_LEASE_DURATION_MS)
+ .acquireSnapshots(bucketsToLease)
+ .get();
+ leasedBucketsByTable
+ .computeIfAbsent(tableId, k -> ConcurrentHashMap.newKeySet())
+ .addAll(bucketsToLease.keySet());
+ } catch (Exception e) {
+ if (ExceptionUtils.findThrowable(e, UnsupportedVersionException.class).isPresent()) {
+ LOG.warn(
+ "Failed to acquire kv snapshot lease for tiering table {} because the "
+ + "server does not support kv snapshot lease API. Snapshots may "
+ + "be cleaned up earlier than expected. Please upgrade the Fluss "
+ + "server to version 0.9 or later.",
+ tableId,
+ e);
+ } else {
+ LOG.error(
+ "Failed to acquire kv snapshot lease for tiering table {}. "
+ + "Tiering will proceed without snapshot protection; the "
+ + "snapshot may be garbage-collected while tiering is in progress.",
+ tableId,
+ e);
+ }
+ }
+ }
+
+ /**
+ * Release the kv snapshot lease held for a specific table. Called when a table finishes
+ * tiering, fails, or is abandoned due to failover. Missing leases (log-only tables, or tables
+ * for which acquire failed) are handled as no-ops.
+ */
+ private void maybeReleaseKvSnapshotLease(long tableId) {
+ Set<TableBucket> buckets = leasedBucketsByTable.remove(tableId);
+ if (flussAdmin == null || buckets == null || buckets.isEmpty()) {
+ return;
+ }
Review Comment:
`leasedBucketsByTable.remove(tableId)` happens before the RPC release succeeds. If `releaseSnapshots(..)` fails transiently, the enumerator loses the ability to retry and the server-side snapshots may remain leased until expiry. Consider only removing from `leasedBucketsByTable` after a successful release, or re-inserting the buckets on failure (similar to FlinkSourceEnumerator re-enqueueing buckets on release failure).
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/state/TieringSourceEnumeratorStateSerializer.java:
##########
@@ -38,14 +54,41 @@ public int getVersion() {
@Override
public byte[] serialize(TieringSourceEnumeratorState obj) throws IOException {
- // no need to store anything
- return new byte[0];
+ final DataOutputSerializer out = SERIALIZER_CACHE.get();
+ String leaseId = obj.getKvSnapshotLeaseId();
+ if (leaseId != null) {
+ out.writeBoolean(true);
+ out.writeUTF(leaseId);
+ } else {
+ out.writeBoolean(false);
+ }
+ final byte[] result = out.getCopyOfBuffer();
+ out.clear();
+ return result;
Review Comment:
`DataOutputSerializer` is cleared only on the happy path. If `writeUTF` (or any future added write) throws, the ThreadLocal serializer keeps a dirty buffer and subsequent serializations can become corrupted. Wrap the writes in a try/finally and clear the serializer in the finally block.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -533,6 +616,145 @@ public void close() throws IOException {
}
}
+ /**
+ * Acquire kv snapshot lease for all {@link TieringSnapshotSplit}s of the given table so that
+ * snapshots referenced by these splits will not be cleaned up by the Fluss server during
+ * tiering. Bucket-snapshot mappings are remembered in {@link #leasedBucketsByTable} so they can
+ * be released on finish/fail.
+ *
+ * <p>Falls back to a warning (no exception thrown) when the server does not support the kv
+ * snapshot lease API, to preserve compatibility with older Fluss clusters.
+ */
+ private void maybeAcquireKvSnapshotLease(long tableId, List<TieringSplit> tieringSplits) {
+ if (flussAdmin == null) {
+ return;
+ }
+ Map<TableBucket, Long> bucketsToLease = new HashMap<>();
+ for (TieringSplit split : tieringSplits) {
+ if (split.isTieringSnapshotSplit()) {
+ TieringSnapshotSplit snapshotSplit = split.asTieringSnapshotSplit();
+ bucketsToLease.put(snapshotSplit.getTableBucket(), snapshotSplit.getSnapshotId());
+ }
+ }
+ if (bucketsToLease.isEmpty()) {
+ return;
+ }
+ LOG.info(
+ "Try to acquire kv snapshot lease {} for tiering table {} with {} buckets.",
+ kvSnapshotLeaseId,
+ tableId,
+ bucketsToLease.size());
+ try {
+ flussAdmin
+ .createKvSnapshotLease(kvSnapshotLeaseId, KV_SNAPSHOT_LEASE_DURATION_MS)
+ .acquireSnapshots(bucketsToLease)
+ .get();
+ leasedBucketsByTable
Review Comment:
`acquireSnapshots(...)` returns an `AcquireKvSnapshotLeaseResult` that can report unavailable snapshots (already GC’ed / missing). The current code ignores the result and then records *all* requested buckets as leased, which can hide partial acquisition and later attempt to tier snapshots that are already gone. Please inspect `getUnavailableSnapshots()` and at minimum log/metric it and avoid tracking unavailable buckets; ideally fail/retry split generation for the table when any requested snapshot is unavailable.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -533,6 +616,145 @@ public void close() throws IOException {
}
}
+ /**
+ * Acquire kv snapshot lease for all {@link TieringSnapshotSplit}s of the given table so that
+ * snapshots referenced by these splits will not be cleaned up by the Fluss server during
+ * tiering. Bucket-snapshot mappings are remembered in {@link #leasedBucketsByTable} so they can
+ * be released on finish/fail.
+ *
+ * <p>Falls back to a warning (no exception thrown) when the server does not support the kv
+ * snapshot lease API, to preserve compatibility with older Fluss clusters.
+ */
+ private void maybeAcquireKvSnapshotLease(long tableId, List<TieringSplit> tieringSplits) {
+ if (flussAdmin == null) {
+ return;
+ }
+ Map<TableBucket, Long> bucketsToLease = new HashMap<>();
+ for (TieringSplit split : tieringSplits) {
+ if (split.isTieringSnapshotSplit()) {
+ TieringSnapshotSplit snapshotSplit = split.asTieringSnapshotSplit();
+ bucketsToLease.put(snapshotSplit.getTableBucket(), snapshotSplit.getSnapshotId());
+ }
+ }
+ if (bucketsToLease.isEmpty()) {
+ return;
+ }
+ LOG.info(
+ "Try to acquire kv snapshot lease {} for tiering table {} with {} buckets.",
+ kvSnapshotLeaseId,
+ tableId,
+ bucketsToLease.size());
+ try {
+ flussAdmin
+ .createKvSnapshotLease(kvSnapshotLeaseId, KV_SNAPSHOT_LEASE_DURATION_MS)
+ .acquireSnapshots(bucketsToLease)
+ .get();
+ leasedBucketsByTable
+ .computeIfAbsent(tableId, k -> ConcurrentHashMap.newKeySet())
+ .addAll(bucketsToLease.keySet());
+ } catch (Exception e) {
+ if (ExceptionUtils.findThrowable(e, UnsupportedVersionException.class).isPresent()) {
+ LOG.warn(
+ "Failed to acquire kv snapshot lease for tiering table {} because the "
+ + "server does not support kv snapshot lease API. Snapshots may "
+ + "be cleaned up earlier than expected. Please upgrade the Fluss "
+ + "server to version 0.9 or later.",
+ tableId,
+ e);
+ } else {
+ LOG.error(
+ "Failed to acquire kv snapshot lease for tiering table {}. "
+ + "Tiering will proceed without snapshot protection; the "
+ + "snapshot may be garbage-collected while tiering is in progress.",
+ tableId,
+ e);
+ }
+ }
+ }
+
+ /**
+ * Release the kv snapshot lease held for a specific table. Called when a table finishes
+ * tiering, fails, or is abandoned due to failover. Missing leases (log-only tables, or tables
+ * for which acquire failed) are handled as no-ops.
+ */
+ private void maybeReleaseKvSnapshotLease(long tableId) {
+ Set<TableBucket> buckets = leasedBucketsByTable.remove(tableId);
+ if (flussAdmin == null || buckets == null || buckets.isEmpty()) {
+ return;
+ }
+ LOG.info(
+ "Try to release kv snapshot lease {} for tiering table {} with {} buckets.",
+ kvSnapshotLeaseId,
+ tableId,
+ buckets.size());
+ try {
+ flussAdmin
+ .createKvSnapshotLease(kvSnapshotLeaseId, KV_SNAPSHOT_LEASE_DURATION_MS)
+ .releaseSnapshots(buckets)
+ .get();
+ } catch (Exception e) {
+ if (ExceptionUtils.findThrowable(e, UnsupportedVersionException.class).isPresent()) {
+ LOG.warn(
+ "Failed to release kv snapshot lease for tiering table {} because the "
+ + "server does not support kv snapshot lease API.",
+ tableId,
+ e);
+ } else {
+ LOG.error("Failed to release kv snapshot lease for tiering table {}.", tableId, e);
+ }
+ }
+ }
+
+ /**
+ * Asynchronous variant of {@link #maybeReleaseKvSnapshotLease(long)} used during failover to
+ * avoid blocking the coordinator thread when multiple tables need to be released.
+ */
+ private void maybeReleaseKvSnapshotLeaseAsync(long tableId) {
+ Set<TableBucket> buckets = leasedBucketsByTable.remove(tableId);
+ if (flussAdmin == null || buckets == null || buckets.isEmpty()) {
+ return;
+ }
Review Comment:
Same as the sync path: this async release removes the table entry from `leasedBucketsByTable` before the release completes. If the future completes exceptionally, there is no way to retry later, so the lease can remain held until expiry. Consider keeping the buckets tracked until completion, and restoring them on failure to allow later retries/cleanup.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] Support kv snapshot lease in tiering service (#2898) [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #3286:
URL: https://github.com/apache/fluss/pull/3286#discussion_r3212825081
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -245,11 +307,23 @@ public void addReader(int subtaskId) {
int globalMaxAttempt = max(maxAttempts);
if (maxAttempts.size() == 1 && globalMaxAttempt >= 1) {
LOG.info(
- "Failover completed. All {} subtasks reached the same attempt number {}. Current registered readers are {}",
+ "All {} subtasks reached the same attempt number {}. Current registered readers are {}. Waiting for failed-table report to complete before clearing failover state.",
Review Comment:
why change this failover logic?
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -313,10 +391,18 @@ private void handleSourceReaderFailOver() {
tieringTableEpochs);
// we need to make all as failed
failedTableEpochs.putAll(new HashMap<>(tieringTableEpochs));
+ // release all currently leased buckets for tables that are being marked failed;
+ // take a snapshot of the keys first to avoid concurrent modification.
+ Set<Long> tableIdsToRelease = new HashSet<>(tieringTableEpochs.keySet());
tieringTableEpochs.clear();
tieringReachMaxDurationsTables.clear();
// also clean all pending splits since we mark all as failed
pendingSplits.clear();
+ // Release leases asynchronously to avoid blocking the coordinator thread when
+ // multiple tables are involved and RPC calls may time out.
+ for (Long tableId : tableIdsToRelease) {
Review Comment:
can we just release all in `leasedBucketsByTable`? So that we won't need to collect `tableIdsToRelease`.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -96,6 +102,19 @@ public class TieringSourceEnumerator
private static final Logger LOG = LoggerFactory.getLogger(TieringSourceEnumerator.class);
+ /**
+ * KV snapshot lease duration for the whole tiering job. One lease covers the entire job
+ * lifecycle; it is renewed implicitly by every {@code acquireSnapshots} call, so a relatively
+ * long duration is safe and also bounds the worst-case leaked-lease lifetime if the job dies
+ * abnormally.
+ *
+ * <p>TODO: introduce an explicit periodic lease-renewal mechanism so that a single tiering
+ * round that exceeds {@link #KV_SNAPSHOT_LEASE_DURATION_MS} (e.g. for very large tables) will
Review Comment:
nit: Javadoc pointing to itself
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java:
##########
@@ -696,13 +705,25 @@ private void waitUntilTieringTableSplitAssignmentReady(
int expectedSplitsNum,
long sleepMs)
throws Throwable {
+ long startTime = System.currentTimeMillis();
+ long timeoutMs = 10000; // 10秒超时
+
while (context.getSplitsAssignmentSequence().size() < expectedSplitsNum) {
+ if (System.currentTimeMillis() - startTime > timeoutMs) {
+ throw new AssertionError(
+ String.format(
+ "等待分配超时: 期望 %d 个分配, 实际 %d 个分配, 超时时间 %dms",
Review Comment:
why change this?
Also, please use engligh
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -184,6 +233,19 @@ public void handleSplitRequest(int subtaskId, @Nullable String requesterHostname
LOG.info("TieringSourceReader {} requests split.", subtaskId);
readersAwaitingSplit.add(subtaskId);
+ // During failover we must not request a new tiering table or assign any splits.
+ // Otherwise we might re-acquire leases for a table whose leases were just released by
+ // handleSourceReaderFailOver, or assign splits to subtasks whose readers from the new
+ // attempt have not been fully registered yet. The pending failed-table report and the
+ // subsequent split request will be driven by the periodic callAsync once failover is
+ // marked complete.
+ if (isFailOvering) {
Review Comment:
why change this failover logic?
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -533,6 +641,208 @@ public void close() throws IOException {
}
}
+ /**
+ * Acquire kv snapshot lease for all {@link TieringSnapshotSplit}s of the given table so that
+ * snapshots referenced by these splits will not be cleaned up by the Fluss server during
+ * tiering. Bucket-snapshot mappings are remembered in {@link #leasedBucketsByTable} so they can
+ * be released on finish/fail.
+ *
+ * <p>Falls back to a warning (no exception thrown) when the server does not support the kv
+ * snapshot lease API, to preserve compatibility with older Fluss clusters.
+ */
+ private void maybeAcquireKvSnapshotLease(long tableId, List<TieringSplit> tieringSplits) {
+ if (flussAdmin == null) {
+ return;
+ }
+ Map<TableBucket, Long> bucketsToLease = new HashMap<>();
+ for (TieringSplit split : tieringSplits) {
+ if (split.isTieringSnapshotSplit()) {
+ TieringSnapshotSplit snapshotSplit = split.asTieringSnapshotSplit();
+ bucketsToLease.put(snapshotSplit.getTableBucket(), snapshotSplit.getSnapshotId());
+ }
+ }
+ if (bucketsToLease.isEmpty()) {
+ return;
+ }
+ LOG.info(
+ "Try to acquire kv snapshot lease {} for tiering table {} with {} buckets.",
+ kvSnapshotLeaseId,
+ tableId,
+ bucketsToLease.size());
+ try {
+ Set<TableBucket> unavailableBuckets =
+ flussAdmin
+ .createKvSnapshotLease(kvSnapshotLeaseId, KV_SNAPSHOT_LEASE_DURATION_MS)
+ .acquireSnapshots(bucketsToLease)
+ .get()
+ .getUnavailableTableBucketSet();
+ // Only record successfully leased buckets so we don't later try to release
+ // buckets that were never actually acquired (e.g. snapshots already GC'ed or
+ // missing on the server).
+ Set<TableBucket> acquiredBuckets = new HashSet<>(bucketsToLease.keySet());
+ if (!unavailableBuckets.isEmpty()) {
+ LOG.warn(
+ "Failed to acquire kv snapshot lease for {} of {} buckets of tiering "
+ + "table {}: {}. The corresponding snapshots may have already "
+ + "been garbage-collected; tiering for those buckets may fail "
+ + "later when the snapshots are accessed.",
+ unavailableBuckets.size(),
+ bucketsToLease.size(),
+ tableId,
+ unavailableBuckets);
+ acquiredBuckets.removeAll(unavailableBuckets);
+ }
+ if (!acquiredBuckets.isEmpty()) {
+ leasedBucketsByTable
+ .computeIfAbsent(tableId, k -> ConcurrentHashMap.newKeySet())
+ .addAll(acquiredBuckets);
+ }
+ } catch (Exception e) {
+ if (ExceptionUtils.findThrowable(e, UnsupportedVersionException.class).isPresent()) {
+ LOG.warn(
+ "Failed to acquire kv snapshot lease for tiering table {} because the "
+ + "server does not support kv snapshot lease API. Snapshots may "
+ + "be cleaned up earlier than expected. Please upgrade the Fluss "
+ + "server to version 0.9 or later.",
+ tableId,
+ e);
+ } else {
+ LOG.error(
+ "Failed to acquire kv snapshot lease for tiering table {}. "
+ + "Tiering will proceed without snapshot protection; the "
+ + "snapshot may be garbage-collected while tiering is in progress.",
+ tableId,
+ e);
+ }
+ }
+ }
+
+ /**
+ * Release the kv snapshot lease held for a specific table. Called when a table finishes
+ * tiering, fails, or is abandoned due to failover. Missing leases (log-only tables, or tables
+ * for which acquire failed) are handled as no-ops.
+ */
+ private void maybeReleaseKvSnapshotLease(long tableId) {
+ // Peek the buckets without removing so that, if the release RPC fails, we can keep
+ // the entry in leasedBucketsByTable for a later retry instead of permanently
+ // forgetting which buckets are still leased on the server side.
+ Set<TableBucket> buckets = leasedBucketsByTable.get(tableId);
+ if (flussAdmin == null || buckets == null || buckets.isEmpty()) {
+ // Nothing to release; clean up any empty entry.
+ leasedBucketsByTable.remove(tableId);
+ return;
+ }
+ LOG.info(
+ "Try to release kv snapshot lease {} for tiering table {} with {} buckets.",
+ kvSnapshotLeaseId,
+ tableId,
+ buckets.size());
+ // Take a defensive copy of the buckets to release so concurrent updates to the
+ // tracked set do not affect the in-flight RPC payload.
+ Set<TableBucket> bucketsToRelease = new HashSet<>(buckets);
+ try {
+ flussAdmin
+ .createKvSnapshotLease(kvSnapshotLeaseId, KV_SNAPSHOT_LEASE_DURATION_MS)
+ .releaseSnapshots(bucketsToRelease)
+ .get();
+ // Only drop the bookkeeping entry after the server confirms the release.
+ Set<TableBucket> tracked = leasedBucketsByTable.get(tableId);
+ if (tracked != null) {
+ tracked.removeAll(bucketsToRelease);
+ if (tracked.isEmpty()) {
+ leasedBucketsByTable.remove(tableId);
+ }
+ }
+ } catch (Exception e) {
+ if (ExceptionUtils.findThrowable(e, UnsupportedVersionException.class).isPresent()) {
+ // Server does not support the lease API; drop tracking since release is a
+ // no-op and there is no point retrying.
+ leasedBucketsByTable.remove(tableId);
+ LOG.warn(
+ "Failed to release kv snapshot lease for tiering table {} because the "
+ + "server does not support kv snapshot lease API.",
+ tableId,
+ e);
+ } else {
+ // Keep the buckets tracked so we (or the next failover/close) can retry.
+ LOG.error(
+ "Failed to release kv snapshot lease for tiering table {}; the buckets "
+ + "remain tracked and will be retried on next release attempt.",
+ tableId,
+ e);
+ }
+ }
+ }
+
+ /**
+ * Asynchronous variant of {@link #maybeReleaseKvSnapshotLease(long)} used during failover to
+ * avoid blocking the coordinator thread when multiple tables need to be released.
+ */
+ private void maybeReleaseKvSnapshotLeaseAsync(long tableId) {
Review Comment:
maybeReleaseKvSnapshotLeaseAsync is most same to maybeReleaseKvSnapshotLease, can we just keep one single method?
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -533,6 +641,208 @@ public void close() throws IOException {
}
}
+ /**
+ * Acquire kv snapshot lease for all {@link TieringSnapshotSplit}s of the given table so that
+ * snapshots referenced by these splits will not be cleaned up by the Fluss server during
+ * tiering. Bucket-snapshot mappings are remembered in {@link #leasedBucketsByTable} so they can
+ * be released on finish/fail.
+ *
+ * <p>Falls back to a warning (no exception thrown) when the server does not support the kv
+ * snapshot lease API, to preserve compatibility with older Fluss clusters.
+ */
+ private void maybeAcquireKvSnapshotLease(long tableId, List<TieringSplit> tieringSplits) {
+ if (flussAdmin == null) {
+ return;
+ }
+ Map<TableBucket, Long> bucketsToLease = new HashMap<>();
+ for (TieringSplit split : tieringSplits) {
+ if (split.isTieringSnapshotSplit()) {
+ TieringSnapshotSplit snapshotSplit = split.asTieringSnapshotSplit();
+ bucketsToLease.put(snapshotSplit.getTableBucket(), snapshotSplit.getSnapshotId());
+ }
+ }
+ if (bucketsToLease.isEmpty()) {
+ return;
+ }
+ LOG.info(
+ "Try to acquire kv snapshot lease {} for tiering table {} with {} buckets.",
+ kvSnapshotLeaseId,
+ tableId,
+ bucketsToLease.size());
+ try {
+ Set<TableBucket> unavailableBuckets =
+ flussAdmin
+ .createKvSnapshotLease(kvSnapshotLeaseId, KV_SNAPSHOT_LEASE_DURATION_MS)
+ .acquireSnapshots(bucketsToLease)
+ .get()
+ .getUnavailableTableBucketSet();
+ // Only record successfully leased buckets so we don't later try to release
+ // buckets that were never actually acquired (e.g. snapshots already GC'ed or
+ // missing on the server).
+ Set<TableBucket> acquiredBuckets = new HashSet<>(bucketsToLease.keySet());
+ if (!unavailableBuckets.isEmpty()) {
+ LOG.warn(
+ "Failed to acquire kv snapshot lease for {} of {} buckets of tiering "
+ + "table {}: {}. The corresponding snapshots may have already "
+ + "been garbage-collected; tiering for those buckets may fail "
+ + "later when the snapshots are accessed.",
+ unavailableBuckets.size(),
+ bucketsToLease.size(),
+ tableId,
+ unavailableBuckets);
+ acquiredBuckets.removeAll(unavailableBuckets);
+ }
+ if (!acquiredBuckets.isEmpty()) {
+ leasedBucketsByTable
+ .computeIfAbsent(tableId, k -> ConcurrentHashMap.newKeySet())
+ .addAll(acquiredBuckets);
+ }
+ } catch (Exception e) {
+ if (ExceptionUtils.findThrowable(e, UnsupportedVersionException.class).isPresent()) {
+ LOG.warn(
+ "Failed to acquire kv snapshot lease for tiering table {} because the "
+ + "server does not support kv snapshot lease API. Snapshots may "
+ + "be cleaned up earlier than expected. Please upgrade the Fluss "
+ + "server to version 0.9 or later.",
+ tableId,
+ e);
+ } else {
+ LOG.error(
+ "Failed to acquire kv snapshot lease for tiering table {}. "
+ + "Tiering will proceed without snapshot protection; the "
+ + "snapshot may be garbage-collected while tiering is in progress.",
+ tableId,
+ e);
+ }
+ }
+ }
+
+ /**
+ * Release the kv snapshot lease held for a specific table. Called when a table finishes
+ * tiering, fails, or is abandoned due to failover. Missing leases (log-only tables, or tables
+ * for which acquire failed) are handled as no-ops.
+ */
+ private void maybeReleaseKvSnapshotLease(long tableId) {
+ // Peek the buckets without removing so that, if the release RPC fails, we can keep
+ // the entry in leasedBucketsByTable for a later retry instead of permanently
+ // forgetting which buckets are still leased on the server side.
+ Set<TableBucket> buckets = leasedBucketsByTable.get(tableId);
+ if (flussAdmin == null || buckets == null || buckets.isEmpty()) {
+ // Nothing to release; clean up any empty entry.
+ leasedBucketsByTable.remove(tableId);
+ return;
+ }
+ LOG.info(
+ "Try to release kv snapshot lease {} for tiering table {} with {} buckets.",
+ kvSnapshotLeaseId,
+ tableId,
+ buckets.size());
+ // Take a defensive copy of the buckets to release so concurrent updates to the
+ // tracked set do not affect the in-flight RPC payload.
+ Set<TableBucket> bucketsToRelease = new HashSet<>(buckets);
+ try {
+ flussAdmin
+ .createKvSnapshotLease(kvSnapshotLeaseId, KV_SNAPSHOT_LEASE_DURATION_MS)
+ .releaseSnapshots(bucketsToRelease)
+ .get();
+ // Only drop the bookkeeping entry after the server confirms the release.
+ Set<TableBucket> tracked = leasedBucketsByTable.get(tableId);
+ if (tracked != null) {
+ tracked.removeAll(bucketsToRelease);
+ if (tracked.isEmpty()) {
+ leasedBucketsByTable.remove(tableId);
+ }
+ }
+ } catch (Exception e) {
+ if (ExceptionUtils.findThrowable(e, UnsupportedVersionException.class).isPresent()) {
+ // Server does not support the lease API; drop tracking since release is a
+ // no-op and there is no point retrying.
+ leasedBucketsByTable.remove(tableId);
+ LOG.warn(
+ "Failed to release kv snapshot lease for tiering table {} because the "
+ + "server does not support kv snapshot lease API.",
+ tableId,
+ e);
+ } else {
+ // Keep the buckets tracked so we (or the next failover/close) can retry.
Review Comment:
when will we do the retry?
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -96,6 +102,19 @@ public class TieringSourceEnumerator
private static final Logger LOG = LoggerFactory.getLogger(TieringSourceEnumerator.class);
+ /**
+ * KV snapshot lease duration for the whole tiering job. One lease covers the entire job
+ * lifecycle; it is renewed implicitly by every {@code acquireSnapshots} call, so a relatively
+ * long duration is safe and also bounds the worst-case leaked-lease lifetime if the job dies
+ * abnormally.
+ *
+ * <p>TODO: introduce an explicit periodic lease-renewal mechanism so that a single tiering
+ * round that exceeds {@link #KV_SNAPSHOT_LEASE_DURATION_MS} (e.g. for very large tables) will
+ * not see its snapshots garbage-collected mid-flight. Tracked as a follow-up issue; tiering
+ * rounds are typically minute-level today so a 1-day lease is sufficient in practice.
+ */
+ private static final long KV_SNAPSHOT_LEASE_DURATION_MS = Duration.ofDays(1).toMillis();
Review Comment:
change to `Duration.ofHours(6)`? I think 6 hource should be fine for most of case.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/state/TieringSourceEnumeratorState.java:
##########
@@ -19,23 +19,55 @@
import org.apache.fluss.flink.tiering.source.enumerator.TieringSourceEnumerator;
-/** The marker class of stateless component {@link TieringSourceEnumerator}. */
+import javax.annotation.Nullable;
+
+import java.util.Objects;
+
+/**
+ * The state of the {@link TieringSourceEnumerator}. Stores the KV snapshot lease id so that it can
+ * be recovered after a checkpoint restore, avoiding orphaned leases on the server side.
+ */
public class TieringSourceEnumeratorState {
Review Comment:
It will breaks the statless design of tiering service. I think we can just ignore the kvSnapshotLeaseId and delegate fluss cluster to do snapshot ttl.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -533,6 +641,208 @@ public void close() throws IOException {
}
}
+ /**
+ * Acquire kv snapshot lease for all {@link TieringSnapshotSplit}s of the given table so that
+ * snapshots referenced by these splits will not be cleaned up by the Fluss server during
+ * tiering. Bucket-snapshot mappings are remembered in {@link #leasedBucketsByTable} so they can
+ * be released on finish/fail.
+ *
+ * <p>Falls back to a warning (no exception thrown) when the server does not support the kv
+ * snapshot lease API, to preserve compatibility with older Fluss clusters.
+ */
+ private void maybeAcquireKvSnapshotLease(long tableId, List<TieringSplit> tieringSplits) {
+ if (flussAdmin == null) {
Review Comment:
when flussAdmin will be null?
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -533,6 +641,208 @@ public void close() throws IOException {
}
}
+ /**
+ * Acquire kv snapshot lease for all {@link TieringSnapshotSplit}s of the given table so that
+ * snapshots referenced by these splits will not be cleaned up by the Fluss server during
+ * tiering. Bucket-snapshot mappings are remembered in {@link #leasedBucketsByTable} so they can
+ * be released on finish/fail.
+ *
+ * <p>Falls back to a warning (no exception thrown) when the server does not support the kv
+ * snapshot lease API, to preserve compatibility with older Fluss clusters.
+ */
+ private void maybeAcquireKvSnapshotLease(long tableId, List<TieringSplit> tieringSplits) {
+ if (flussAdmin == null) {
+ return;
+ }
+ Map<TableBucket, Long> bucketsToLease = new HashMap<>();
+ for (TieringSplit split : tieringSplits) {
+ if (split.isTieringSnapshotSplit()) {
+ TieringSnapshotSplit snapshotSplit = split.asTieringSnapshotSplit();
+ bucketsToLease.put(snapshotSplit.getTableBucket(), snapshotSplit.getSnapshotId());
+ }
+ }
+ if (bucketsToLease.isEmpty()) {
+ return;
+ }
+ LOG.info(
+ "Try to acquire kv snapshot lease {} for tiering table {} with {} buckets.",
+ kvSnapshotLeaseId,
+ tableId,
+ bucketsToLease.size());
+ try {
+ Set<TableBucket> unavailableBuckets =
+ flussAdmin
+ .createKvSnapshotLease(kvSnapshotLeaseId, KV_SNAPSHOT_LEASE_DURATION_MS)
+ .acquireSnapshots(bucketsToLease)
+ .get()
+ .getUnavailableTableBucketSet();
+ // Only record successfully leased buckets so we don't later try to release
+ // buckets that were never actually acquired (e.g. snapshots already GC'ed or
+ // missing on the server).
+ Set<TableBucket> acquiredBuckets = new HashSet<>(bucketsToLease.keySet());
+ if (!unavailableBuckets.isEmpty()) {
+ LOG.warn(
+ "Failed to acquire kv snapshot lease for {} of {} buckets of tiering "
+ + "table {}: {}. The corresponding snapshots may have already "
+ + "been garbage-collected; tiering for those buckets may fail "
+ + "later when the snapshots are accessed.",
+ unavailableBuckets.size(),
+ bucketsToLease.size(),
+ tableId,
+ unavailableBuckets);
+ acquiredBuckets.removeAll(unavailableBuckets);
+ }
+ if (!acquiredBuckets.isEmpty()) {
+ leasedBucketsByTable
+ .computeIfAbsent(tableId, k -> ConcurrentHashMap.newKeySet())
+ .addAll(acquiredBuckets);
+ }
+ } catch (Exception e) {
+ if (ExceptionUtils.findThrowable(e, UnsupportedVersionException.class).isPresent()) {
+ LOG.warn(
+ "Failed to acquire kv snapshot lease for tiering table {} because the "
+ + "server does not support kv snapshot lease API. Snapshots may "
+ + "be cleaned up earlier than expected. Please upgrade the Fluss "
+ + "server to version 0.9 or later.",
+ tableId,
+ e);
+ } else {
+ LOG.error(
+ "Failed to acquire kv snapshot lease for tiering table {}. "
+ + "Tiering will proceed without snapshot protection; the "
+ + "snapshot may be garbage-collected while tiering is in progress.",
+ tableId,
+ e);
+ }
+ }
+ }
+
+ /**
+ * Release the kv snapshot lease held for a specific table. Called when a table finishes
+ * tiering, fails, or is abandoned due to failover. Missing leases (log-only tables, or tables
+ * for which acquire failed) are handled as no-ops.
+ */
+ private void maybeReleaseKvSnapshotLease(long tableId) {
+ // Peek the buckets without removing so that, if the release RPC fails, we can keep
+ // the entry in leasedBucketsByTable for a later retry instead of permanently
+ // forgetting which buckets are still leased on the server side.
+ Set<TableBucket> buckets = leasedBucketsByTable.get(tableId);
+ if (flussAdmin == null || buckets == null || buckets.isEmpty()) {
+ // Nothing to release; clean up any empty entry.
+ leasedBucketsByTable.remove(tableId);
+ return;
+ }
+ LOG.info(
+ "Try to release kv snapshot lease {} for tiering table {} with {} buckets.",
+ kvSnapshotLeaseId,
+ tableId,
+ buckets.size());
+ // Take a defensive copy of the buckets to release so concurrent updates to the
+ // tracked set do not affect the in-flight RPC payload.
+ Set<TableBucket> bucketsToRelease = new HashSet<>(buckets);
+ try {
+ flussAdmin
+ .createKvSnapshotLease(kvSnapshotLeaseId, KV_SNAPSHOT_LEASE_DURATION_MS)
+ .releaseSnapshots(bucketsToRelease)
+ .get();
+ // Only drop the bookkeeping entry after the server confirms the release.
+ Set<TableBucket> tracked = leasedBucketsByTable.get(tableId);
Review Comment:
can we just call `leasedBucketsByTable.remove(tableId);`?
IIUC, `bucketsToRelease` is same with the buckets in the `leasedBucketsByTable`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org