From d9bc1a60b4d94560a3a4b09e5ffb52ae8e164453 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=99=BD=E9=B5=BA?= Date: Mon, 6 Jul 2026 17:57:49 +0800 Subject: [PATCH] [flink-action][common][server] Support KV shared SST orphan file cleanup --- .../fluss/client/admin/FlussAdminITCase.java | 4 +- .../fluss}/kv/autoinc/AutoIncIDRange.java | 2 +- .../fluss}/kv/snapshot/CompletedSnapshot.java | 23 +-- .../fluss}/kv/snapshot/KvFileHandle.java | 2 +- .../kv/snapshot/KvFileHandleAndLocalPath.java | 2 +- .../fluss}/kv/snapshot/KvSnapshotHandle.java | 63 ++---- .../fluss/kv/snapshot}/SnapshotUtil.java | 4 +- .../json}/CompletedSnapshotJsonSerde.java | 148 ++++++++++---- .../json/KvSnapshotMetadataJsonSerdeTest.java | 155 +++++++++++++++ .../action/orphan/audit/AuditLogger.java | 15 ++ .../orphan/build/ActiveRefsFetcher.java | 55 ++++++ .../orphan/build/KvSharedSstFetchResult.java | 88 +++++++++ .../action/orphan/job/BucketCleanTask.java | 11 ++ .../action/orphan/job/BucketCleaner.java | 4 - .../orphan/job/ScanAndCleanFunction.java | 3 +- .../orphan/job/ScopeEnumeratorFunction.java | 11 ++ .../action/orphan/rule/BucketActiveRefs.java | 33 +++- .../action/orphan/rule/KvSharedSstRule.java | 27 ++- .../action/orphan/OrphanFilesCleanITCase.java | 181 ++++++++++++++++++ .../orphan/build/ActiveRefsFetcherTest.java | 109 +++++++++++ .../action/orphan/job/BucketCleanerTest.java | 56 ++++++ .../orphan/rule/KvSharedSstRuleTest.java | 42 +++- .../apache/fluss/server/RpcServiceBase.java | 2 +- .../CompletedSnapshotStoreManager.java | 2 +- .../CoordinatorEventProcessor.java | 2 +- .../coordinator/CoordinatorService.java | 4 +- .../lease/KvSnapshotLeaseManager.java | 2 +- .../server/entity/CommitKvSnapshotData.java | 2 +- .../fluss/server/kv/KvRecoverHelper.java | 2 +- .../org/apache/fluss/server/kv/KvTablet.java | 4 +- .../kv/autoinc/AutoIncrementManager.java | 1 + .../BoundedSegmentSequenceGenerator.java | 1 + .../server/kv/autoinc/SequenceGenerator.java | 2 + .../CompletedKvSnapshotCommitter.java | 2 + .../kv/snapshot/CompletedSnapshotHandle.java | 2 + .../kv/snapshot/CompletedSnapshotStore.java | 6 +- .../DefaultCompletedKvSnapshotCommitter.java | 1 + .../kv/snapshot/DefaultSnapshotContext.java | 1 + .../kv/snapshot/KvSnapshotDataDownloader.java | 1 + .../kv/snapshot/KvSnapshotDataUploader.java | 4 +- .../kv/snapshot/KvSnapshotDownloadSpec.java | 2 + .../kv/snapshot/KvTabletSnapshotTarget.java | 1 + .../kv/snapshot/PeriodicSnapshotManager.java | 2 +- .../kv/snapshot/PlaceholderKvFileHandler.java | 2 + .../kv/snapshot/RocksIncrementalSnapshot.java | 3 + .../kv/snapshot/SharedKvFileRegistry.java | 11 +- .../kv/snapshot/SharedKvFileRegistryKey.java | 2 + .../server/kv/snapshot/SnapshotContext.java | 1 + .../server/kv/snapshot/SnapshotLocation.java | 1 + .../server/kv/snapshot/SnapshotResult.java | 1 + .../server/kv/snapshot/SnapshotsCleaner.java | 2 + .../fluss/server/kv/snapshot/TabletState.java | 2 +- .../apache/fluss/server/replica/Replica.java | 6 +- .../server/utils/ServerRpcMessageUtils.java | 6 +- .../CompletedSnapshotStoreManagerTest.java | 2 +- .../CoordinatorEventProcessorTest.java | 2 +- .../CoordinatorServiceOrphanRpcsITCase.java | 4 +- .../autoinc/SegmentSequenceGeneratorTest.java | 1 + .../CompletedSnapshotJsonSerdeTest.java | 9 +- .../snapshot/CompletedSnapshotStoreTest.java | 4 + .../kv/snapshot/CompletedSnapshotTest.java | 12 +- .../KvSnapshotDataDownloaderTest.java | 3 + .../snapshot/KvSnapshotDataUploaderTest.java | 2 + .../snapshot/KvTabletSnapshotTargetTest.java | 2 + .../RocksIncrementalSnapshotTest.java | 4 +- .../kv/snapshot/SharedKvFileRegistryTest.java | 2 + .../kv/snapshot/SnapshotsCleanerTest.java | 6 +- .../TestingCompletedKvSnapshotCommitter.java | 1 + .../TestingCompletedSnapshotHandle.java | 7 +- .../ZooKeeperCompletedSnapshotStoreTest.java | 1 + .../replica/KvRecoverFromRemoteLogITCase.java | 2 +- .../replica/KvReplicaRestoreITCase.java | 2 +- .../server/replica/KvSnapshotITCase.java | 2 +- .../server/replica/ReplicaManagerTest.java | 2 +- .../fluss/server/replica/ReplicaTest.java | 4 +- .../fluss/server/replica/ReplicaTestBase.java | 2 +- .../testutils/FlussClusterExtension.java | 2 +- .../fluss/server/testutils/KvTestUtils.java | 6 +- 78 files changed, 1032 insertions(+), 173 deletions(-) rename {fluss-server/src/main/java/org/apache/fluss/server => fluss-common/src/main/java/org/apache/fluss}/kv/autoinc/AutoIncIDRange.java (98%) rename {fluss-server/src/main/java/org/apache/fluss/server => fluss-common/src/main/java/org/apache/fluss}/kv/snapshot/CompletedSnapshot.java (90%) rename {fluss-server/src/main/java/org/apache/fluss/server => fluss-common/src/main/java/org/apache/fluss}/kv/snapshot/KvFileHandle.java (98%) rename {fluss-server/src/main/java/org/apache/fluss/server => fluss-common/src/main/java/org/apache/fluss}/kv/snapshot/KvFileHandleAndLocalPath.java (98%) rename {fluss-server/src/main/java/org/apache/fluss/server => fluss-common/src/main/java/org/apache/fluss}/kv/snapshot/KvSnapshotHandle.java (66%) rename {fluss-server/src/main/java/org/apache/fluss/server/utils => fluss-common/src/main/java/org/apache/fluss/kv/snapshot}/SnapshotUtil.java (96%) rename {fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot => fluss-common/src/main/java/org/apache/fluss/utils/json}/CompletedSnapshotJsonSerde.java (65%) create mode 100644 fluss-common/src/test/java/org/apache/fluss/utils/json/KvSnapshotMetadataJsonSerdeTest.java create mode 100644 fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/build/KvSharedSstFetchResult.java diff --git a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java index 68181844ea..aa9f5313e1 100644 --- a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java +++ b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java @@ -57,6 +57,8 @@ import org.apache.fluss.exception.TooManyPartitionsException; import org.apache.fluss.fs.FsPath; import org.apache.fluss.fs.FsPathAndFileName; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; +import org.apache.fluss.kv.snapshot.KvSnapshotHandle; import org.apache.fluss.metadata.AggFunctions; import org.apache.fluss.metadata.DatabaseChange; import org.apache.fluss.metadata.DatabaseDescriptor; @@ -77,8 +79,6 @@ import org.apache.fluss.rpc.messages.ListOffsetsRequest; import org.apache.fluss.rpc.messages.ListOffsetsResponse; import org.apache.fluss.server.coordinator.CoordinatorContext; -import org.apache.fluss.server.kv.snapshot.CompletedSnapshot; -import org.apache.fluss.server.kv.snapshot.KvSnapshotHandle; import org.apache.fluss.server.log.LogTablet; import org.apache.fluss.server.metadata.ServerInfo; import org.apache.fluss.server.replica.Replica; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/AutoIncIDRange.java b/fluss-common/src/main/java/org/apache/fluss/kv/autoinc/AutoIncIDRange.java similarity index 98% rename from fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/AutoIncIDRange.java rename to fluss-common/src/main/java/org/apache/fluss/kv/autoinc/AutoIncIDRange.java index 3c1297c670..d34598149d 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/AutoIncIDRange.java +++ b/fluss-common/src/main/java/org/apache/fluss/kv/autoinc/AutoIncIDRange.java @@ -17,7 +17,7 @@ * under the License. */ -package org.apache.fluss.server.kv.autoinc; +package org.apache.fluss.kv.autoinc; import java.util.Objects; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshot.java b/fluss-common/src/main/java/org/apache/fluss/kv/snapshot/CompletedSnapshot.java similarity index 90% rename from fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshot.java rename to fluss-common/src/main/java/org/apache/fluss/kv/snapshot/CompletedSnapshot.java index 86cd25858b..a3994784f8 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshot.java +++ b/fluss-common/src/main/java/org/apache/fluss/kv/snapshot/CompletedSnapshot.java @@ -15,13 +15,13 @@ * limitations under the License. */ -package org.apache.fluss.server.kv.snapshot; +package org.apache.fluss.kv.snapshot; import org.apache.fluss.annotation.VisibleForTesting; import org.apache.fluss.fs.FileSystem; import org.apache.fluss.fs.FsPath; +import org.apache.fluss.kv.autoinc.AutoIncIDRange; import org.apache.fluss.metadata.TableBucket; -import org.apache.fluss.server.kv.autoinc.AutoIncIDRange; import org.apache.fluss.utils.concurrent.FutureUtils; import javax.annotation.Nullable; @@ -106,7 +106,7 @@ public CompletedSnapshot( } @VisibleForTesting - CompletedSnapshot( + public CompletedSnapshot( TableBucket tableBucket, long snapshotID, FsPath snapshotLocation, @@ -152,22 +152,11 @@ public long getSnapshotSize() { return kvSnapshotHandle.getSnapshotSize(); } - /** - * Register all shared kv files in the given registry. This method is called before the snapshot - * is added into the store. - * - * @param sharedKvFileRegistry The registry where shared kv files are registered - */ - public void registerSharedKvFilesAfterRestored(SharedKvFileRegistry sharedKvFileRegistry) { - sharedKvFileRegistry.registerAllAfterRestored(this); - } - public CompletableFuture discardAsync(Executor ioExecutor) { - // it'll discard the snapshot files for kv, it'll always discard - // the private files; for shared files, only if they're not be registered in - // SharedKvRegistry, can the files be deleted. + // Discard private files only; shared files are managed by the SharedKvFileRegistry + // on the server side. CompletableFuture discardKvFuture = - FutureUtils.runAsync(kvSnapshotHandle::discard, ioExecutor); + FutureUtils.runAsync(kvSnapshotHandle::discardPrivateFiles, ioExecutor); CompletableFuture discardMetaFileFuture = FutureUtils.runAsync(this::disposeMetadata, ioExecutor); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvFileHandle.java b/fluss-common/src/main/java/org/apache/fluss/kv/snapshot/KvFileHandle.java similarity index 98% rename from fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvFileHandle.java rename to fluss-common/src/main/java/org/apache/fluss/kv/snapshot/KvFileHandle.java index 7f537815a6..c117c1dd2d 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvFileHandle.java +++ b/fluss-common/src/main/java/org/apache/fluss/kv/snapshot/KvFileHandle.java @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.fluss.server.kv.snapshot; +package org.apache.fluss.kv.snapshot; import org.apache.fluss.fs.FileSystem; import org.apache.fluss.fs.FsPath; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvFileHandleAndLocalPath.java b/fluss-common/src/main/java/org/apache/fluss/kv/snapshot/KvFileHandleAndLocalPath.java similarity index 98% rename from fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvFileHandleAndLocalPath.java rename to fluss-common/src/main/java/org/apache/fluss/kv/snapshot/KvFileHandleAndLocalPath.java index ac9e7d8192..7ce087ecc1 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvFileHandleAndLocalPath.java +++ b/fluss-common/src/main/java/org/apache/fluss/kv/snapshot/KvFileHandleAndLocalPath.java @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.fluss.server.kv.snapshot; +package org.apache.fluss.kv.snapshot; import java.util.Objects; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotHandle.java b/fluss-common/src/main/java/org/apache/fluss/kv/snapshot/KvSnapshotHandle.java similarity index 66% rename from fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotHandle.java rename to fluss-common/src/main/java/org/apache/fluss/kv/snapshot/KvSnapshotHandle.java index a016f98755..85c03bdc3f 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotHandle.java +++ b/fluss-common/src/main/java/org/apache/fluss/kv/snapshot/KvSnapshotHandle.java @@ -15,9 +15,7 @@ * limitations under the License. */ -package org.apache.fluss.server.kv.snapshot; - -import org.apache.fluss.server.utils.SnapshotUtil; +package org.apache.fluss.kv.snapshot; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -26,9 +24,6 @@ import java.util.Objects; import java.util.stream.Collectors; -import static org.apache.fluss.utils.Preconditions.checkNotNull; -import static org.apache.fluss.utils.Preconditions.checkState; - /** * A handle to the snapshot of a kv tablet. It contains the share file handles and the private file * handles from which we can rebuild the kv tablet. @@ -45,16 +40,6 @@ public class KvSnapshotHandle { /** The size of the incremental snapshot. */ private final long incrementalSize; - /** - * Once the shared states are registered, it is the {@link SharedKvFileRegistry}'s - * responsibility to cleanup those shared states. But in the cases where the state handle is - * discarded before performing the registration, the handle should delete all the shared states - * created by it. - * - *

This variable is not null iff the handles was registered. - */ - private transient SharedKvFileRegistry sharedKvFileRegistry; - public KvSnapshotHandle( List sharedFileHandles, List privateFileHandles, @@ -97,43 +82,35 @@ public long getSnapshotSize() { return snapshotSize; } - public void discard() { - SharedKvFileRegistry registry = this.sharedKvFileRegistry; - final boolean isRegistered = (registry != null); - + /** + * Discards only the private files of this snapshot. Use this when shared files are managed + * externally by a registry. + */ + public void discardPrivateFiles() { try { SnapshotUtil.bestEffortDiscardAllKvFiles( privateFileHandles.stream() .map(KvFileHandleAndLocalPath::getKvFileHandle) .collect(Collectors.toList())); } catch (Exception e) { - LOG.warn("Could not properly discard misc file states.", e); - } - - if (!isRegistered) { - try { - SnapshotUtil.bestEffortDiscardAllKvFiles( - sharedFileHandles.stream() - .map(KvFileHandleAndLocalPath::getKvFileHandle) - .collect(Collectors.toSet())); - } catch (Exception e) { - LOG.warn("Could not properly discard new sst file states.", e); - } + LOG.warn("Could not properly discard private file states.", e); } } - public void registerKvFileHandles(SharedKvFileRegistry registry, long snapshotID) { - checkState( - sharedKvFileRegistry != registry, - "The kv file handle has already registered its shared kv files to the given registry."); - - sharedKvFileRegistry = checkNotNull(registry); + /** + * Discards all files (private + shared). Use this for failed or aborted snapshots that were + * never registered with a shared file registry. + */ + public void discardAll() { + discardPrivateFiles(); - for (KvFileHandleAndLocalPath handleAndLocalPath : sharedFileHandles) { - registry.registerReference( - SharedKvFileRegistryKey.fromKvFileHandle(handleAndLocalPath.getKvFileHandle()), - handleAndLocalPath.getKvFileHandle(), - snapshotID); + try { + SnapshotUtil.bestEffortDiscardAllKvFiles( + sharedFileHandles.stream() + .map(KvFileHandleAndLocalPath::getKvFileHandle) + .collect(Collectors.toSet())); + } catch (Exception e) { + LOG.warn("Could not properly discard shared sst file states.", e); } } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/SnapshotUtil.java b/fluss-common/src/main/java/org/apache/fluss/kv/snapshot/SnapshotUtil.java similarity index 96% rename from fluss-server/src/main/java/org/apache/fluss/server/utils/SnapshotUtil.java rename to fluss-common/src/main/java/org/apache/fluss/kv/snapshot/SnapshotUtil.java index 32e679ae11..b4c96f9233 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/utils/SnapshotUtil.java +++ b/fluss-common/src/main/java/org/apache/fluss/kv/snapshot/SnapshotUtil.java @@ -15,9 +15,8 @@ * limitations under the License. */ -package org.apache.fluss.server.utils; +package org.apache.fluss.kv.snapshot; -import org.apache.fluss.server.kv.snapshot.KvFileHandle; import org.apache.fluss.utils.ExceptionUtils; import org.apache.fluss.utils.function.ThrowingConsumer; @@ -44,6 +43,7 @@ public static void bestEffortDiscardAllKvFiles( applyToAllWhileSuppressingExceptions(handlesToDiscard, KvFileHandle::discard); } + /** Quietly discards a single kv file handle, logging any exception. */ public static void discardKvFileQuietly(KvFileHandle kvFileHandle) { if (kvFileHandle == null) { return; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotJsonSerde.java b/fluss-common/src/main/java/org/apache/fluss/utils/json/CompletedSnapshotJsonSerde.java similarity index 65% rename from fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotJsonSerde.java rename to fluss-common/src/main/java/org/apache/fluss/utils/json/CompletedSnapshotJsonSerde.java index 99a7c3d230..5b9d94cee1 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotJsonSerde.java +++ b/fluss-common/src/main/java/org/apache/fluss/utils/json/CompletedSnapshotJsonSerde.java @@ -15,58 +15,78 @@ * limitations under the License. */ -package org.apache.fluss.server.kv.snapshot; +package org.apache.fluss.utils.json; import org.apache.fluss.fs.FsPath; +import org.apache.fluss.kv.autoinc.AutoIncIDRange; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; +import org.apache.fluss.kv.snapshot.KvFileHandle; +import org.apache.fluss.kv.snapshot.KvFileHandleAndLocalPath; +import org.apache.fluss.kv.snapshot.KvSnapshotHandle; import org.apache.fluss.metadata.TableBucket; -import org.apache.fluss.server.kv.autoinc.AutoIncIDRange; import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator; import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode; -import org.apache.fluss.utils.json.JsonDeserializer; -import org.apache.fluss.utils.json.JsonSerdeUtils; -import org.apache.fluss.utils.json.JsonSerializer; import java.io.IOException; import java.util.ArrayList; +import java.util.HashSet; import java.util.List; +import java.util.Set; -/** Json serializer and deserializer for {@link CompletedSnapshot}. */ +/** + * Json serializer and deserializer for {@link CompletedSnapshot}. + * + *

This class is the single source of truth for KV snapshot {@code _METADATA} JSON format + * constants. It provides full serialization/deserialization as well as lightweight extraction + * utilities (e.g., {@link #parseSharedSstLocalPaths(byte[])}) for consumers that do not need a full + * domain object. + */ public class CompletedSnapshotJsonSerde implements JsonSerializer, JsonDeserializer { - public static final CompletedSnapshotJsonSerde INSTANCE = new CompletedSnapshotJsonSerde(); - - private static final int VERSION = 1; - private static final String VERSION_KEY = "version"; - // for table bucket the snapshot belongs to - private static final String TABLE_ID = "table_id"; - private static final String PARTITION_ID = "partition_id"; - private static final String BUCKET_ID = "bucket_id"; - - private static final String SNAPSHOT_ID = "snapshot_id"; - private static final String SNAPSHOT_LOCATION = "snapshot_location"; - - // for kv snapshot's files - private static final String KV_SNAPSHOT_HANDLE = "kv_snapshot_handle"; - private static final String KV_SHARED_FILES_HANDLE = "shared_file_handles"; - private static final String KV_PRIVATE_FILES_HANDLE = "private_file_handles"; - private static final String KV_FILE_HANDLE = "kv_file_handle"; - private static final String KV_FILE_PATH = "path"; - private static final String KV_FILE_SIZE = "size"; - private static final String KV_FILE_LOCAL_PATH = "local_path"; - private static final String SNAPSHOT_INCREMENTAL_SIZE = "snapshot_incremental_size"; - // --------------------------------------------------------------------------------- - // kv tablet state for the snapshot + // JSON field name constants — _METADATA schema version 1 // --------------------------------------------------------------------------------- - // for the next log offset when the snapshot is triggered; - private static final String LOG_OFFSET = "log_offset"; - private static final String ROW_COUNT = "row_count"; - private static final String AUTO_INC_ID_RANGE = "auto_inc_id_range"; - private static final String AUTO_INC_COLUMN_ID = "column_id"; - private static final String AUTO_INC_ID_START = "start"; - private static final String AUTO_INC_ID_END = "end"; + /** Current schema version. */ + public static final int VERSION = 1; + + public static final String VERSION_KEY = "version"; + + // Table bucket identification + public static final String TABLE_ID = "table_id"; + public static final String PARTITION_ID = "partition_id"; + public static final String BUCKET_ID = "bucket_id"; + + // Snapshot identity + public static final String SNAPSHOT_ID = "snapshot_id"; + public static final String SNAPSHOT_LOCATION = "snapshot_location"; + + // KV snapshot handle structure + public static final String KV_SNAPSHOT_HANDLE = "kv_snapshot_handle"; + public static final String KV_SHARED_FILES_HANDLE = "shared_file_handles"; + public static final String KV_PRIVATE_FILES_HANDLE = "private_file_handles"; + + // Individual file entry fields + public static final String KV_FILE_HANDLE = "kv_file_handle"; + public static final String KV_FILE_PATH = "path"; + public static final String KV_FILE_SIZE = "size"; + public static final String KV_FILE_LOCAL_PATH = "local_path"; + + // Snapshot size + public static final String SNAPSHOT_INCREMENTAL_SIZE = "snapshot_incremental_size"; + + // KV tablet state + public static final String LOG_OFFSET = "log_offset"; + public static final String ROW_COUNT = "row_count"; + + // Auto-increment ID ranges + public static final String AUTO_INC_ID_RANGE = "auto_inc_id_range"; + public static final String AUTO_INC_COLUMN_ID = "column_id"; + public static final String AUTO_INC_ID_START = "start"; + public static final String AUTO_INC_ID_END = "end"; + + public static final CompletedSnapshotJsonSerde INSTANCE = new CompletedSnapshotJsonSerde(); @Override public void serialize(CompletedSnapshot completedSnapshot, JsonGenerator generator) @@ -240,6 +260,10 @@ private List deserializeKvFileHandles( return kvFileHandleAndLocalPaths; } + // --------------------------------------------------------------------------------- + // Static convenience methods + // --------------------------------------------------------------------------------- + /** Serialize the {@link CompletedSnapshot} to json bytes. */ public static byte[] toJson(CompletedSnapshot completedSnapshot) { return JsonSerdeUtils.writeValueAsBytes(completedSnapshot, INSTANCE); @@ -249,4 +273,56 @@ public static byte[] toJson(CompletedSnapshot completedSnapshot) { public static CompletedSnapshot fromJson(byte[] json) { return JsonSerdeUtils.readValue(json, INSTANCE); } + + /** + * Parses a {@code _METADATA} JSON payload and returns the set of shared SST file names (local + * path basenames) referenced by the snapshot. + * + *

This method navigates the JSON tree: {@code kv_snapshot_handle → shared_file_handles[*] → + * local_path} and collects non-empty values. + * + * @param metadataJsonBytes raw bytes of the {@code _METADATA} file + * @return set of shared SST file names (e.g. {@code "abc-def-0.sst"}) + * @throws IOException if the bytes cannot be parsed as valid JSON or the expected structure is + * missing + */ + public static Set parseSharedSstLocalPaths(byte[] metadataJsonBytes) + throws IOException { + JsonNode root = JsonSerdeUtils.OBJECT_MAPPER_INSTANCE.readTree(metadataJsonBytes); + JsonNode kvSnapshotHandle = root.get(KV_SNAPSHOT_HANDLE); + if (kvSnapshotHandle == null) { + throw new IOException("Missing '" + KV_SNAPSHOT_HANDLE + "' in _METADATA JSON payload"); + } + JsonNode sharedFilesNode = kvSnapshotHandle.get(KV_SHARED_FILES_HANDLE); + if (sharedFilesNode == null || !sharedFilesNode.isArray()) { + throw new IOException( + "Missing or non-array '" + + KV_SHARED_FILES_HANDLE + + "' in _METADATA JSON payload"); + } + + Set fileNames = new HashSet<>(); + for (JsonNode entry : sharedFilesNode) { + JsonNode localPathNode = entry.get(KV_FILE_LOCAL_PATH); + if (localPathNode == null || !localPathNode.isTextual()) { + throw new IOException( + "Missing or non-textual '" + + KV_FILE_LOCAL_PATH + + "' in " + + KV_SHARED_FILES_HANDLE + + " entry"); + } + String localPath = localPathNode.asText(); + if (localPath.isEmpty()) { + throw new IOException( + "Empty '" + + KV_FILE_LOCAL_PATH + + "' in " + + KV_SHARED_FILES_HANDLE + + " entry"); + } + fileNames.add(localPath); + } + return fileNames; + } } diff --git a/fluss-common/src/test/java/org/apache/fluss/utils/json/KvSnapshotMetadataJsonSerdeTest.java b/fluss-common/src/test/java/org/apache/fluss/utils/json/KvSnapshotMetadataJsonSerdeTest.java new file mode 100644 index 0000000000..75d8f4dd71 --- /dev/null +++ b/fluss-common/src/test/java/org/apache/fluss/utils/json/KvSnapshotMetadataJsonSerdeTest.java @@ -0,0 +1,155 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.utils.json; + +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.Set; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link CompletedSnapshotJsonSerde#parseSharedSstLocalPaths}. */ +class KvSnapshotMetadataJsonSerdeTest { + + @Test + void parsesSharedSstLocalPathsFromValidMetadata() throws IOException { + String json = + "{" + + "\"kv_snapshot_handle\":{" + + " \"shared_file_handles\":[" + + " {\"local_path\":\"abc-001.sst\",\"size\":1024}," + + " {\"local_path\":\"def-002.sst\",\"size\":2048}," + + " {\"local_path\":\"ghi-003.sst\",\"size\":512}" + + " ]" + + "}" + + "}"; + + Set result = + CompletedSnapshotJsonSerde.parseSharedSstLocalPaths( + json.getBytes(StandardCharsets.UTF_8)); + + assertThat(result).containsExactlyInAnyOrder("abc-001.sst", "def-002.sst", "ghi-003.sst"); + } + + @Test + void returnsEmptySetWhenSharedFileHandlesArrayIsEmpty() throws IOException { + String json = "{" + "\"kv_snapshot_handle\":{" + " \"shared_file_handles\":[]" + "}" + "}"; + + Set result = + CompletedSnapshotJsonSerde.parseSharedSstLocalPaths( + json.getBytes(StandardCharsets.UTF_8)); + + assertThat(result).isEmpty(); + } + + @Test + void throwsOnMissingKvSnapshotHandle() { + String json = "{\"other_field\":\"value\"}"; + + assertThatThrownBy( + () -> + CompletedSnapshotJsonSerde.parseSharedSstLocalPaths( + json.getBytes(StandardCharsets.UTF_8))) + .isInstanceOf(IOException.class) + .hasMessageContaining("kv_snapshot_handle"); + } + + @Test + void throwsOnMissingSharedFileHandles() { + String json = "{\"kv_snapshot_handle\":{\"other_field\":\"value\"}}"; + + assertThatThrownBy( + () -> + CompletedSnapshotJsonSerde.parseSharedSstLocalPaths( + json.getBytes(StandardCharsets.UTF_8))) + .isInstanceOf(IOException.class) + .hasMessageContaining("shared_file_handles"); + } + + @Test + void throwsOnMalformedJson() { + String json = "not valid json at all"; + + assertThatThrownBy( + () -> + CompletedSnapshotJsonSerde.parseSharedSstLocalPaths( + json.getBytes(StandardCharsets.UTF_8))) + .isInstanceOf(IOException.class); + } + + @Test + void throwsOnSharedFileHandleMissingLocalPath() { + String json = + "{" + + "\"kv_snapshot_handle\":{" + + " \"shared_file_handles\":[" + + " {\"local_path\":\"valid.sst\",\"size\":100}," + + " {\"size\":200}" + + " ]" + + "}" + + "}"; + + assertThatThrownBy( + () -> + CompletedSnapshotJsonSerde.parseSharedSstLocalPaths( + json.getBytes(StandardCharsets.UTF_8))) + .isInstanceOf(IOException.class) + .hasMessageContaining("local_path"); + } + + @Test + void throwsOnSharedFileHandleEmptyLocalPath() { + String json = + "{" + + "\"kv_snapshot_handle\":{" + + " \"shared_file_handles\":[" + + " {\"local_path\":\"\",\"size\":300}" + + " ]" + + "}" + + "}"; + + assertThatThrownBy( + () -> + CompletedSnapshotJsonSerde.parseSharedSstLocalPaths( + json.getBytes(StandardCharsets.UTF_8))) + .isInstanceOf(IOException.class) + .hasMessageContaining("local_path"); + } + + @Test + void deduplicatesRepeatedLocalPaths() throws IOException { + String json = + "{" + + "\"kv_snapshot_handle\":{" + + " \"shared_file_handles\":[" + + " {\"local_path\":\"same.sst\",\"size\":100}," + + " {\"local_path\":\"same.sst\",\"size\":200}" + + " ]" + + "}" + + "}"; + + Set result = + CompletedSnapshotJsonSerde.parseSharedSstLocalPaths( + json.getBytes(StandardCharsets.UTF_8)); + + assertThat(result).containsExactly("same.sst"); + } +} diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/audit/AuditLogger.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/audit/AuditLogger.java index 26adf5f00e..213ff25db4 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/audit/AuditLogger.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/audit/AuditLogger.java @@ -144,6 +144,21 @@ public void logSkipKvBucket(long tableId, Long partitionId, int bucketId, String Instant.now()); } + /** + * Skip shared SST cleanup for a single bucket because the active set could not be determined + * (metadata read failure). The bucket's snap-private and log cleanup proceed normally. + */ + public void logSkipKvSharedSst(long tableId, Long partitionId, int bucketId, String reason) { + AUDIT.warn( + "action=skip_kv_shared_sst reason={} table_id={} partition_id={}" + + " bucket_id={} ts={}", + reason, + tableId, + partitionId, + bucketId, + Instant.now()); + } + /** * Skip log cleanup for one (tableId, partitionId) target — emitted when {@code * ListRemoteLogManifests} fails after retries. {@code partitionId} is null for non-partitioned diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/build/ActiveRefsFetcher.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/build/ActiveRefsFetcher.java index 223c6b97c4..88fc0db1f3 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/build/ActiveRefsFetcher.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/build/ActiveRefsFetcher.java @@ -32,6 +32,7 @@ import org.apache.fluss.utils.FlussPaths; import org.apache.fluss.utils.IOUtils; import org.apache.fluss.utils.RetryUtils; +import org.apache.fluss.utils.json.CompletedSnapshotJsonSerde; import javax.annotation.Nullable; @@ -254,6 +255,60 @@ public KvActiveRefsFetchResult fetchKvActiveSnapDirs(long tableId, @Nullable Lon return KvActiveRefsFetchResult.ok(dirsByBucket); } + /** + * Fetches the set of active shared SST file names for a single bucket by second-reading the + * {@code _METADATA} files of its active snapshots. + * + *

For each active snapshot directory, the method constructs the metadata path ({@code + * {kvTabletDir}/{snapDir}/_METADATA}), reads the file, and extracts shared SST basenames via + * {@link CompletedSnapshotJsonSerde}. The union across all active snapshots forms the complete + * active set. + * + *

Failure handling: + * + *

    + *
  • {@link java.io.FileNotFoundException} on a single metadata file (concurrent snapshot + * removal) — reported as failure so the caller skips shared SST cleanup for this bucket. + *
  • Other {@link IOException} — reported as {@link KvSharedSstFetchResult#failed(String)}; + * the caller must skip shared SST cleanup for this bucket. + *
+ */ + public KvSharedSstFetchResult fetchKvSharedSstFileNames( + FsPath kvTabletDir, Set activeSnapDirs) { + Set sharedSstFileNames = new HashSet<>(); + for (String snapDir : activeSnapDirs) { + FsPath metadataPath = new FsPath(new FsPath(kvTabletDir, snapDir), "_METADATA"); + byte[] metadataBytes; + try { + remoteFsOpRateLimiter.acquire(); + metadataBytes = metadataReader.read(metadataPath); + } catch (FileNotFoundException e) { + return KvSharedSstFetchResult.failed( + String.format( + "Snapshot metadata not found %s: %s", + metadataPath, + e.getMessage() != null ? e.getMessage() : e.getClass().getName())); + } catch (IOException e) { + return KvSharedSstFetchResult.failed( + String.format( + "IO error reading snapshot metadata %s: %s", + metadataPath, + e.getMessage() != null ? e.getMessage() : e.getClass().getName())); + } + try { + sharedSstFileNames.addAll( + CompletedSnapshotJsonSerde.parseSharedSstLocalPaths(metadataBytes)); + } catch (IOException e) { + return KvSharedSstFetchResult.failed( + String.format( + "Failed to parse snapshot metadata %s: %s", + metadataPath, + e.getMessage() != null ? e.getMessage() : e.getClass().getName())); + } + } + return KvSharedSstFetchResult.ok(sharedSstFileNames); + } + private static String formatRpcFailureReason( long tableId, @Nullable Long partitionId, @Nullable Throwable cause) { String reason = diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/build/KvSharedSstFetchResult.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/build/KvSharedSstFetchResult.java new file mode 100644 index 0000000000..55ce509e78 --- /dev/null +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/build/KvSharedSstFetchResult.java @@ -0,0 +1,88 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.flink.action.orphan.build; + +import org.apache.fluss.annotation.Internal; + +import javax.annotation.Nullable; + +import java.util.Collections; +import java.util.HashSet; +import java.util.Set; + +/** + * Result of fetching active shared SST file names for a single bucket by second-reading the active + * snapshots' {@code _METADATA} files. + * + *

Two outcomes: + * + *

    + *
  • OK: All (or a safe subset of) metadata reads succeeded; {@link + * #sharedSstFileNames()} contains the union of shared SST file basenames referenced by active + * snapshots. + *
  • Failed: A non-transient IO error prevented safe determination of the active set; the + * caller must skip shared SST cleanup for this bucket to uphold the "may leak, must not + * mis-delete" invariant. + *
+ */ +@Internal +public final class KvSharedSstFetchResult { + + private final boolean ok; + private final Set sharedSstFileNames; + @Nullable private final String failureReason; + + private KvSharedSstFetchResult( + boolean ok, Set sharedSstFileNames, @Nullable String failureReason) { + this.ok = ok; + this.sharedSstFileNames = Collections.unmodifiableSet(new HashSet<>(sharedSstFileNames)); + this.failureReason = failureReason; + } + + /** All active snapshot metadata reads succeeded; the active set is complete. */ + public static KvSharedSstFetchResult ok(Set sharedSstFileNames) { + return new KvSharedSstFetchResult(true, sharedSstFileNames, null); + } + + /** + * A non-transient IO error prevented safe determination of the active set. Shared SST cleanup + * must be skipped for this bucket. + */ + public static KvSharedSstFetchResult failed(String reason) { + return new KvSharedSstFetchResult(false, Collections.emptySet(), reason); + } + + /** Whether all metadata reads completed successfully. */ + public boolean allMetadataReadOk() { + return ok; + } + + /** + * Union of shared SST file basenames referenced by active snapshots. Empty when {@link + * #allMetadataReadOk()} is false. + */ + public Set sharedSstFileNames() { + return sharedSstFileNames; + } + + /** Failure reason for audit logging; {@code null} when {@link #allMetadataReadOk()} is true. */ + @Nullable + public String failureReason() { + return failureReason; + } +} diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/BucketCleanTask.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/BucketCleanTask.java index 70499fd285..02d67aee3f 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/BucketCleanTask.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/BucketCleanTask.java @@ -39,6 +39,7 @@ public final class BucketCleanTask implements CleanTask { private final Set logSegmentRelativePaths; private final Set logActiveManifestPaths; private final Set kvActiveSnapDirs; + private final Set kvSharedSstFileNames; private final long cutoffMillis; private final boolean dryRun; private final boolean allowDeleteManifest; @@ -49,6 +50,7 @@ public BucketCleanTask( Set logSegmentRelativePaths, Set logActiveManifestPaths, Set kvActiveSnapDirs, + Set kvSharedSstFileNames, long cutoffMillis, boolean dryRun, boolean allowDeleteManifest) { @@ -57,6 +59,7 @@ public BucketCleanTask( this.logSegmentRelativePaths = new HashSet<>(logSegmentRelativePaths); this.logActiveManifestPaths = new HashSet<>(logActiveManifestPaths); this.kvActiveSnapDirs = new HashSet<>(kvActiveSnapDirs); + this.kvSharedSstFileNames = new HashSet<>(kvSharedSstFileNames); this.cutoffMillis = cutoffMillis; this.dryRun = dryRun; this.allowDeleteManifest = allowDeleteManifest; @@ -89,6 +92,14 @@ public Set kvActiveSnapDirs() { return kvActiveSnapDirs; } + /** + * Active shared SST file names (basenames) resolved from active snapshots' {@code _METADATA} + * files. Empty when metadata could not be read (rules treat empty as "keep all"). + */ + public Set kvSharedSstFileNames() { + return kvSharedSstFileNames; + } + public long cutoffMillis() { return cutoffMillis; } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/BucketCleaner.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/BucketCleaner.java index a1e13cf424..e4b21c6aa8 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/BucketCleaner.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/BucketCleaner.java @@ -29,7 +29,6 @@ import org.apache.fluss.fs.FileSystem; import org.apache.fluss.fs.FsPath; import org.apache.fluss.shaded.guava32.com.google.common.util.concurrent.RateLimiter; -import org.apache.fluss.utils.FlussPaths; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -116,9 +115,6 @@ private void walkAndCleanDir(FsPath root, BucketActiveRefs activeRefs, BucketCle for (FileStatus child : children) { FsPath childPath = child.getPath(); if (child.isDir()) { - if (FlussPaths.REMOTE_KV_SNAPSHOT_SHARED_DIR.equals(childPath.getName())) { - continue; - } stack.push( new DirVisit( childPath, false, child.getModificationTime() < cutoffMillis)); diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/ScanAndCleanFunction.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/ScanAndCleanFunction.java index 2b54174643..d2c8ef80ac 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/ScanAndCleanFunction.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/ScanAndCleanFunction.java @@ -123,7 +123,8 @@ private CleanStats processBucketTask(BucketCleanTask task) throws IOException { new BucketActiveRefs( task.logSegmentRelativePaths(), task.kvActiveSnapDirs(), - task.logActiveManifestPaths()); + task.logActiveManifestPaths(), + task.kvSharedSstFileNames()); RuleDispatcher dispatcher = new RuleDispatcher(task.allowDeleteManifest()); SafeDeleter safeDeleter = createSafeDeleter(anyDir.getFileSystem(), task.dryRun()); BucketCleaner cleaner = diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/ScopeEnumeratorFunction.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/ScopeEnumeratorFunction.java index eede19cc68..ebe7f19ced 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/ScopeEnumeratorFunction.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/ScopeEnumeratorFunction.java @@ -31,6 +31,7 @@ import org.apache.fluss.flink.action.orphan.audit.AuditLogger; import org.apache.fluss.flink.action.orphan.build.ActiveRefsFetcher; import org.apache.fluss.flink.action.orphan.build.KvActiveRefsFetchResult; +import org.apache.fluss.flink.action.orphan.build.KvSharedSstFetchResult; import org.apache.fluss.flink.action.orphan.build.LogActiveRefsFetchResult; import org.apache.fluss.flink.action.orphan.build.MaxKnownIdsTracker; import org.apache.fluss.flink.action.orphan.config.OrphanCleanConfig; @@ -445,6 +446,7 @@ private void emitBucketTasksForTarget( String kvTabletDir = null; Set kvActiveSnaps = Collections.emptySet(); + Set kvSharedSstFileNames = Collections.emptySet(); if (kvTargetOk && kvActiveByBucket.containsKey(bucketId)) { kvTabletDir = FlussPaths.remoteKvTabletDir( @@ -453,6 +455,14 @@ private void emitBucketTasksForTarget( tableBucket) .toString(); kvActiveSnaps = kvActiveByBucket.get(bucketId); + KvSharedSstFetchResult sstResult = + fetcher.fetchKvSharedSstFileNames(new FsPath(kvTabletDir), kvActiveSnaps); + if (sstResult.allMetadataReadOk()) { + kvSharedSstFileNames = sstResult.sharedSstFileNames(); + } else { + audit.logSkipKvSharedSst( + liveTable.tableId, partitionId, bucketId, sstResult.failureReason()); + } } else if (kvTargetOk) { audit.logSkipKvBucket(liveTable.tableId, partitionId, bucketId, "empty_active_set"); } @@ -468,6 +478,7 @@ private void emitBucketTasksForTarget( logSegmentRelativePaths, logActiveManifestPaths, kvActiveSnaps, + kvSharedSstFileNames, config.olderThanMillis(), config.dryRun(), config.allowDeleteManifest())); diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/rule/BucketActiveRefs.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/rule/BucketActiveRefs.java index 73a847dd75..dbe5489289 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/rule/BucketActiveRefs.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/rule/BucketActiveRefs.java @@ -29,21 +29,39 @@ public final class BucketActiveRefs { private static final BucketActiveRefs EMPTY = new BucketActiveRefs( - Collections.emptySet(), Collections.emptySet(), Collections.emptySet()); + Collections.emptySet(), + Collections.emptySet(), + Collections.emptySet(), + Collections.emptySet()); private final Set logSegmentRelativePaths; private final Set kvActiveSnapDirs; private final Set logActiveManifestPaths; + private final Set kvSharedSstFileNames; public BucketActiveRefs( Set logSegmentRelativePaths, Set kvActiveSnapDirs, Set logActiveManifestPaths) { + this( + logSegmentRelativePaths, + kvActiveSnapDirs, + logActiveManifestPaths, + Collections.emptySet()); + } + + public BucketActiveRefs( + Set logSegmentRelativePaths, + Set kvActiveSnapDirs, + Set logActiveManifestPaths, + Set kvSharedSstFileNames) { this.logSegmentRelativePaths = Collections.unmodifiableSet(new HashSet<>(logSegmentRelativePaths)); this.kvActiveSnapDirs = Collections.unmodifiableSet(new HashSet<>(kvActiveSnapDirs)); this.logActiveManifestPaths = Collections.unmodifiableSet(new HashSet<>(logActiveManifestPaths)); + this.kvSharedSstFileNames = + Collections.unmodifiableSet(new HashSet<>(kvSharedSstFileNames)); } public static BucketActiveRefs empty() { @@ -81,4 +99,17 @@ public Set kvActiveSnapDirs() { public Set logActiveManifestPaths() { return logActiveManifestPaths; } + + /** + * Returns the set of active shared SST file names (basenames only, e.g. {@code + * "abc-def-0.sst"}) for the bucket. Built from the union of {@code + * shared_file_handles[*].local_path} across all active snapshots' {@code _METADATA} files. + * + *

An empty set means no shared SST active set could be determined (either all active + * snapshots have zero shared files — which is unusual for PK tables — or the metadata read + * failed). Rules must treat an empty set conservatively (keep all files). + */ + public Set kvSharedSstFileNames() { + return kvSharedSstFileNames; + } } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/rule/KvSharedSstRule.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/rule/KvSharedSstRule.java index 8fc1e5b2c0..c3903cd00f 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/rule/KvSharedSstRule.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/rule/KvSharedSstRule.java @@ -24,12 +24,13 @@ /** * Rule for shared SST files under the {@code shared/} KV directory. * - *

Always returns {@link Decision#KEEP_ACTIVE}. The true active set for shared SSTs lives inside - * the engine's {@code SharedKvFileRegistry}; orphan cleanup has no read path into that registry, so - * any deletion here would be a guess. Per the action's hard constraint "prefer leak over - * mis-delete," the rule never deletes, and as a consequence orphan PK-table / orphan-partition - * directories permanently retain their {@code shared/} subtree as accepted residue (recovering that - * residue would require a registry-backed GC channel that is out of scope for this action). + *

Determines whether a shared SST file is still referenced by any active snapshot. The active + * set is built from the union of {@code shared_file_handles[*].local_path} across all active + * snapshots' {@code _METADATA} files. + * + *

Safety: when the active set is empty (either because metadata reads failed or the bucket + * genuinely has no shared SSTs in any active snapshot), the rule conservatively returns {@link + * Decision#KEEP_ACTIVE} to prevent mis-deletion. */ @Internal public final class KvSharedSstRule implements FileRule { @@ -48,6 +49,18 @@ public Decision evaluate(FileMeta file, BucketActiveRefs activeRefs, long cutoff if (!file.path().getName().endsWith(".sst")) { return Decision.SKIP_UNKNOWN; } - return Decision.KEEP_ACTIVE; + + String fileName = file.path().getName(); + if (activeRefs.kvSharedSstFileNames().contains(fileName)) { + return Decision.KEEP_ACTIVE; + } + + // Empty active set means the metadata could not be read or genuinely no shared SSTs + // exist in any active snapshot — conservatively keep all files to prevent mis-deletion. + if (activeRefs.kvSharedSstFileNames().isEmpty()) { + return Decision.KEEP_ACTIVE; + } + + return file.modificationTime() < cutoffMillis ? Decision.DELETE : Decision.DEFER; } } diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/action/orphan/OrphanFilesCleanITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/action/orphan/OrphanFilesCleanITCase.java index 6128495b2f..0a28e92d5e 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/action/orphan/OrphanFilesCleanITCase.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/action/orphan/OrphanFilesCleanITCase.java @@ -354,6 +354,144 @@ void optInCleansOrphanTableDirWhenEnabled() throws Exception { && m.contains(layout.orphanFile.toString())); } + @Test + void sharedSstOrphanCleaned() throws Exception { + String dbName = newDatabaseName("sharedsst"); + TablePath tablePath = createPrimaryKeyTable(dbName, "sst_pk"); + TableInfo tableInfo = admin.getTableInfo(tablePath).get(); + TableBucket tableBucket = new TableBucket(tableInfo.getTableId(), 0); + FsPath remoteKvTabletDir = + FlussPaths.remoteKvTabletDir( + new FsPath(remoteDataRoot().resolve("kv").toUri().toString()), + PhysicalTablePath.of(tablePath), + tableBucket); + + // Seed two active snapshots with valid _METADATA referencing shared SSTs + seedKvSnapshotsWithSharedSst( + tableBucket, + remoteKvTabletDir, + new long[] {1L, 2L}, + new String[][] { + {"active-001.sst", "shared-both.sst"}, + {"active-002.sst", "shared-both.sst"} + }); + + // Plant shared SST files: some active, some orphan + Path sharedDir = localPath(new FsPath(remoteKvTabletDir, "shared")); + Files.createDirectories(sharedDir); + Path active1 = Files.write(sharedDir.resolve("active-001.sst"), new byte[] {0x42}); + Path active2 = Files.write(sharedDir.resolve("active-002.sst"), new byte[] {0x42}); + Path sharedBoth = Files.write(sharedDir.resolve("shared-both.sst"), new byte[] {0x42}); + Path orphan1 = Files.write(sharedDir.resolve("orphan-999.sst"), new byte[] {0x42}); + Path orphan2 = Files.write(sharedDir.resolve("orphan-888.sst"), new byte[] {0x42}); + makeOld(active1); + makeOld(active2); + makeOld(sharedBoth); + makeOld(orphan1); + makeOld(orphan2); + makeOld(sharedDir); + + // Also seed a log manifest so the bucket gets a valid BucketCleanTask. + seedActiveBucketManifest(tablePath); + + runCleanerForDatabase(false, dbName); + + // Active SST files must survive + assertThat(Files.exists(active1)).as("active-001.sst must survive").isTrue(); + assertThat(Files.exists(active2)).as("active-002.sst must survive").isTrue(); + assertThat(Files.exists(sharedBoth)).as("shared-both.sst must survive").isTrue(); + + // Orphan SST files must be deleted + assertThat(Files.exists(orphan1)).as("orphan-999.sst must be deleted").isFalse(); + assertThat(Files.exists(orphan2)).as("orphan-888.sst must be deleted").isFalse(); + + // Audit confirms deletions + assertThat(auditMessages()) + .anyMatch( + m -> + m.contains("action=deleted") + && m.contains("rule=kv-shared-sst") + && m.contains("orphan-999.sst")); + assertThat(auditMessages()) + .anyMatch( + m -> + m.contains("action=deleted") + && m.contains("rule=kv-shared-sst") + && m.contains("orphan-888.sst")); + } + + @Test + void sharedSstPreservedWhenYoung() throws Exception { + String dbName = newDatabaseName("sstfresh"); + TablePath tablePath = createPrimaryKeyTable(dbName, "sst_fresh_pk"); + TableInfo tableInfo = admin.getTableInfo(tablePath).get(); + TableBucket tableBucket = new TableBucket(tableInfo.getTableId(), 0); + FsPath remoteKvTabletDir = + FlussPaths.remoteKvTabletDir( + new FsPath(remoteDataRoot().resolve("kv").toUri().toString()), + PhysicalTablePath.of(tablePath), + tableBucket); + + // Seed an active snapshot with valid _METADATA referencing only one file + seedKvSnapshotsWithSharedSst( + tableBucket, remoteKvTabletDir, new long[] {1L}, new String[][] {{"active.sst"}}); + + // Plant shared SST: an unreferenced but YOUNG file + Path sharedDir = localPath(new FsPath(remoteKvTabletDir, "shared")); + Files.createDirectories(sharedDir); + Path activeSst = Files.write(sharedDir.resolve("active.sst"), new byte[] {0x42}); + Path youngOrphan = Files.write(sharedDir.resolve("young-orphan.sst"), new byte[] {0x42}); + makeOld(activeSst); + // youngOrphan keeps its default (fresh) mtime — within cutoff + makeOld(sharedDir); + + seedActiveBucketManifest(tablePath); + + runCleanerForDatabase(false, dbName); + + // Young orphan must be deferred (not deleted) + assertThat(Files.exists(youngOrphan)).as("young orphan must be deferred").isTrue(); + assertThat(Files.exists(activeSst)).as("active.sst must survive").isTrue(); + assertThat(auditMessages()) + .noneMatch(m -> m.contains("action=deleted") && m.contains("young-orphan.sst")); + } + + @Test + void sharedSstCleanupSkippedOnMetadataReadFailure() throws Exception { + String dbName = newDatabaseName("sstfail"); + TablePath tablePath = createPrimaryKeyTable(dbName, "sst_fail_pk"); + TableInfo tableInfo = admin.getTableInfo(tablePath).get(); + TableBucket tableBucket = new TableBucket(tableInfo.getTableId(), 0); + FsPath remoteKvTabletDir = + FlussPaths.remoteKvTabletDir( + new FsPath(remoteDataRoot().resolve("kv").toUri().toString()), + PhysicalTablePath.of(tablePath), + tableBucket); + + // Seed snapshot with INVALID _METADATA (causes parse failure) + seedKvSnapshots(tableBucket, remoteKvTabletDir, new long[] {1L}); + + // Plant shared SST files + Path sharedDir = localPath(new FsPath(remoteKvTabletDir, "shared")); + Files.createDirectories(sharedDir); + Path sst1 = Files.write(sharedDir.resolve("might-be-orphan.sst"), new byte[] {0x42}); + Path sst2 = Files.write(sharedDir.resolve("might-be-active.sst"), new byte[] {0x42}); + makeOld(sst1); + makeOld(sst2); + makeOld(sharedDir); + + seedActiveBucketManifest(tablePath); + + runCleanerForDatabase(false, dbName); + + // All shared SST files must be preserved (conservative: metadata read failure) + assertThat(Files.exists(sst1)).as("sst1 must be preserved on failure").isTrue(); + assertThat(Files.exists(sst2)).as("sst2 must be preserved on failure").isTrue(); + + // Skip audit must fire + assertThat(auditMessages()).anyMatch(m -> m.contains("action=skip_kv_shared_sst")); + } + @Test void pkOrphanTableRetainsSharedSstEvenWithOptIn() throws Exception { String dbName = newDatabaseName("orphankv"); @@ -964,6 +1102,49 @@ private void seedKvSnapshots( } } + /** + * Seeds KV snapshots with valid {@code _METADATA} JSON referencing the given shared SST file + * names per snapshot. This enables the shared SST cleanup to build a proper active set. + */ + private void seedKvSnapshotsWithSharedSst( + TableBucket tableBucket, + FsPath remoteKvTabletDir, + long[] snapshotIds, + String[][] sharedSstFilesPerSnapshot) + throws Exception { + ZooKeeperClient zk = FLUSS_CLUSTER_EXTENSION.getZooKeeperClient(); + for (int i = 0; i < snapshotIds.length; i++) { + long snapshotId = snapshotIds[i]; + FsPath snapshotDir = FlussPaths.remoteKvSnapshotDir(remoteKvTabletDir, snapshotId); + Path localSnapshotDir = localPath(snapshotDir); + Files.createDirectories(localSnapshotDir); + + // Build valid _METADATA JSON with shared_file_handles + StringBuilder json = new StringBuilder(); + json.append("{\"kv_snapshot_handle\":{\"shared_file_handles\":["); + String[] sharedFiles = sharedSstFilesPerSnapshot[i]; + for (int j = 0; j < sharedFiles.length; j++) { + if (j > 0) { + json.append(","); + } + json.append("{\"local_path\":\"").append(sharedFiles[j]).append("\","); + json.append("\"size\":1024}"); + } + json.append("]}}"); + + Path metadataFile = localSnapshotDir.resolve("_METADATA"); + Files.write(metadataFile, json.toString().getBytes(StandardCharsets.UTF_8)); + makeOld(metadataFile); + + makeOld(localSnapshotDir); + + zk.registerTableBucketSnapshot( + tableBucket, + new BucketSnapshot( + snapshotId, snapshotId, snapshotDir.toString() + "/_METADATA")); + } + } + private Path seedManifestAndSegment( FsPath remoteLogTabletDir, FsPath manifestPath, diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/action/orphan/build/ActiveRefsFetcherTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/action/orphan/build/ActiveRefsFetcherTest.java index 7144b4f031..b787ee447c 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/action/orphan/build/ActiveRefsFetcherTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/action/orphan/build/ActiveRefsFetcherTest.java @@ -292,6 +292,115 @@ void fetchKvActiveSnapDirsWithPartitionIdRoutesCorrectly() { .isEqualTo(1); } + // ------------------------------------------------------------------------- + // fetchKvSharedSstFileNames tests + // ------------------------------------------------------------------------- + + @Test + void fetchKvSharedSstFileNamesReadsMetadataAndExtractsFileNames() { + FsPath kvTabletDir = new FsPath("oss://b/kv/db/t-7/0"); + Set activeSnapDirs = new HashSet<>(Arrays.asList("snap-5", "snap-10")); + + String metadataSnap5 = + "{\"kv_snapshot_handle\":{\"shared_file_handles\":[" + + "{\"local_path\":\"aaa.sst\",\"size\":100}," + + "{\"local_path\":\"bbb.sst\",\"size\":200}" + + "]}}"; + String metadataSnap10 = + "{\"kv_snapshot_handle\":{\"shared_file_handles\":[" + + "{\"local_path\":\"bbb.sst\",\"size\":200}," + + "{\"local_path\":\"ccc.sst\",\"size\":300}" + + "]}}"; + + StubManifestReader reader = new StubManifestReader(); + reader.returnBytes( + new FsPath("oss://b/kv/db/t-7/0/snap-5/_METADATA"), + metadataSnap5.getBytes(StandardCharsets.UTF_8)); + reader.returnBytes( + new FsPath("oss://b/kv/db/t-7/0/snap-10/_METADATA"), + metadataSnap10.getBytes(StandardCharsets.UTF_8)); + + StubAdmin admin = new StubAdmin(new AtomicInteger()); + ActiveRefsFetcher fetcher = new ActiveRefsFetcher(admin, reader, /* maxRetries= */ 3); + + KvSharedSstFetchResult result = + fetcher.fetchKvSharedSstFileNames(kvTabletDir, activeSnapDirs); + + assertThat(result.allMetadataReadOk()).isTrue(); + assertThat(result.sharedSstFileNames()) + .containsExactlyInAnyOrder("aaa.sst", "bbb.sst", "ccc.sst"); + } + + @Test + void fetchKvSharedSstFileNamesReportsFailureOnFileNotFound() { + FsPath kvTabletDir = new FsPath("oss://b/kv/db/t-7/0"); + Set activeSnapDirs = new HashSet<>(Arrays.asList("snap-5", "snap-10")); + + String metadataSnap5 = + "{\"kv_snapshot_handle\":{\"shared_file_handles\":[" + + "{\"local_path\":\"aaa.sst\",\"size\":100}" + + "]}}"; + + StubManifestReader reader = new StubManifestReader(); + reader.returnBytes( + new FsPath("oss://b/kv/db/t-7/0/snap-5/_METADATA"), + metadataSnap5.getBytes(StandardCharsets.UTF_8)); + // snap-10 metadata file not found — concurrent removal + reader.failWithNotFound(new FsPath("oss://b/kv/db/t-7/0/snap-10/_METADATA")); + + StubAdmin admin = new StubAdmin(new AtomicInteger()); + ActiveRefsFetcher fetcher = new ActiveRefsFetcher(admin, reader, /* maxRetries= */ 3); + + KvSharedSstFetchResult result = + fetcher.fetchKvSharedSstFileNames(kvTabletDir, activeSnapDirs); + + assertThat(result.allMetadataReadOk()).isFalse(); + assertThat(result.failureReason()).contains("Snapshot metadata not found"); + assertThat(result.sharedSstFileNames()).isEmpty(); + } + + @Test + void fetchKvSharedSstFileNamesReportsFailureOnIoError() { + FsPath kvTabletDir = new FsPath("oss://b/kv/db/t-7/0"); + Set activeSnapDirs = new HashSet<>(Arrays.asList("snap-5")); + + StubManifestReader reader = new StubManifestReader(); + reader.failWithIo( + new FsPath("oss://b/kv/db/t-7/0/snap-5/_METADATA"), + new IOException("connection reset")); + + StubAdmin admin = new StubAdmin(new AtomicInteger()); + ActiveRefsFetcher fetcher = new ActiveRefsFetcher(admin, reader, /* maxRetries= */ 3); + + KvSharedSstFetchResult result = + fetcher.fetchKvSharedSstFileNames(kvTabletDir, activeSnapDirs); + + assertThat(result.allMetadataReadOk()).isFalse(); + assertThat(result.failureReason()).contains("connection reset"); + assertThat(result.sharedSstFileNames()).isEmpty(); + } + + @Test + void fetchKvSharedSstFileNamesReportsFailureOnParseError() { + FsPath kvTabletDir = new FsPath("oss://b/kv/db/t-7/0"); + Set activeSnapDirs = new HashSet<>(Arrays.asList("snap-5")); + + StubManifestReader reader = new StubManifestReader(); + // Invalid JSON in _METADATA + reader.returnBytes( + new FsPath("oss://b/kv/db/t-7/0/snap-5/_METADATA"), + "not valid json".getBytes(StandardCharsets.UTF_8)); + + StubAdmin admin = new StubAdmin(new AtomicInteger()); + ActiveRefsFetcher fetcher = new ActiveRefsFetcher(admin, reader, /* maxRetries= */ 3); + + KvSharedSstFetchResult result = + fetcher.fetchKvSharedSstFileNames(kvTabletDir, activeSnapDirs); + + assertThat(result.allMetadataReadOk()).isFalse(); + assertThat(result.failureReason()).contains("Failed to parse snapshot metadata"); + } + // ------------------------------------------------------------------------- // Test fixtures // ------------------------------------------------------------------------- diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/action/orphan/job/BucketCleanerTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/action/orphan/job/BucketCleanerTest.java index b0fc5484f5..41f84b6088 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/action/orphan/job/BucketCleanerTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/action/orphan/job/BucketCleanerTest.java @@ -32,7 +32,10 @@ import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.attribute.FileTime; +import java.util.Arrays; import java.util.Collections; +import java.util.HashSet; +import java.util.Set; import static org.assertj.core.api.Assertions.assertThat; @@ -112,6 +115,59 @@ void scansButDoesNotDeleteUnknownDotFiles(@TempDir Path tmp) throws IOException assertThat(Files.exists(segmentDir)).isTrue(); } + @Test + void scansSharedDirectoryAndDeletesOrphanSst(@TempDir Path tmp) throws IOException { + Path bucketRoot = Files.createDirectories(tmp.resolve("bucket")); + Path sharedDir = Files.createDirectories(bucketRoot.resolve("shared")); + Path activeSst = Files.write(sharedDir.resolve("active.sst"), new byte[] {0x42}); + Path orphanSst = Files.write(sharedDir.resolve("orphan.sst"), new byte[] {0x42}); + long cutoff = System.currentTimeMillis() - 1000L; + makeOld(activeSst, cutoff - 1000L); + makeOld(orphanSst, cutoff - 1000L); + makeOld(sharedDir, cutoff - 1000L); + makeOld(bucketRoot, cutoff - 1000L); + + Set activeSharedSstFiles = new HashSet<>(Arrays.asList("active.sst")); + BucketActiveRefs activeRefs = + new BucketActiveRefs( + Collections.emptySet(), + Collections.emptySet(), + Collections.emptySet(), + activeSharedSstFiles); + + BucketCleaner cleaner = createCleaner(bucketRoot, cutoff); + BucketCleaner.BucketCleanStats stats = + cleaner.clean(activeRefs, new FsPath(bucketRoot.toString())); + + assertThat(stats.scanned).isEqualTo(2L); + // orphan.sst deleted; active.sst kept + assertThat(stats.deleted).isGreaterThanOrEqualTo(1L); + assertThat(Files.exists(activeSst)).isTrue(); + assertThat(Files.exists(orphanSst)).isFalse(); + } + + @Test + void keepsAllSharedSstWhenActiveSetIsEmpty(@TempDir Path tmp) throws IOException { + Path bucketRoot = Files.createDirectories(tmp.resolve("bucket")); + Path sharedDir = Files.createDirectories(bucketRoot.resolve("shared")); + Path sst1 = Files.write(sharedDir.resolve("file1.sst"), new byte[] {0x42}); + Path sst2 = Files.write(sharedDir.resolve("file2.sst"), new byte[] {0x42}); + long cutoff = System.currentTimeMillis() - 1000L; + makeOld(sst1, cutoff - 1000L); + makeOld(sst2, cutoff - 1000L); + makeOld(sharedDir, cutoff - 1000L); + makeOld(bucketRoot, cutoff - 1000L); + + // Empty active set = conservative keep-all behavior + BucketCleaner cleaner = createCleaner(bucketRoot, cutoff); + BucketCleaner.BucketCleanStats stats = + cleaner.clean(BucketActiveRefs.empty(), new FsPath(bucketRoot.toString())); + + // Both files kept (KEEP_ACTIVE due to empty active set) + assertThat(Files.exists(sst1)).isTrue(); + assertThat(Files.exists(sst2)).isTrue(); + } + private static void makeOld(Path path, long timestampMillis) throws IOException { Files.setLastModifiedTime(path, FileTime.fromMillis(timestampMillis)); } diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/action/orphan/rule/KvSharedSstRuleTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/action/orphan/rule/KvSharedSstRuleTest.java index c6267d31c8..ff81a2b0b3 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/action/orphan/rule/KvSharedSstRuleTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/action/orphan/rule/KvSharedSstRuleTest.java @@ -21,6 +21,7 @@ import org.junit.jupiter.api.Test; +import java.util.Arrays; import java.util.Collections; import java.util.HashSet; import java.util.Set; @@ -37,7 +38,25 @@ class KvSharedSstRuleTest { private final KvSharedSstRule rule = new KvSharedSstRule(); @Test - void keepsExpiredUnreferencedSharedSst() { + void deletesExpiredUnreferencedSharedSst() { + FileMeta file = file("/kv/db/t-1/0/shared/abc-001.sst", NOW - 2 * DAY_MS); + + assertThat(rule.evaluate(file, sharedSstActiveRefs("other-file.sst"), CUTOFF_MS)) + .isEqualTo(Decision.DELETE); + } + + @Test + void defersYoungUnreferencedSharedSst() { + FileMeta file = file("/kv/db/t-1/0/shared/abc-001.sst", NOW - DAY_MS / 2); + + assertThat(rule.evaluate(file, sharedSstActiveRefs("other-file.sst"), CUTOFF_MS)) + .isEqualTo(Decision.DEFER); + } + + @Test + void keepsExpiredSharedSstWhenActiveSetIsEmpty() { + // Empty active set = metadata read failure or no shared SSTs in any snapshot. + // Must conservatively keep all files. FileMeta file = file("/kv/db/t-1/0/shared/abc-001.sst", NOW - 2 * DAY_MS); assertThat(rule.evaluate(file, BucketActiveRefs.empty(), CUTOFF_MS)) @@ -47,15 +66,9 @@ void keepsExpiredUnreferencedSharedSst() { @Test void keepsReferencedSharedSst() { FileMeta file = file("/kv/db/t-1/0/shared/abc-001.sst", NOW - 2 * DAY_MS); - Set sharedFiles = new HashSet(); - sharedFiles.add("abc-001.sst"); - BucketActiveRefs activeRefs = - new BucketActiveRefs( - Collections.emptySet(), - Collections.emptySet(), - sharedFiles); - - assertThat(rule.evaluate(file, activeRefs, CUTOFF_MS)).isEqualTo(Decision.KEEP_ACTIVE); + + assertThat(rule.evaluate(file, sharedSstActiveRefs("abc-001.sst"), CUTOFF_MS)) + .isEqualTo(Decision.KEEP_ACTIVE); } @Test @@ -74,6 +87,15 @@ void skipsSstOutsideSharedDirectory() { .isEqualTo(Decision.SKIP_UNKNOWN); } + private static BucketActiveRefs sharedSstActiveRefs(String... fileNames) { + Set sharedFiles = new HashSet<>(Arrays.asList(fileNames)); + return new BucketActiveRefs( + Collections.emptySet(), + Collections.emptySet(), + Collections.emptySet(), + sharedFiles); + } + private static FileMeta file(String path, long modificationTime) { return new FileMeta(new FsPath(path), 1L, modificationTime); } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java b/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java index f7b6de9eb8..7ddff9dc36 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java @@ -30,6 +30,7 @@ import org.apache.fluss.exception.TableNotPartitionedException; import org.apache.fluss.fs.FileSystem; import org.apache.fluss.fs.token.ObtainedSecurityToken; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; import org.apache.fluss.metadata.DatabaseInfo; import org.apache.fluss.metadata.PhysicalTablePath; import org.apache.fluss.metadata.ResolvedPartitionSpec; @@ -83,7 +84,6 @@ import org.apache.fluss.server.authorizer.Authorizer; import org.apache.fluss.server.coordinator.CoordinatorService; import org.apache.fluss.server.coordinator.MetadataManager; -import org.apache.fluss.server.kv.snapshot.CompletedSnapshot; import org.apache.fluss.server.metadata.MetadataProvider; import org.apache.fluss.server.metadata.PartitionMetadata; import org.apache.fluss.server.metadata.ServerMetadataCache; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManager.java index f7c1b9bf6b..b64922232b 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManager.java @@ -19,11 +19,11 @@ import org.apache.fluss.annotation.VisibleForTesting; import org.apache.fluss.exception.ApiException; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TablePath; import org.apache.fluss.metrics.MetricNames; import org.apache.fluss.metrics.groups.MetricGroup; -import org.apache.fluss.server.kv.snapshot.CompletedSnapshot; import org.apache.fluss.server.kv.snapshot.CompletedSnapshotHandle; import org.apache.fluss.server.kv.snapshot.CompletedSnapshotHandleStore; import org.apache.fluss.server.kv.snapshot.CompletedSnapshotStore; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java index 537e2df63f..f94d42389b 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java @@ -40,6 +40,7 @@ import org.apache.fluss.exception.TabletServerNotAvailableException; import org.apache.fluss.exception.UnknownServerException; import org.apache.fluss.exception.UnknownTableOrBucketException; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; import org.apache.fluss.metadata.PhysicalTablePath; import org.apache.fluss.metadata.SchemaInfo; import org.apache.fluss.metadata.TableBucket; @@ -104,7 +105,6 @@ import org.apache.fluss.server.entity.CommitRemoteLogManifestData; import org.apache.fluss.server.entity.DeleteReplicaResultForBucket; import org.apache.fluss.server.entity.NotifyLeaderAndIsrResultForBucket; -import org.apache.fluss.server.kv.snapshot.CompletedSnapshot; import org.apache.fluss.server.kv.snapshot.CompletedSnapshotStore; import org.apache.fluss.server.metadata.CoordinatorMetadataCache; import org.apache.fluss.server.metadata.ServerInfo; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java index 0772f5e3db..4dccf1f459 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java @@ -43,6 +43,7 @@ import org.apache.fluss.exception.UnknownTableOrBucketException; import org.apache.fluss.fs.FileSystem; import org.apache.fluss.fs.FsPath; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; import org.apache.fluss.lake.committer.TieringStats; import org.apache.fluss.lake.lakestorage.LakeCatalog; import org.apache.fluss.metadata.DataLakeFormat; @@ -167,8 +168,6 @@ import org.apache.fluss.server.entity.DatabasePropertyChanges; import org.apache.fluss.server.entity.LakeTieringTableInfo; import org.apache.fluss.server.entity.TablePropertyChanges; -import org.apache.fluss.server.kv.snapshot.CompletedSnapshot; -import org.apache.fluss.server.kv.snapshot.CompletedSnapshotJsonSerde; import org.apache.fluss.server.metadata.CoordinatorMetadataCache; import org.apache.fluss.server.metadata.CoordinatorMetadataProvider; import org.apache.fluss.server.utils.ServerRpcMessageUtils; @@ -184,6 +183,7 @@ import org.apache.fluss.server.zk.data.producer.ProducerOffsets; import org.apache.fluss.utils.IOUtils; import org.apache.fluss.utils.concurrent.FutureUtils; +import org.apache.fluss.utils.json.CompletedSnapshotJsonSerde; import org.apache.fluss.utils.json.TableBucketOffsets; import org.slf4j.Logger; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/lease/KvSnapshotLeaseManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/lease/KvSnapshotLeaseManager.java index 6c79a668e8..0458558f27 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/lease/KvSnapshotLeaseManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/lease/KvSnapshotLeaseManager.java @@ -18,11 +18,11 @@ package org.apache.fluss.server.coordinator.lease; import org.apache.fluss.annotation.VisibleForTesting; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TableBucketSnapshot; import org.apache.fluss.metadata.TablePartition; import org.apache.fluss.metrics.MetricNames; -import org.apache.fluss.server.kv.snapshot.CompletedSnapshot; import org.apache.fluss.server.metrics.group.CoordinatorMetricGroup; import org.apache.fluss.server.zk.ZooKeeperClient; import org.apache.fluss.server.zk.data.lease.KvSnapshotTableLease; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/entity/CommitKvSnapshotData.java b/fluss-server/src/main/java/org/apache/fluss/server/entity/CommitKvSnapshotData.java index cdf1615d77..a69a27e0ed 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/entity/CommitKvSnapshotData.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/entity/CommitKvSnapshotData.java @@ -17,8 +17,8 @@ package org.apache.fluss.server.entity; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; import org.apache.fluss.rpc.messages.CommitKvSnapshotRequest; -import org.apache.fluss.server.kv.snapshot.CompletedSnapshot; /** The data for request {@link CommitKvSnapshotRequest}. */ public class CommitKvSnapshotData { diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvRecoverHelper.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvRecoverHelper.java index f78a258ecb..2c2e3c70e6 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvRecoverHelper.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvRecoverHelper.java @@ -17,6 +17,7 @@ package org.apache.fluss.server.kv; +import org.apache.fluss.kv.autoinc.AutoIncIDRange; import org.apache.fluss.metadata.KvFormat; import org.apache.fluss.metadata.LogFormat; import org.apache.fluss.metadata.Schema; @@ -35,7 +36,6 @@ import org.apache.fluss.row.encode.RowEncoder; import org.apache.fluss.row.encode.ValueEncoder; import org.apache.fluss.row.indexed.IndexedRow; -import org.apache.fluss.server.kv.autoinc.AutoIncIDRange; import org.apache.fluss.server.log.FetchIsolation; import org.apache.fluss.server.log.LogTablet; import org.apache.fluss.server.zk.ZooKeeperClient; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java index 0683f5e054..d8c57c1aa2 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java @@ -25,6 +25,8 @@ import org.apache.fluss.exception.InvalidTableException; import org.apache.fluss.exception.KvStorageException; import org.apache.fluss.exception.SchemaNotExistException; +import org.apache.fluss.kv.autoinc.AutoIncIDRange; +import org.apache.fluss.kv.snapshot.KvFileHandleAndLocalPath; import org.apache.fluss.memory.MemorySegmentPool; import org.apache.fluss.metadata.ChangelogImage; import org.apache.fluss.metadata.DeleteBehavior; @@ -47,7 +49,6 @@ import org.apache.fluss.row.arrow.ArrowWriterProvider; import org.apache.fluss.row.encode.ValueDecoder; import org.apache.fluss.rpc.protocol.MergeMode; -import org.apache.fluss.server.kv.autoinc.AutoIncIDRange; import org.apache.fluss.server.kv.autoinc.AutoIncrementManager; import org.apache.fluss.server.kv.autoinc.AutoIncrementUpdater; import org.apache.fluss.server.kv.prewrite.KvPreWriteBuffer; @@ -60,7 +61,6 @@ import org.apache.fluss.server.kv.rowmerger.RowMerger; import org.apache.fluss.server.kv.scan.OpenScanResult; import org.apache.fluss.server.kv.scan.ScannerContext; -import org.apache.fluss.server.kv.snapshot.KvFileHandleAndLocalPath; import org.apache.fluss.server.kv.snapshot.KvSnapshotDataUploader; import org.apache.fluss.server.kv.snapshot.RocksIncrementalSnapshot; import org.apache.fluss.server.kv.snapshot.TabletState; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/AutoIncrementManager.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/AutoIncrementManager.java index 8a69e1f1b9..8b1ce8a6cc 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/AutoIncrementManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/AutoIncrementManager.java @@ -19,6 +19,7 @@ package org.apache.fluss.server.kv.autoinc; import org.apache.fluss.config.TableConfig; +import org.apache.fluss.kv.autoinc.AutoIncIDRange; import org.apache.fluss.metadata.KvFormat; import org.apache.fluss.metadata.Schema; import org.apache.fluss.metadata.SchemaGetter; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/BoundedSegmentSequenceGenerator.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/BoundedSegmentSequenceGenerator.java index 1198f6b85c..447487dacb 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/BoundedSegmentSequenceGenerator.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/BoundedSegmentSequenceGenerator.java @@ -20,6 +20,7 @@ import org.apache.fluss.exception.FlussRuntimeException; import org.apache.fluss.exception.SequenceOverflowException; +import org.apache.fluss.kv.autoinc.AutoIncIDRange; import org.apache.fluss.metadata.TablePath; import org.apache.fluss.server.SequenceIDCounter; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/SequenceGenerator.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/SequenceGenerator.java index 648479376e..36514467f5 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/SequenceGenerator.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/SequenceGenerator.java @@ -18,6 +18,8 @@ package org.apache.fluss.server.kv.autoinc; +import org.apache.fluss.kv.autoinc.AutoIncIDRange; + /** SequenceGenerator is used to generate auto increment column ID. */ public interface SequenceGenerator { diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedKvSnapshotCommitter.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedKvSnapshotCommitter.java index 417459ac07..85a56190c2 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedKvSnapshotCommitter.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedKvSnapshotCommitter.java @@ -17,6 +17,8 @@ package org.apache.fluss.server.kv.snapshot; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; + /** An interface for reporting a {@link CompletedSnapshot}. */ public interface CompletedKvSnapshotCommitter { diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotHandle.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotHandle.java index e0bcdc3325..f1c4e2f847 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotHandle.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotHandle.java @@ -20,7 +20,9 @@ import org.apache.fluss.fs.FSDataInputStream; import org.apache.fluss.fs.FileSystem; import org.apache.fluss.fs.FsPath; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; import org.apache.fluss.utils.IOUtils; +import org.apache.fluss.utils.json.CompletedSnapshotJsonSerde; import java.io.ByteArrayOutputStream; import java.io.IOException; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotStore.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotStore.java index 7cf3e88db6..7d7bfa3eab 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotStore.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotStore.java @@ -21,7 +21,9 @@ import org.apache.fluss.fs.FSDataOutputStream; import org.apache.fluss.fs.FileSystem; import org.apache.fluss.fs.FsPath; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.utils.json.CompletedSnapshotJsonSerde; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -160,7 +162,7 @@ void addSnapshotAndSubsumeOldestOne( checkNotNull(snapshot, "Snapshot"); // register the completed snapshot to the shared registry - snapshot.registerSharedKvFilesAfterRestored(sharedKvFileRegistry); + sharedKvFileRegistry.registerAllAfterRestored(snapshot); CompletedSnapshotHandle completedSnapshotHandle = store(snapshot); completedSnapshotHandleStore.add( @@ -208,7 +210,7 @@ void addSnapshotAndSubsumeOldestOne( // Snapshot metadata/private files cleanup: use the latest snapshot // ID + 1 so subsumed snapshots can be cleaned even when a lower // snapshot has a lease. This is safe because - // KvSnapshotHandle.discard() only deletes private files and + // KvSnapshotHandle.discardPrivateFiles() only deletes private files and // metadata, not shared SST files registered in SharedKvFileRegistry. snapshotsCleaner.cleanSubsumedSnapshots( snapshot.getSnapshotID() + 1, stillInUseIds, postCleanup, ioExecutor); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/DefaultCompletedKvSnapshotCommitter.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/DefaultCompletedKvSnapshotCommitter.java index 7dece9daf2..f3d41ad4d1 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/DefaultCompletedKvSnapshotCommitter.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/DefaultCompletedKvSnapshotCommitter.java @@ -19,6 +19,7 @@ import org.apache.fluss.cluster.ServerNode; import org.apache.fluss.exception.FlussRuntimeException; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; import org.apache.fluss.rpc.GatewayClientProxy; import org.apache.fluss.rpc.RpcClient; import org.apache.fluss.rpc.gateway.CoordinatorGateway; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/DefaultSnapshotContext.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/DefaultSnapshotContext.java index 492d8aca41..a6086eb621 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/DefaultSnapshotContext.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/DefaultSnapshotContext.java @@ -22,6 +22,7 @@ import org.apache.fluss.config.cluster.ServerReconfigurable; import org.apache.fluss.exception.ConfigException; import org.apache.fluss.fs.FsPath; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.server.kv.KvSnapshotResource; import org.apache.fluss.server.zk.ZooKeeperClient; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataDownloader.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataDownloader.java index 48b235300a..141104ca6c 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataDownloader.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataDownloader.java @@ -21,6 +21,7 @@ import org.apache.fluss.fs.FsPathAndFileName; import org.apache.fluss.fs.utils.FileDownloadSpec; import org.apache.fluss.fs.utils.FileDownloadUtils; +import org.apache.fluss.kv.snapshot.KvSnapshotHandle; import org.apache.fluss.utils.CloseableRegistry; import java.util.ArrayList; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataUploader.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataUploader.java index ef4d3b6eee..953fc90a6e 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataUploader.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataUploader.java @@ -18,7 +18,9 @@ package org.apache.fluss.server.kv.snapshot; import org.apache.fluss.exception.FlussRuntimeException; -import org.apache.fluss.server.utils.SnapshotUtil; +import org.apache.fluss.kv.snapshot.KvFileHandle; +import org.apache.fluss.kv.snapshot.KvFileHandleAndLocalPath; +import org.apache.fluss.kv.snapshot.SnapshotUtil; import org.apache.fluss.utils.CloseableRegistry; import org.apache.fluss.utils.ExceptionUtils; import org.apache.fluss.utils.IOUtils; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDownloadSpec.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDownloadSpec.java index 83aae52d1f..1dda09c4ca 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDownloadSpec.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDownloadSpec.java @@ -17,6 +17,8 @@ package org.apache.fluss.server.kv.snapshot; +import org.apache.fluss.kv.snapshot.KvSnapshotHandle; + import java.nio.file.Path; /** diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvTabletSnapshotTarget.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvTabletSnapshotTarget.java index fe4591dc78..82ca0ea754 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvTabletSnapshotTarget.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvTabletSnapshotTarget.java @@ -22,6 +22,7 @@ import org.apache.fluss.exception.FlussRuntimeException; import org.apache.fluss.fs.FileSystem; import org.apache.fluss.fs.FsPath; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.server.SequenceIDCounter; import org.apache.fluss.server.zk.ZooKeeperClient; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/PeriodicSnapshotManager.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/PeriodicSnapshotManager.java index c69e946344..7c5d8fb13a 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/PeriodicSnapshotManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/PeriodicSnapshotManager.java @@ -323,7 +323,7 @@ private void discardFailedUploads(RunnableFuture snapshotedRunna try { SnapshotResult snapshotResult = snapshotedRunnableFuture.get(); if (snapshotResult != null) { - snapshotResult.getKvSnapshotHandle().discard(); + snapshotResult.getKvSnapshotHandle().discardAll(); FsPath remoteSnapshotPath = snapshotResult.getSnapshotPath(); remoteSnapshotPath.getFileSystem().delete(remoteSnapshotPath, true); } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/PlaceholderKvFileHandler.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/PlaceholderKvFileHandler.java index 0ad884ea66..a2c7a49d89 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/PlaceholderKvFileHandler.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/PlaceholderKvFileHandler.java @@ -17,6 +17,8 @@ package org.apache.fluss.server.kv.snapshot; +import org.apache.fluss.kv.snapshot.KvFileHandle; + /** * A placeholder handle for shared kv files that will be replaced by an original that was created in * a previous snapshot. This class is used in the referenced kv files of {@link KvSnapshotHandle}. diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/RocksIncrementalSnapshot.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/RocksIncrementalSnapshot.java index 3ae7318f6c..1bddca9736 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/RocksIncrementalSnapshot.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/RocksIncrementalSnapshot.java @@ -17,6 +17,9 @@ package org.apache.fluss.server.kv.snapshot; +import org.apache.fluss.kv.snapshot.KvFileHandle; +import org.apache.fluss.kv.snapshot.KvFileHandleAndLocalPath; +import org.apache.fluss.kv.snapshot.KvSnapshotHandle; import org.apache.fluss.server.utils.ResourceGuard; import org.apache.fluss.utils.CloseableRegistry; import org.apache.fluss.utils.ExceptionUtils; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SharedKvFileRegistry.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SharedKvFileRegistry.java index 558e847d96..a99e5217e7 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SharedKvFileRegistry.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SharedKvFileRegistry.java @@ -17,6 +17,10 @@ package org.apache.fluss.server.kv.snapshot; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; +import org.apache.fluss.kv.snapshot.KvFileHandle; +import org.apache.fluss.kv.snapshot.KvFileHandleAndLocalPath; +import org.apache.fluss.kv.snapshot.KvSnapshotHandle; import org.apache.fluss.utils.concurrent.Executors; import org.slf4j.Logger; @@ -200,7 +204,12 @@ public void registerAll(KvSnapshotHandle kvSnapshotHandle, long snapshotID) { } synchronized (registeredKvEntries) { - kvSnapshotHandle.registerKvFileHandles(this, snapshotID); + for (KvFileHandleAndLocalPath hlp : kvSnapshotHandle.getSharedKvFileHandles()) { + registerReference( + SharedKvFileRegistryKey.fromKvFileHandle(hlp.getKvFileHandle()), + hlp.getKvFileHandle(), + snapshotID); + } } } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SharedKvFileRegistryKey.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SharedKvFileRegistryKey.java index 6b01ae19a4..e426560daf 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SharedKvFileRegistryKey.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SharedKvFileRegistryKey.java @@ -17,6 +17,8 @@ package org.apache.fluss.server.kv.snapshot; +import org.apache.fluss.kv.snapshot.KvFileHandle; + import java.io.Serializable; import java.nio.charset.StandardCharsets; import java.util.UUID; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotContext.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotContext.java index 18c32d5312..fac16233ea 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotContext.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotContext.java @@ -19,6 +19,7 @@ import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.fs.FsPath; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.server.zk.ZooKeeperClient; import org.apache.fluss.utils.function.FunctionWithException; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotLocation.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotLocation.java index fe517f2682..ba1d02f456 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotLocation.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotLocation.java @@ -20,6 +20,7 @@ import org.apache.fluss.fs.FSDataOutputStream; import org.apache.fluss.fs.FileSystem; import org.apache.fluss.fs.FsPath; +import org.apache.fluss.kv.snapshot.KvFileHandle; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotResult.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotResult.java index b8bdafddbb..7946df7eb4 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotResult.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotResult.java @@ -18,6 +18,7 @@ package org.apache.fluss.server.kv.snapshot; import org.apache.fluss.fs.FsPath; +import org.apache.fluss.kv.snapshot.KvSnapshotHandle; import java.io.Serializable; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotsCleaner.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotsCleaner.java index dafabb1ded..1bfde4aa06 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotsCleaner.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotsCleaner.java @@ -17,6 +17,8 @@ package org.apache.fluss.server.kv.snapshot; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/TabletState.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/TabletState.java index a5f19cee74..7d772d580c 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/TabletState.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/TabletState.java @@ -19,7 +19,7 @@ package org.apache.fluss.server.kv.snapshot; -import org.apache.fluss.server.kv.autoinc.AutoIncIDRange; +import org.apache.fluss.kv.autoinc.AutoIncIDRange; import javax.annotation.Nullable; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java index e11dd69c14..32be13e485 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java @@ -33,6 +33,9 @@ import org.apache.fluss.exception.NotLeaderOrFollowerException; import org.apache.fluss.exception.TooManyScannersException; import org.apache.fluss.fs.FsPath; +import org.apache.fluss.kv.autoinc.AutoIncIDRange; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; +import org.apache.fluss.kv.snapshot.KvFileHandleAndLocalPath; import org.apache.fluss.metadata.ChangelogImage; import org.apache.fluss.metadata.LogFormat; import org.apache.fluss.metadata.PhysicalTablePath; @@ -60,14 +63,11 @@ import org.apache.fluss.server.kv.KvRecoverHelper; import org.apache.fluss.server.kv.KvTablet; import org.apache.fluss.server.kv.RemoteLogFetcher; -import org.apache.fluss.server.kv.autoinc.AutoIncIDRange; import org.apache.fluss.server.kv.rocksdb.RocksDBKvBuilder; import org.apache.fluss.server.kv.scan.OpenScanResult; import org.apache.fluss.server.kv.scan.ScannerContext; import org.apache.fluss.server.kv.scan.ScannerManager; import org.apache.fluss.server.kv.snapshot.CompletedKvSnapshotCommitter; -import org.apache.fluss.server.kv.snapshot.CompletedSnapshot; -import org.apache.fluss.server.kv.snapshot.KvFileHandleAndLocalPath; import org.apache.fluss.server.kv.snapshot.KvSnapshotDataDownloader; import org.apache.fluss.server.kv.snapshot.KvSnapshotDownloadSpec; import org.apache.fluss.server.kv.snapshot.KvTabletSnapshotTarget; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java index 266abe3bce..7cca6a25cf 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java @@ -29,6 +29,8 @@ import org.apache.fluss.config.cluster.ConfigEntry; import org.apache.fluss.fs.FsPath; import org.apache.fluss.fs.token.ObtainedSecurityToken; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; +import org.apache.fluss.kv.snapshot.KvSnapshotHandle; import org.apache.fluss.lake.committer.LakeCommitResult; import org.apache.fluss.metadata.DatabaseChange; import org.apache.fluss.metadata.DatabaseSummary; @@ -185,9 +187,6 @@ import org.apache.fluss.server.entity.NotifyRemoteLogOffsetsData; import org.apache.fluss.server.entity.StopReplicaData; import org.apache.fluss.server.entity.StopReplicaResultForBucket; -import org.apache.fluss.server.kv.snapshot.CompletedSnapshot; -import org.apache.fluss.server.kv.snapshot.CompletedSnapshotJsonSerde; -import org.apache.fluss.server.kv.snapshot.KvSnapshotHandle; import org.apache.fluss.server.log.FilterInfo; import org.apache.fluss.server.metadata.BucketMetadata; import org.apache.fluss.server.metadata.ClusterMetadata; @@ -200,6 +199,7 @@ import org.apache.fluss.server.zk.data.PartitionRegistration; import org.apache.fluss.server.zk.data.lake.LakeTable; import org.apache.fluss.server.zk.data.lake.LakeTableSnapshot; +import org.apache.fluss.utils.json.CompletedSnapshotJsonSerde; import org.apache.fluss.utils.json.DataTypeJsonSerde; import org.apache.fluss.utils.json.JsonSerdeUtils; import org.apache.fluss.utils.json.TableBucketOffsets; diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManagerTest.java index 2558be96f9..332ecaed7c 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManagerTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManagerTest.java @@ -17,8 +17,8 @@ package org.apache.fluss.server.coordinator; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; import org.apache.fluss.metadata.TableBucket; -import org.apache.fluss.server.kv.snapshot.CompletedSnapshot; import org.apache.fluss.server.kv.snapshot.CompletedSnapshotHandle; import org.apache.fluss.server.kv.snapshot.CompletedSnapshotHandleStore; import org.apache.fluss.server.kv.snapshot.CompletedSnapshotStore; diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java index 001a90f74e..d14d38b8ca 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java @@ -27,6 +27,7 @@ import org.apache.fluss.exception.InvalidAlterTableException; import org.apache.fluss.exception.InvalidCoordinatorException; import org.apache.fluss.fs.FsPath; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; import org.apache.fluss.metadata.DatabaseDescriptor; import org.apache.fluss.metadata.Schema; import org.apache.fluss.metadata.TableBucket; @@ -60,7 +61,6 @@ import org.apache.fluss.server.entity.CommitKvSnapshotData; import org.apache.fluss.server.entity.CommitRemoteLogManifestData; import org.apache.fluss.server.entity.TablePropertyChanges; -import org.apache.fluss.server.kv.snapshot.CompletedSnapshot; import org.apache.fluss.server.kv.snapshot.ZooKeeperCompletedSnapshotHandleStore; import org.apache.fluss.server.metadata.BucketMetadata; import org.apache.fluss.server.metadata.ClusterMetadata; diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorServiceOrphanRpcsITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorServiceOrphanRpcsITCase.java index d5bb630155..44aa2e941c 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorServiceOrphanRpcsITCase.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorServiceOrphanRpcsITCase.java @@ -18,6 +18,8 @@ package org.apache.fluss.server.coordinator; import org.apache.fluss.fs.FsPath; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; +import org.apache.fluss.kv.snapshot.KvSnapshotHandle; import org.apache.fluss.metadata.Schema; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TableDescriptor; @@ -29,11 +31,9 @@ import org.apache.fluss.rpc.messages.ListRemoteLogManifestsResponse; import org.apache.fluss.rpc.messages.PbKvSnapshot; import org.apache.fluss.rpc.messages.PbRemoteLogManifestEntry; -import org.apache.fluss.server.kv.snapshot.CompletedSnapshot; import org.apache.fluss.server.kv.snapshot.CompletedSnapshotHandle; import org.apache.fluss.server.kv.snapshot.CompletedSnapshotHandleStore; import org.apache.fluss.server.kv.snapshot.CompletedSnapshotStore; -import org.apache.fluss.server.kv.snapshot.KvSnapshotHandle; import org.apache.fluss.server.kv.snapshot.SharedKvFileRegistry; import org.apache.fluss.server.testutils.FlussClusterExtension; import org.apache.fluss.server.testutils.RpcMessageTestUtils; diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/autoinc/SegmentSequenceGeneratorTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/autoinc/SegmentSequenceGeneratorTest.java index 0070f1563f..8b269f679f 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/autoinc/SegmentSequenceGeneratorTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/autoinc/SegmentSequenceGeneratorTest.java @@ -20,6 +20,7 @@ import org.apache.fluss.exception.FlussRuntimeException; import org.apache.fluss.exception.SequenceOverflowException; +import org.apache.fluss.kv.autoinc.AutoIncIDRange; import org.apache.fluss.metadata.TablePath; import org.junit.jupiter.api.BeforeEach; diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotJsonSerdeTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotJsonSerdeTest.java index 2bd9339778..64bd80989a 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotJsonSerdeTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotJsonSerdeTest.java @@ -18,15 +18,20 @@ package org.apache.fluss.server.kv.snapshot; import org.apache.fluss.fs.FsPath; +import org.apache.fluss.kv.autoinc.AutoIncIDRange; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; +import org.apache.fluss.kv.snapshot.KvFileHandle; +import org.apache.fluss.kv.snapshot.KvFileHandleAndLocalPath; +import org.apache.fluss.kv.snapshot.KvSnapshotHandle; import org.apache.fluss.metadata.TableBucket; -import org.apache.fluss.server.kv.autoinc.AutoIncIDRange; +import org.apache.fluss.utils.json.CompletedSnapshotJsonSerde; import org.apache.fluss.utils.json.JsonSerdeTestBase; import java.util.Arrays; import java.util.Collections; import java.util.List; -/** Test for {@link org.apache.fluss.server.kv.snapshot.CompletedSnapshotJsonSerde}. */ +/** Test for {@link CompletedSnapshotJsonSerde}. */ class CompletedSnapshotJsonSerdeTest extends JsonSerdeTestBase { protected CompletedSnapshotJsonSerdeTest() { diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotStoreTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotStoreTest.java index 659ab3fbeb..aee1eb9280 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotStoreTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotStoreTest.java @@ -20,6 +20,10 @@ import org.apache.fluss.exception.FlussException; import org.apache.fluss.exception.FlussRuntimeException; import org.apache.fluss.fs.FsPath; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; +import org.apache.fluss.kv.snapshot.KvFileHandle; +import org.apache.fluss.kv.snapshot.KvFileHandleAndLocalPath; +import org.apache.fluss.kv.snapshot.KvSnapshotHandle; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.utils.concurrent.ExecutorThreadFactory; import org.apache.fluss.utils.types.Tuple2; diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotTest.java index 44b6d4f3b6..481fee4924 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotTest.java @@ -18,6 +18,10 @@ package org.apache.fluss.server.kv.snapshot; import org.apache.fluss.fs.FsPath; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; +import org.apache.fluss.kv.snapshot.KvFileHandle; +import org.apache.fluss.kv.snapshot.KvFileHandleAndLocalPath; +import org.apache.fluss.kv.snapshot.KvSnapshotHandle; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.utils.concurrent.Executors; @@ -60,7 +64,7 @@ void testCleanup(@TempDir Path tempDir) throws Exception { SharedKvFileRegistry sharedKvFileRegistry = new SharedKvFileRegistry(); // register the snapshot to a registry - snapshot.registerSharedKvFilesAfterRestored(sharedKvFileRegistry); + sharedKvFileRegistry.registerAllAfterRestored(snapshot); Executor ioExecutor = Executors.directExecutor(); snapshot.discardAsync(ioExecutor).get(); @@ -78,7 +82,11 @@ void testCleanup(@TempDir Path tempDir) throws Exception { FsPath.fromLocalFile(snapshotPath.toFile()), kvSnapshotHandle); snapshot.discardAsync(ioExecutor).get(); - // share files should be deleted since it has not been registered + // share files should be deleted since discardAll cleans both private and shared + checkCompletedSnapshotCleanUp(snapshotPath, kvSnapshotHandle, false); + + // now explicitly discard all (simulates abort/failure path) + kvSnapshotHandle.discardAll(); checkCompletedSnapshotCleanUp(snapshotPath, kvSnapshotHandle, true); } diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataDownloaderTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataDownloaderTest.java index a4b9494a5d..0a9e115efd 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataDownloaderTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataDownloaderTest.java @@ -17,6 +17,9 @@ package org.apache.fluss.server.kv.snapshot; +import org.apache.fluss.kv.snapshot.KvFileHandle; +import org.apache.fluss.kv.snapshot.KvFileHandleAndLocalPath; +import org.apache.fluss.kv.snapshot.KvSnapshotHandle; import org.apache.fluss.utils.CloseableRegistry; import org.junit.jupiter.api.AfterAll; diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataUploaderTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataUploaderTest.java index ef50485937..27862fbec2 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataUploaderTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataUploaderTest.java @@ -20,6 +20,8 @@ import org.apache.fluss.fs.FSDataInputStream; import org.apache.fluss.fs.FsPath; import org.apache.fluss.fs.local.LocalFileSystem; +import org.apache.fluss.kv.snapshot.KvFileHandle; +import org.apache.fluss.kv.snapshot.KvFileHandleAndLocalPath; import org.apache.fluss.utils.CloseableRegistry; import org.apache.fluss.utils.IOUtils; diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvTabletSnapshotTargetTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvTabletSnapshotTargetTest.java index 29773970ea..c785385559 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvTabletSnapshotTargetTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvTabletSnapshotTargetTest.java @@ -22,6 +22,8 @@ import org.apache.fluss.exception.FlussException; import org.apache.fluss.exception.FlussRuntimeException; import org.apache.fluss.fs.FsPath; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; +import org.apache.fluss.kv.snapshot.KvFileHandleAndLocalPath; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.server.SequenceIDCounter; import org.apache.fluss.server.kv.rocksdb.RocksDBExtension; diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/RocksIncrementalSnapshotTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/RocksIncrementalSnapshotTest.java index 64398516ab..489e10fe2f 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/RocksIncrementalSnapshotTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/RocksIncrementalSnapshotTest.java @@ -19,6 +19,8 @@ import org.apache.fluss.fs.FsPath; import org.apache.fluss.fs.local.LocalFileSystem; +import org.apache.fluss.kv.snapshot.KvFileHandleAndLocalPath; +import org.apache.fluss.kv.snapshot.KvSnapshotHandle; import org.apache.fluss.server.kv.rocksdb.RocksDBExtension; import org.apache.fluss.server.kv.rocksdb.RocksDBKv; import org.apache.fluss.server.testutils.KvTestUtils; @@ -135,7 +137,7 @@ void testIncrementalSnapshot(@TempDir Path snapshotBaseDir, @TempDir Path snapsh KvSnapshotHandle kvSnapshotHandle5 = snapshot(5L, incrementalSnapshot, snapshotLocation, closeableRegistry); // discard the snapshot handle - kvSnapshotHandle5.discard(); + kvSnapshotHandle5.discardAll(); // we can still restore from cp4 Path dest3 = snapshotDownDir.resolve("restore3"); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/SharedKvFileRegistryTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/SharedKvFileRegistryTest.java index a8082cb926..2594bc8c1c 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/SharedKvFileRegistryTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/SharedKvFileRegistryTest.java @@ -17,6 +17,8 @@ package org.apache.fluss.server.kv.snapshot; +import org.apache.fluss.kv.snapshot.KvFileHandle; + import org.junit.jupiter.api.Test; import java.util.Collections; diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/SnapshotsCleanerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/SnapshotsCleanerTest.java index 2e66bd945c..1e43771781 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/SnapshotsCleanerTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/SnapshotsCleanerTest.java @@ -19,6 +19,10 @@ import org.apache.fluss.exception.FlussRuntimeException; import org.apache.fluss.fs.FsPath; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; +import org.apache.fluss.kv.snapshot.KvFileHandle; +import org.apache.fluss.kv.snapshot.KvFileHandleAndLocalPath; +import org.apache.fluss.kv.snapshot.KvSnapshotHandle; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.utils.concurrent.Executors; @@ -122,7 +126,7 @@ public TestKvSnapshotHandle( } @Override - public void discard() { + public void discardPrivateFiles() { isDiscarded = true; } diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/TestingCompletedKvSnapshotCommitter.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/TestingCompletedKvSnapshotCommitter.java index 4475d0aa7b..4f5fd79176 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/TestingCompletedKvSnapshotCommitter.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/TestingCompletedKvSnapshotCommitter.java @@ -17,6 +17,7 @@ package org.apache.fluss.server.kv.snapshot; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; import org.apache.fluss.metadata.TableBucket; import java.time.Duration; diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/TestingCompletedSnapshotHandle.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/TestingCompletedSnapshotHandle.java index d8dbc8f763..df941540d7 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/TestingCompletedSnapshotHandle.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/TestingCompletedSnapshotHandle.java @@ -17,12 +17,11 @@ package org.apache.fluss.server.kv.snapshot; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; + import java.io.IOException; -/** - * Wrapping {@link org.apache.fluss.server.kv.snapshot.CompletedSnapshot} as {@link - * org.apache.fluss.server.kv.snapshot.CompletedSnapshotHandle} for testing purpose. - */ +/** Wrapping {@link CompletedSnapshot} as {@link CompletedSnapshotHandle} for testing purpose. */ public class TestingCompletedSnapshotHandle extends CompletedSnapshotHandle { private final CompletedSnapshot snapshot; diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/ZooKeeperCompletedSnapshotStoreTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/ZooKeeperCompletedSnapshotStoreTest.java index ab8110fa5d..bf514cfbd1 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/ZooKeeperCompletedSnapshotStoreTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/ZooKeeperCompletedSnapshotStoreTest.java @@ -20,6 +20,7 @@ import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; import org.apache.fluss.fs.FsPath; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.server.zk.NOPErrorHandler; import org.apache.fluss.server.zk.ZooKeeperClient; diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/KvRecoverFromRemoteLogITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/KvRecoverFromRemoteLogITCase.java index 47c5ec88f2..46558a897f 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/replica/KvRecoverFromRemoteLogITCase.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/KvRecoverFromRemoteLogITCase.java @@ -21,6 +21,7 @@ import org.apache.fluss.config.Configuration; import org.apache.fluss.config.MemorySize; import org.apache.fluss.exception.NotLeaderOrFollowerException; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TableDescriptor; import org.apache.fluss.metadata.TablePath; @@ -29,7 +30,6 @@ import org.apache.fluss.rpc.gateway.TabletServerGateway; import org.apache.fluss.rpc.messages.PbLookupRespForBucket; import org.apache.fluss.rpc.messages.PutKvRequest; -import org.apache.fluss.server.kv.snapshot.CompletedSnapshot; import org.apache.fluss.server.log.LogTablet; import org.apache.fluss.server.testutils.FlussClusterExtension; import org.apache.fluss.server.zk.ZooKeeperClient; diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/KvReplicaRestoreITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/KvReplicaRestoreITCase.java index 9aa6142f3b..db5072c135 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/replica/KvReplicaRestoreITCase.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/KvReplicaRestoreITCase.java @@ -21,6 +21,7 @@ import org.apache.fluss.config.Configuration; import org.apache.fluss.config.MemorySize; import org.apache.fluss.exception.NotLeaderOrFollowerException; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; import org.apache.fluss.metadata.LogFormat; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TableDescriptor; @@ -30,7 +31,6 @@ import org.apache.fluss.rpc.gateway.TabletServerGateway; import org.apache.fluss.rpc.messages.PbLookupRespForBucket; import org.apache.fluss.rpc.messages.PutKvRequest; -import org.apache.fluss.server.kv.snapshot.CompletedSnapshot; import org.apache.fluss.server.kv.snapshot.ZooKeeperCompletedSnapshotHandleStore; import org.apache.fluss.server.testutils.FlussClusterExtension; import org.apache.fluss.utils.ExceptionUtils; diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/KvSnapshotITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/KvSnapshotITCase.java index bf178eb2a2..750aebf449 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/replica/KvSnapshotITCase.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/KvSnapshotITCase.java @@ -20,13 +20,13 @@ import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; import org.apache.fluss.fs.FsPath; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TablePath; import org.apache.fluss.record.KvRecordBatch; import org.apache.fluss.rpc.gateway.TabletServerGateway; import org.apache.fluss.rpc.messages.PutKvRequest; import org.apache.fluss.server.coordinator.CoordinatorService; -import org.apache.fluss.server.kv.snapshot.CompletedSnapshot; import org.apache.fluss.server.kv.snapshot.ZooKeeperCompletedSnapshotHandleStore; import org.apache.fluss.server.tablet.TabletServer; import org.apache.fluss.server.testutils.FlussClusterExtension; diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java index ff6b0246ee..c424fddb52 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java @@ -26,6 +26,7 @@ import org.apache.fluss.exception.InvalidRequiredAcksException; import org.apache.fluss.exception.NotLeaderOrFollowerException; import org.apache.fluss.exception.UnknownTableOrBucketException; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; import org.apache.fluss.metadata.DataLakeFormat; import org.apache.fluss.metadata.KvFormat; import org.apache.fluss.metadata.LogFormat; @@ -67,7 +68,6 @@ import org.apache.fluss.server.entity.StopReplicaResultForBucket; import org.apache.fluss.server.kv.KvTablet; import org.apache.fluss.server.kv.rocksdb.RocksDBKv; -import org.apache.fluss.server.kv.snapshot.CompletedSnapshot; import org.apache.fluss.server.log.FetchParams; import org.apache.fluss.server.log.ListOffsetsParam; import org.apache.fluss.server.log.LogTablet; diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java index 4eefc0e2fe..42909d4c84 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java @@ -19,6 +19,7 @@ import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.exception.OutOfOrderSequenceException; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; import org.apache.fluss.metadata.LogFormat; import org.apache.fluss.metadata.PhysicalTablePath; import org.apache.fluss.metadata.SchemaGetter; @@ -35,7 +36,6 @@ import org.apache.fluss.rpc.protocol.MergeMode; import org.apache.fluss.server.entity.NotifyLeaderAndIsrData; import org.apache.fluss.server.kv.KvTablet; -import org.apache.fluss.server.kv.snapshot.CompletedSnapshot; import org.apache.fluss.server.kv.snapshot.TestingCompletedKvSnapshotCommitter; import org.apache.fluss.server.log.FetchParams; import org.apache.fluss.server.log.LogAppendInfo; @@ -617,7 +617,7 @@ void testBrokenSnapshotRecovery(@TempDir File snapshotKvTabletDir) throws Except // now simulate the latest snapshot (snapshot2) being broken by // deleting its metadata files and unshared SST files // This simulates file corruption while ZK metadata remains intact - snapshot2.getKvSnapshotHandle().discard(); + snapshot2.getKvSnapshotHandle().discardAll(); // ZK metadata should still show snapshot2 as latest (file corruption hasn't been detected // yet) diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java index bad818a9e9..9f0a87dc53 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java @@ -23,6 +23,7 @@ import org.apache.fluss.config.Configuration; import org.apache.fluss.config.MemorySize; import org.apache.fluss.fs.FsPath; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; import org.apache.fluss.metadata.PhysicalTablePath; import org.apache.fluss.metadata.Schema; import org.apache.fluss.metadata.TableBucket; @@ -39,7 +40,6 @@ import org.apache.fluss.server.kv.KvManager; import org.apache.fluss.server.kv.scan.ScannerManager; import org.apache.fluss.server.kv.snapshot.CompletedKvSnapshotCommitter; -import org.apache.fluss.server.kv.snapshot.CompletedSnapshot; import org.apache.fluss.server.kv.snapshot.KvSnapshotDataDownloader; import org.apache.fluss.server.kv.snapshot.KvSnapshotDataUploader; import org.apache.fluss.server.kv.snapshot.SnapshotContext; diff --git a/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java b/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java index aeb8a848df..37b653cfbc 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java @@ -24,6 +24,7 @@ import org.apache.fluss.config.Configuration; import org.apache.fluss.config.MemorySize; import org.apache.fluss.fs.local.LocalFileSystem; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; import org.apache.fluss.metadata.PhysicalTablePath; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TablePath; @@ -49,7 +50,6 @@ import org.apache.fluss.server.coordinator.MetadataManager; import org.apache.fluss.server.coordinator.rebalance.RebalanceManager; import org.apache.fluss.server.entity.NotifyLeaderAndIsrData; -import org.apache.fluss.server.kv.snapshot.CompletedSnapshot; import org.apache.fluss.server.kv.snapshot.CompletedSnapshotHandle; import org.apache.fluss.server.kv.snapshot.PeriodicSnapshotManager; import org.apache.fluss.server.metadata.ServerInfo; diff --git a/fluss-server/src/test/java/org/apache/fluss/server/testutils/KvTestUtils.java b/fluss-server/src/test/java/org/apache/fluss/server/testutils/KvTestUtils.java index 1ff32c7e7a..9c82f2b323 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/testutils/KvTestUtils.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/testutils/KvTestUtils.java @@ -19,6 +19,9 @@ import org.apache.fluss.config.Configuration; import org.apache.fluss.fs.FsPath; +import org.apache.fluss.kv.snapshot.CompletedSnapshot; +import org.apache.fluss.kv.snapshot.KvFileHandleAndLocalPath; +import org.apache.fluss.kv.snapshot.KvSnapshotHandle; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.rpc.messages.LookupResponse; import org.apache.fluss.rpc.messages.PbLookupRespForBucket; @@ -29,11 +32,8 @@ import org.apache.fluss.server.kv.rocksdb.RocksDBKv; import org.apache.fluss.server.kv.rocksdb.RocksDBKvBuilder; import org.apache.fluss.server.kv.rocksdb.RocksDBResourceContainer; -import org.apache.fluss.server.kv.snapshot.CompletedSnapshot; -import org.apache.fluss.server.kv.snapshot.KvFileHandleAndLocalPath; import org.apache.fluss.server.kv.snapshot.KvSnapshotDataDownloader; import org.apache.fluss.server.kv.snapshot.KvSnapshotDownloadSpec; -import org.apache.fluss.server.kv.snapshot.KvSnapshotHandle; import org.apache.fluss.server.kv.snapshot.PlaceholderKvFileHandler; import org.apache.fluss.utils.CloseableRegistry; import org.apache.fluss.utils.FileUtils;