diff --git a/fluss-common/src/main/java/org/apache/fluss/metadata/KvSnapshotFileMetadata.java b/fluss-common/src/main/java/org/apache/fluss/metadata/KvSnapshotFileMetadata.java new file mode 100644 index 00000000000..eab51e56952 --- /dev/null +++ b/fluss-common/src/main/java/org/apache/fluss/metadata/KvSnapshotFileMetadata.java @@ -0,0 +1,260 @@ +/* + * 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.metadata; + +import org.apache.fluss.annotation.Internal; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Objects; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** Immutable standard metadata stored in a KV snapshot {@code _METADATA} file. */ +@Internal +public final class KvSnapshotFileMetadata { + + private final TableBucket tableBucket; + private final long snapshotId; + private final String snapshotLocation; + private final List sharedFiles; + private final List privateFiles; + private final long incrementalSize; + private final long logOffset; + private final @Nullable Long rowCount; + private final @Nullable List autoIncrementRanges; + + /** Creates immutable standard KV snapshot file metadata. */ + public KvSnapshotFileMetadata( + TableBucket tableBucket, + long snapshotId, + String snapshotLocation, + List sharedFiles, + List privateFiles, + long incrementalSize, + long logOffset, + @Nullable Long rowCount, + @Nullable List autoIncrementRanges) { + this.tableBucket = checkNotNull(tableBucket, "Table bucket must not be null."); + this.snapshotId = snapshotId; + this.snapshotLocation = + checkNotNull(snapshotLocation, "Snapshot location must not be null."); + this.sharedFiles = immutableCopy(sharedFiles, "Shared files must not be null."); + this.privateFiles = immutableCopy(privateFiles, "Private files must not be null."); + this.incrementalSize = incrementalSize; + this.logOffset = logOffset; + this.rowCount = rowCount; + this.autoIncrementRanges = + autoIncrementRanges == null + ? null + : immutableCopy( + autoIncrementRanges, + "Auto-increment ranges must not contain null entries."); + } + + /** Returns the table bucket described by this metadata. */ + public TableBucket getTableBucket() { + return tableBucket; + } + + /** Returns the snapshot ID. */ + public long getSnapshotId() { + return snapshotId; + } + + /** Returns the snapshot location. */ + public String getSnapshotLocation() { + return snapshotLocation; + } + + /** Returns the shared snapshot files. */ + public List getSharedFiles() { + return sharedFiles; + } + + /** Returns the private snapshot files. */ + public List getPrivateFiles() { + return privateFiles; + } + + /** Returns the incremental snapshot size. */ + public long getIncrementalSize() { + return incrementalSize; + } + + /** Returns the next log offset at snapshot time. */ + public long getLogOffset() { + return logOffset; + } + + /** Returns the row count when present. */ + @Nullable + public Long getRowCount() { + return rowCount; + } + + /** Returns the auto-increment ranges when present. */ + @Nullable + public List getAutoIncrementRanges() { + return autoIncrementRanges; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + KvSnapshotFileMetadata that = (KvSnapshotFileMetadata) o; + return snapshotId == that.snapshotId + && incrementalSize == that.incrementalSize + && logOffset == that.logOffset + && Objects.equals(tableBucket, that.tableBucket) + && Objects.equals(snapshotLocation, that.snapshotLocation) + && Objects.equals(sharedFiles, that.sharedFiles) + && Objects.equals(privateFiles, that.privateFiles) + && Objects.equals(rowCount, that.rowCount) + && Objects.equals(autoIncrementRanges, that.autoIncrementRanges); + } + + @Override + public int hashCode() { + return Objects.hash( + tableBucket, + snapshotId, + snapshotLocation, + sharedFiles, + privateFiles, + incrementalSize, + logOffset, + rowCount, + autoIncrementRanges); + } + + private static List immutableCopy(List values, String message) { + checkNotNull(values, message); + ArrayList copy = new ArrayList<>(values.size()); + for (T value : values) { + copy.add(checkNotNull(value, message)); + } + return Collections.unmodifiableList(copy); + } + + /** Immutable file reference stored in standard KV snapshot metadata. */ + @Internal + public static final class FileHandle { + + private final String path; + private final long size; + private final String localPath; + + /** Creates an immutable file reference. */ + public FileHandle(String path, long size, String localPath) { + this.path = checkNotNull(path, "File path must not be null."); + this.size = size; + this.localPath = checkNotNull(localPath, "File local path must not be null."); + } + + /** Returns the remote file path. */ + public String getPath() { + return path; + } + + /** Returns the file size. */ + public long getSize() { + return size; + } + + /** Returns the local-path identity stored in the metadata. */ + public String getLocalPath() { + return localPath; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + FileHandle that = (FileHandle) o; + return size == that.size + && Objects.equals(path, that.path) + && Objects.equals(localPath, that.localPath); + } + + @Override + public int hashCode() { + return Objects.hash(path, size, localPath); + } + } + + /** Immutable auto-increment range stored in standard KV snapshot metadata. */ + @Internal + public static final class AutoIncrementRange { + + private final int columnId; + private final long start; + private final long end; + + /** Creates an immutable auto-increment range. */ + public AutoIncrementRange(int columnId, long start, long end) { + this.columnId = columnId; + this.start = start; + this.end = end; + } + + /** Returns the auto-increment column ID. */ + public int getColumnId() { + return columnId; + } + + /** Returns the range start. */ + public long getStart() { + return start; + } + + /** Returns the range end. */ + public long getEnd() { + return end; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + AutoIncrementRange that = (AutoIncrementRange) o; + return columnId == that.columnId && start == that.start && end == that.end; + } + + @Override + public int hashCode() { + return Objects.hash(columnId, start, end); + } + } +} diff --git a/fluss-common/src/main/java/org/apache/fluss/metadata/KvSnapshotFileMetadataJsonSerde.java b/fluss-common/src/main/java/org/apache/fluss/metadata/KvSnapshotFileMetadataJsonSerde.java new file mode 100644 index 00000000000..63cc5ed15e2 --- /dev/null +++ b/fluss-common/src/main/java/org/apache/fluss/metadata/KvSnapshotFileMetadataJsonSerde.java @@ -0,0 +1,181 @@ +/* + * 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.metadata; + +import org.apache.fluss.annotation.Internal; +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.List; + +/** Version-1 JSON serde for standard KV snapshot file metadata. */ +@Internal +public final class KvSnapshotFileMetadataJsonSerde + implements JsonSerializer, + JsonDeserializer { + + public static final KvSnapshotFileMetadataJsonSerde INSTANCE = + new KvSnapshotFileMetadataJsonSerde(); + + private static final int VERSION = 1; + private static final String VERSION_KEY = "version"; + 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"; + 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"; + 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"; + + private KvSnapshotFileMetadataJsonSerde() {} + + @Override + public void serialize(KvSnapshotFileMetadata metadata, JsonGenerator generator) + throws IOException { + generator.writeStartObject(); + generator.writeNumberField(VERSION_KEY, VERSION); + + TableBucket tableBucket = metadata.getTableBucket(); + generator.writeNumberField(TABLE_ID, tableBucket.getTableId()); + if (tableBucket.getPartitionId() != null) { + generator.writeNumberField(PARTITION_ID, tableBucket.getPartitionId()); + } + generator.writeNumberField(BUCKET_ID, tableBucket.getBucket()); + generator.writeNumberField(SNAPSHOT_ID, metadata.getSnapshotId()); + generator.writeStringField(SNAPSHOT_LOCATION, metadata.getSnapshotLocation()); + + generator.writeObjectFieldStart(KV_SNAPSHOT_HANDLE); + generator.writeArrayFieldStart(KV_SHARED_FILES_HANDLE); + serializeFileHandles(generator, metadata.getSharedFiles()); + generator.writeEndArray(); + generator.writeArrayFieldStart(KV_PRIVATE_FILES_HANDLE); + serializeFileHandles(generator, metadata.getPrivateFiles()); + generator.writeEndArray(); + generator.writeNumberField(SNAPSHOT_INCREMENTAL_SIZE, metadata.getIncrementalSize()); + generator.writeEndObject(); + + generator.writeNumberField(LOG_OFFSET, metadata.getLogOffset()); + if (metadata.getRowCount() != null) { + generator.writeNumberField(ROW_COUNT, metadata.getRowCount()); + } + if (metadata.getAutoIncrementRanges() != null + && !metadata.getAutoIncrementRanges().isEmpty()) { + generator.writeArrayFieldStart(AUTO_INC_ID_RANGE); + for (KvSnapshotFileMetadata.AutoIncrementRange range : + metadata.getAutoIncrementRanges()) { + generator.writeStartObject(); + generator.writeNumberField(AUTO_INC_COLUMN_ID, range.getColumnId()); + generator.writeNumberField(AUTO_INC_ID_START, range.getStart()); + generator.writeNumberField(AUTO_INC_ID_END, range.getEnd()); + generator.writeEndObject(); + } + generator.writeEndArray(); + } + generator.writeEndObject(); + } + + @Override + public KvSnapshotFileMetadata deserialize(JsonNode node) { + JsonNode partitionIdNode = node.get(PARTITION_ID); + TableBucket tableBucket = + new TableBucket( + node.get(TABLE_ID).asLong(), + partitionIdNode == null ? null : partitionIdNode.asLong(), + node.get(BUCKET_ID).asInt()); + JsonNode snapshotHandle = node.get(KV_SNAPSHOT_HANDLE); + + Long rowCount = node.has(ROW_COUNT) ? node.get(ROW_COUNT).asLong() : null; + List ranges = null; + if (node.has(AUTO_INC_ID_RANGE)) { + ranges = new ArrayList<>(); + for (JsonNode range : node.get(AUTO_INC_ID_RANGE)) { + ranges.add( + new KvSnapshotFileMetadata.AutoIncrementRange( + range.get(AUTO_INC_COLUMN_ID).asInt(), + range.get(AUTO_INC_ID_START).asLong(), + range.get(AUTO_INC_ID_END).asLong())); + } + } + + return new KvSnapshotFileMetadata( + tableBucket, + node.get(SNAPSHOT_ID).asLong(), + node.get(SNAPSHOT_LOCATION).asText(), + deserializeFileHandles(snapshotHandle, KV_SHARED_FILES_HANDLE), + deserializeFileHandles(snapshotHandle, KV_PRIVATE_FILES_HANDLE), + snapshotHandle.get(SNAPSHOT_INCREMENTAL_SIZE).asLong(), + node.get(LOG_OFFSET).asLong(), + rowCount, + ranges); + } + + /** Serializes standard KV snapshot file metadata to JSON bytes. */ + public static byte[] toJson(KvSnapshotFileMetadata metadata) { + return JsonSerdeUtils.writeValueAsBytes(metadata, INSTANCE); + } + + /** Deserializes standard KV snapshot file metadata from JSON bytes. */ + public static KvSnapshotFileMetadata fromJson(byte[] json) { + return JsonSerdeUtils.readValue(json, INSTANCE); + } + + private static void serializeFileHandles( + JsonGenerator generator, List fileHandles) + throws IOException { + for (KvSnapshotFileMetadata.FileHandle fileHandle : fileHandles) { + generator.writeStartObject(); + generator.writeObjectFieldStart(KV_FILE_HANDLE); + generator.writeStringField(KV_FILE_PATH, fileHandle.getPath()); + generator.writeNumberField(KV_FILE_SIZE, fileHandle.getSize()); + generator.writeEndObject(); + generator.writeStringField(KV_FILE_LOCAL_PATH, fileHandle.getLocalPath()); + generator.writeEndObject(); + } + } + + private static List deserializeFileHandles( + JsonNode snapshotHandle, String fieldName) { + List fileHandles = new ArrayList<>(); + for (JsonNode fileNode : snapshotHandle.get(fieldName)) { + JsonNode handleNode = fileNode.get(KV_FILE_HANDLE); + fileHandles.add( + new KvSnapshotFileMetadata.FileHandle( + handleNode.get(KV_FILE_PATH).asText(), + handleNode.get(KV_FILE_SIZE).asLong(), + fileNode.get(KV_FILE_LOCAL_PATH).asText())); + } + return fileHandles; + } +} diff --git a/fluss-common/src/test/java/org/apache/fluss/metadata/KvSnapshotFileMetadataJsonSerdeTest.java b/fluss-common/src/test/java/org/apache/fluss/metadata/KvSnapshotFileMetadataJsonSerdeTest.java new file mode 100644 index 00000000000..be7d441913b --- /dev/null +++ b/fluss-common/src/test/java/org/apache/fluss/metadata/KvSnapshotFileMetadataJsonSerdeTest.java @@ -0,0 +1,67 @@ +/* + * 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.metadata; + +import org.apache.fluss.utils.json.JsonSerdeTestBase; + +import java.util.Collections; + +/** Compatibility test for {@link KvSnapshotFileMetadataJsonSerde}. */ +class KvSnapshotFileMetadataJsonSerdeTest extends JsonSerdeTestBase { + + static final String GOLDEN_JSON = + "{\"version\":1," + + "\"table_id\":1,\"partition_id\":10,\"bucket_id\":1," + + "\"snapshot_id\":1," + + "\"snapshot_location\":\"oss://bucket/snapshot\"," + + "\"kv_snapshot_handle\":{" + + "\"shared_file_handles\":[{\"kv_file_handle\":{\"path\":\"oss://bucket/snapshot/shared/t1.sst\",\"size\":1},\"local_path\":\"localPath1\"}]," + + "\"private_file_handles\":[{\"kv_file_handle\":{\"path\":\"oss://bucket/snapshot/snapshot1/t2\",\"size\":2},\"local_path\":\"localPath2\"}]," + + "\"snapshot_incremental_size\":3},\"log_offset\":10,\"row_count\":1234," + + "\"auto_inc_id_range\":[{\"column_id\":2,\"start\":10000,\"end\":20000}]}"; + + KvSnapshotFileMetadataJsonSerdeTest() { + super(KvSnapshotFileMetadataJsonSerde.INSTANCE); + } + + @Override + protected KvSnapshotFileMetadata[] createObjects() { + return new KvSnapshotFileMetadata[] { + new KvSnapshotFileMetadata( + new TableBucket(1, 10L, 1), + 1, + "oss://bucket/snapshot", + Collections.singletonList( + new KvSnapshotFileMetadata.FileHandle( + "oss://bucket/snapshot/shared/t1.sst", 1, "localPath1")), + Collections.singletonList( + new KvSnapshotFileMetadata.FileHandle( + "oss://bucket/snapshot/snapshot1/t2", 2, "localPath2")), + 3, + 10, + 1234L, + Collections.singletonList( + new KvSnapshotFileMetadata.AutoIncrementRange(2, 10000, 20000))) + }; + } + + @Override + protected String[] expectedJsons() { + return new String[] {GOLDEN_JSON}; + } +} 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 bdafe846172..8b07baca22e 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 @@ -31,7 +31,10 @@ import org.apache.fluss.server.kv.snapshot.SharedKvFileRegistry; import org.apache.fluss.server.kv.snapshot.ZooKeeperCompletedSnapshotHandleStore; import org.apache.fluss.server.metrics.group.CoordinatorMetricGroup; +import org.apache.fluss.server.zk.ZkSequenceIDCounter; import org.apache.fluss.server.zk.ZooKeeperClient; +import org.apache.fluss.server.zk.data.BucketSnapshot; +import org.apache.fluss.server.zk.data.ZkData; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -163,6 +166,61 @@ public CompletedSnapshotStore getOrCreateCompletedSnapshotStore( }); } + /** + * Registers an immutable, externally produced snapshot and adopts its files into retention. + * + *

The caller must keep the target replicas inactive until registration completes, use a + * snapshot ID reserved from the target bucket's snapshot counter, and produce files using the + * target table's schema IDs and KV encoding. Files must belong to the target and remain + * immutable; ownership transfers to snapshot retention after registration. This method does not + * copy files, migrate schemas or coordinate concurrent table writes. + * + *

Run this operation on an IO executor. A failed or uncertain registration can be retried + * with the same handle; callers must not delete its files on failure. + */ + public void registerExternalSnapshot( + TablePath tablePath, + TableBucket tableBucket, + CompletedSnapshotHandle handle, + int coordinatorZkVersion) + throws Exception { + CompletedSnapshot snapshot = handle.retrieveCompleteSnapshot(); + checkArgument( + tableBucket.equals(snapshot.getTableBucket()), + "Snapshot bucket does not match target."); + checkArgument( + handle.getSnapshotId() == snapshot.getSnapshotID() + && handle.getLogOffset() == snapshot.getLogOffset() + && handle.getMetadataFilePath().equals(snapshot.getMetadataFilePath()), + "Snapshot metadata does not match its handle."); + checkArgument( + snapshot.getSnapshotID() >= 0 && snapshot.getLogOffset() >= 0, + "Snapshot ID and log offset must be non-negative."); + long nextSnapshotId = + new ZkSequenceIDCounter( + zooKeeperClient.getCuratorClient(), + ZkData.BucketSnapshotSequenceIdZNode.path(tableBucket)) + .getCurrent(); + checkArgument( + snapshot.getSnapshotID() < nextSnapshotId, + "External snapshot ID must be reserved from the target bucket counter."); + CompletedSnapshotStore store = getOrCreateCompletedSnapshotStore(tablePath, tableBucket); + checkArgument( + !store.getLatestSnapshot().isPresent() + || store.getLatestSnapshot().get().getSnapshotID() + <= snapshot.getSnapshotID() + || store.getActiveSnapshotIds().contains(snapshot.getSnapshotID()), + "Cannot register an older snapshot that has already been subsumed."); + zooKeeperClient.registerExternalTableBucketSnapshot( + tableBucket, + new BucketSnapshot( + handle.getSnapshotId(), + handle.getLogOffset(), + handle.getMetadataFilePath().toString()), + coordinatorZkVersion); + store.adoptAfterNodeConfirmed(snapshot); + } + public void removeCompletedSnapshotStoreByTableBuckets(Set tableBuckets) { for (TableBucket tableBucket : tableBuckets) { bucketCompletedSnapshotStores.remove(tableBucket); @@ -244,10 +302,11 @@ private CompletedSnapshotStore createCompletedSnapshotStore( } /** - * Returns active snapshot IDs per bucket for the given (tableId, partitionId) scope. For - * buckets with an in-memory {@link CompletedSnapshotStore}, the cached active set is returned - * (completed snapshots ∪ still-in-use snapshots, no retention truncation). For other buckets, - * snapshot IDs are read directly from ZK children (no per-snapshot payload fetch). + * Returns active snapshot IDs per bucket for the given (tableId, partitionId) scope. The result + * includes both cached snapshots and every persistent snapshot handle. A registered external + * snapshot must remain protected even if its registration response is lost before the in-memory + * store adopts it. Failure to read persistent handles fails the query so callers cannot mistake + * an uncertain result for an empty active set. */ public Map> getActiveSnapshotIdsByBucket( long tableId, @Nullable Long partitionId, int numBuckets) { @@ -255,11 +314,9 @@ public Map> getActiveSnapshotIdsByBucket( for (int i = 0; i < numBuckets; i++) { TableBucket tb = new TableBucket(tableId, partitionId, i); CompletedSnapshotStore store = bucketCompletedSnapshotStores.get(tb); - Set ids; + Set ids = new HashSet<>(readActiveSnapshotIdsFromZk(tb)); if (store != null) { - ids = store.getActiveSnapshotIds(); - } else { - ids = readActiveSnapshotIdsFromZk(tb); + ids.addAll(store.getActiveSnapshotIds()); } if (!ids.isEmpty()) { result.put(i, ids); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotJsonSerde.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotJsonSerde.java index 3c9201ff80e..e5856917385 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotJsonSerde.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotJsonSerde.java @@ -18,7 +18,8 @@ package org.apache.fluss.server.kv.snapshot; import org.apache.fluss.fs.FsPath; -import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metadata.KvSnapshotFileMetadata; +import org.apache.fluss.metadata.KvSnapshotFileMetadataJsonSerde; 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; @@ -30,223 +31,106 @@ import java.util.ArrayList; import java.util.List; -/** Json serializer and deserializer for {@link CompletedSnapshot}. */ +/** Adapter between server snapshot lifecycle objects and standard snapshot file metadata. */ 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 - // --------------------------------------------------------------------------------- - - // 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"; + /** Creates a serde for completed snapshots. */ + public CompletedSnapshotJsonSerde() {} @Override public void serialize(CompletedSnapshot completedSnapshot, JsonGenerator generator) throws IOException { - generator.writeStartObject(); - - // serialize data version. - generator.writeNumberField(VERSION_KEY, VERSION); - - // serialize table bucket - TableBucket tableBucket = completedSnapshot.getTableBucket(); - generator.writeNumberField(TABLE_ID, tableBucket.getTableId()); - if (tableBucket.getPartitionId() != null) { - generator.writeNumberField(PARTITION_ID, tableBucket.getPartitionId()); - } - generator.writeNumberField(BUCKET_ID, tableBucket.getBucket()); - - // serialize snapshot id - generator.writeNumberField(SNAPSHOT_ID, completedSnapshot.getSnapshotID()); - - // serialize snapshot location - generator.writeStringField( - SNAPSHOT_LOCATION, completedSnapshot.getSnapshotLocation().toString()); - - // serialize kv snapshot handle - generator.writeObjectFieldStart(KV_SNAPSHOT_HANDLE); - KvSnapshotHandle kvSnapshotHandle = completedSnapshot.getKvSnapshotHandle(); - - // serialize shared file handles - generator.writeArrayFieldStart(KV_SHARED_FILES_HANDLE); - serializeKvFileHandles(generator, kvSnapshotHandle.getSharedKvFileHandles()); - generator.writeEndArray(); - - // serialize private file handles - generator.writeArrayFieldStart(KV_PRIVATE_FILES_HANDLE); - serializeKvFileHandles(generator, kvSnapshotHandle.getPrivateFileHandles()); - generator.writeEndArray(); - - // serialize persisted size of this snapshot - generator.writeNumberField( - SNAPSHOT_INCREMENTAL_SIZE, kvSnapshotHandle.getIncrementalSize()); - generator.writeEndObject(); - - // serialize log offset - generator.writeNumberField(LOG_OFFSET, completedSnapshot.getLogOffset()); - - // ROW_COUNT and AUTO_INC_ID_RANGE are added in v0.9, but they are nullable and optional, so - // we don't bump JSON version here to guarantee the RPC protocol compatibility between - // TabletServer and CoordinatorServer. See CoordinatorGateway#commitKvSnapshot RPC. - - // serialize row count if exists - if (completedSnapshot.getRowCount() != null) { - generator.writeNumberField(ROW_COUNT, completedSnapshot.getRowCount()); - } - - // serialize auto-increment id range for each auto-increment column - if (completedSnapshot.getAutoIncIDRanges() != null - && !completedSnapshot.getAutoIncIDRanges().isEmpty()) { - generator.writeArrayFieldStart(AUTO_INC_ID_RANGE); - for (AutoIncIDRange autoIncIDRange : completedSnapshot.getAutoIncIDRanges()) { - generator.writeStartObject(); - generator.writeNumberField(AUTO_INC_COLUMN_ID, autoIncIDRange.getColumnId()); - generator.writeNumberField(AUTO_INC_ID_START, autoIncIDRange.getStart()); - generator.writeNumberField(AUTO_INC_ID_END, autoIncIDRange.getEnd()); - generator.writeEndObject(); - } - generator.writeEndArray(); - } - - generator.writeEndObject(); - } - - private void serializeKvFileHandles( - JsonGenerator generator, List kvFileHandleAndLocalPaths) - throws IOException { - for (KvFileHandleAndLocalPath fileHandleAndLocalPath : kvFileHandleAndLocalPaths) { - generator.writeStartObject(); - - // serialize kv file handle - KvFileHandle kvFileHandle = fileHandleAndLocalPath.getKvFileHandle(); - generator.writeObjectFieldStart(KV_FILE_HANDLE); - generator.writeStringField(KV_FILE_PATH, kvFileHandle.getFilePath()); - generator.writeNumberField(KV_FILE_SIZE, kvFileHandle.getSize()); - generator.writeEndObject(); - - // serialize kv file local path - generator.writeStringField(KV_FILE_LOCAL_PATH, fileHandleAndLocalPath.getLocalPath()); - - generator.writeEndObject(); - } + KvSnapshotFileMetadataJsonSerde.INSTANCE.serialize( + toFileMetadata(completedSnapshot), generator); } @Override public CompletedSnapshot deserialize(JsonNode node) { - JsonNode partitionIdNode = node.get(PARTITION_ID); - Long partitionId = partitionIdNode == null ? null : partitionIdNode.asLong(); - // deserialize table bucket - TableBucket tableBucket = - new TableBucket( - node.get(TABLE_ID).asLong(), partitionId, node.get(BUCKET_ID).asInt()); - - // deserialize snapshot id - long snapshotId = node.get(SNAPSHOT_ID).asLong(); - - // deserialize snapshot location - String snapshotLocation = node.get(SNAPSHOT_LOCATION).asText(); - - // deserialize kv snapshot file handle - JsonNode kvSnapshotFileHandleNode = node.get(KV_SNAPSHOT_HANDLE); - - // deserialize shared file handles - List sharedFileHandles = - deserializeKvFileHandles(kvSnapshotFileHandleNode, KV_SHARED_FILES_HANDLE); - - // deserialize private file handles - List privateFileHandles = - deserializeKvFileHandles(kvSnapshotFileHandleNode, KV_PRIVATE_FILES_HANDLE); - - // deserialize snapshot incremental size - long incrementalSize = kvSnapshotFileHandleNode.get(SNAPSHOT_INCREMENTAL_SIZE).asLong(); + return toCompletedSnapshot(KvSnapshotFileMetadataJsonSerde.INSTANCE.deserialize(node)); + } - // deserialize log offset - long logOffset = node.get(LOG_OFFSET).asLong(); + /** Serializes a completed snapshot to standard metadata JSON bytes. */ + public static byte[] toJson(CompletedSnapshot completedSnapshot) { + return JsonSerdeUtils.writeValueAsBytes(completedSnapshot, INSTANCE); + } - // construct CompletedSnapshot - KvSnapshotHandle kvSnapshotHandle = - KvSnapshotHandle.restore(sharedFileHandles, privateFileHandles, incrementalSize); + /** Deserializes standard metadata JSON bytes into a completed server snapshot. */ + public static CompletedSnapshot fromJson(byte[] json) { + return JsonSerdeUtils.readValue(json, INSTANCE); + } - Long rowCount = null; - if (node.has(ROW_COUNT)) { - rowCount = node.get(ROW_COUNT).asLong(); + private static KvSnapshotFileMetadata toFileMetadata(CompletedSnapshot completedSnapshot) { + KvSnapshotHandle snapshotHandle = completedSnapshot.getKvSnapshotHandle(); + List ranges = null; + if (completedSnapshot.getAutoIncIDRanges() != null) { + ranges = new ArrayList<>(); + for (AutoIncIDRange range : completedSnapshot.getAutoIncIDRanges()) { + ranges.add( + new KvSnapshotFileMetadata.AutoIncrementRange( + range.getColumnId(), range.getStart(), range.getEnd())); + } } + return new KvSnapshotFileMetadata( + completedSnapshot.getTableBucket(), + completedSnapshot.getSnapshotID(), + completedSnapshot.getSnapshotLocation().toString(), + toFileHandles(snapshotHandle.getSharedKvFileHandles()), + toFileHandles(snapshotHandle.getPrivateFileHandles()), + snapshotHandle.getIncrementalSize(), + completedSnapshot.getLogOffset(), + completedSnapshot.getRowCount(), + ranges); + } - List autoIncIDRanges = null; - if (node.has(AUTO_INC_ID_RANGE)) { - autoIncIDRanges = new ArrayList<>(); - for (JsonNode autoIncIDRangeNode : node.get(AUTO_INC_ID_RANGE)) { - int columnId = autoIncIDRangeNode.get(AUTO_INC_COLUMN_ID).asInt(); - long start = autoIncIDRangeNode.get(AUTO_INC_ID_START).asLong(); - long end = autoIncIDRangeNode.get(AUTO_INC_ID_END).asLong(); - autoIncIDRanges.add(new AutoIncIDRange(columnId, start, end)); + /** Converts already-parsed standard metadata into a completed server snapshot. */ + public static CompletedSnapshot toCompletedSnapshot(KvSnapshotFileMetadata metadata) { + List ranges = null; + if (metadata.getAutoIncrementRanges() != null) { + ranges = new ArrayList<>(); + for (KvSnapshotFileMetadata.AutoIncrementRange range : + metadata.getAutoIncrementRanges()) { + ranges.add( + new AutoIncIDRange(range.getColumnId(), range.getStart(), range.getEnd())); } } - return new CompletedSnapshot( - tableBucket, - snapshotId, - new FsPath(snapshotLocation), - kvSnapshotHandle, - logOffset, - rowCount, - autoIncIDRanges); + metadata.getTableBucket(), + metadata.getSnapshotId(), + new FsPath(metadata.getSnapshotLocation()), + KvSnapshotHandle.restore( + toServerFileHandles(metadata.getSharedFiles()), + toServerFileHandles(metadata.getPrivateFiles()), + metadata.getIncrementalSize()), + metadata.getLogOffset(), + metadata.getRowCount(), + ranges); } - private List deserializeKvFileHandles( - JsonNode node, String kvHandleType) { - List kvFileHandleAndLocalPaths = new ArrayList<>(); - for (JsonNode kvFileHandleAndLocalPathNode : node.get(kvHandleType)) { - // deserialize kv file handle - JsonNode kvFileHandleNode = kvFileHandleAndLocalPathNode.get(KV_FILE_HANDLE); - String filePath = kvFileHandleNode.get(KV_FILE_PATH).asText(); - long fileSize = kvFileHandleNode.get(KV_FILE_SIZE).asLong(); - KvFileHandle kvFileHandle = new KvFileHandle(filePath, fileSize); - - // deserialize kv file local path - String localPath = kvFileHandleAndLocalPathNode.get(KV_FILE_LOCAL_PATH).asText(); - KvFileHandleAndLocalPath kvFileHandleAndLocalPath = - KvFileHandleAndLocalPath.of(kvFileHandle, localPath); - kvFileHandleAndLocalPaths.add(kvFileHandleAndLocalPath); + private static List toFileHandles( + List serverHandles) { + List handles = new ArrayList<>(serverHandles.size()); + for (KvFileHandleAndLocalPath serverHandle : serverHandles) { + handles.add( + new KvSnapshotFileMetadata.FileHandle( + serverHandle.getKvFileHandle().getFilePath(), + serverHandle.getKvFileHandle().getSize(), + serverHandle.getLocalPath())); } - return kvFileHandleAndLocalPaths; + return handles; } - /** Serialize the {@link CompletedSnapshot} to json bytes. */ - public static byte[] toJson(CompletedSnapshot completedSnapshot) { - return JsonSerdeUtils.writeValueAsBytes(completedSnapshot, INSTANCE); - } - - /** Deserialize the json bytes to {@link CompletedSnapshot}. */ - public static CompletedSnapshot fromJson(byte[] json) { - return JsonSerdeUtils.readValue(json, INSTANCE); + private static List toServerFileHandles( + List metadataHandles) { + List handles = new ArrayList<>(metadataHandles.size()); + for (KvSnapshotFileMetadata.FileHandle metadataHandle : metadataHandles) { + handles.add( + KvFileHandleAndLocalPath.of( + new KvFileHandle(metadataHandle.getPath(), metadataHandle.getSize()), + metadataHandle.getLocalPath())); + } + return handles; } } 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 7cf3e88db60..4c0d5ffe030 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 @@ -43,6 +43,7 @@ import java.util.concurrent.locks.ReentrantLock; import static org.apache.fluss.utils.Preconditions.checkNotNull; +import static org.apache.fluss.utils.Preconditions.checkState; import static org.apache.fluss.utils.concurrent.LockUtils.inLock; /* This file is based on source code of Apache Flink Project (https://flink.apache.org/), licensed by the Apache @@ -116,6 +117,47 @@ public void add(final CompletedSnapshot completedSnapshot) throws Exception { completedSnapshot, snapshotsCleaner, () -> {})); } + /** + * Adopts a snapshot whose persistent snapshot node has already been confirmed. + * + *

The operation is idempotent for an identical physical bucket and snapshot identity. A + * different snapshot using the same ID is rejected instead of replacing the confirmed node. + */ + public void adoptAfterNodeConfirmed(final CompletedSnapshot snapshot) throws Exception { + checkNotNull(snapshot, "Snapshot"); + inLock( + lock, + () -> { + for (CompletedSnapshot existing : completedSnapshots) { + if (existing.getSnapshotID() == snapshot.getSnapshotID()) { + checkState( + existing.equals(snapshot), + "Conflicting snapshot identity for %s snapshot %s.", + snapshot.getTableBucket(), + snapshot.getSnapshotID()); + return; + } + } + CompletedSnapshot stillInUse = + stillInUseSnapshots.get(snapshot.getSnapshotID()); + if (stillInUse != null) { + checkState( + stillInUse.equals(snapshot), + "Conflicting snapshot identity for %s snapshot %s.", + snapshot.getTableBucket(), + snapshot.getSnapshotID()); + return; + } + checkState( + completedSnapshots.isEmpty() + || completedSnapshots.peekLast().getSnapshotID() + < snapshot.getSnapshotID(), + "Cannot adopt an older snapshot %s.", + snapshot.getSnapshotID()); + adoptConfirmedSnapshot(snapshot, snapshotsCleaner, () -> {}); + }); + } + public long getPhysicalStorageRemoteKvSize() { return sharedKvFileRegistry.getFileSize(); } @@ -159,60 +201,60 @@ void addSnapshotAndSubsumeOldestOne( throws Exception { checkNotNull(snapshot, "Snapshot"); - // register the completed snapshot to the shared registry - snapshot.registerSharedKvFilesAfterRestored(sharedKvFileRegistry); - CompletedSnapshotHandle completedSnapshotHandle = store(snapshot); completedSnapshotHandleStore.add( snapshot.getTableBucket(), snapshot.getSnapshotID(), completedSnapshotHandle); - // Now add the new one. If it fails, we don't want to lose existing data. - inLock( - lock, - () -> { - completedSnapshots.addLast(snapshot); - - // Remove completed snapshot from queue and snapshotStateHandleStore, not - // discard. - subsume( - completedSnapshots, - maxNumberOfSnapshotsToRetain, - completedSnapshot -> { - if (snapshotInUseChecker.isInUse(completedSnapshot)) { - LOG.debug( - "Snapshot {} is still in use, move it to stillInUseSnapshots", - completedSnapshot.getSnapshotID()); - stillInUseSnapshots.put( - completedSnapshot.getSnapshotID(), completedSnapshot); - } else { - remove( - completedSnapshot.getTableBucket(), - completedSnapshot.getSnapshotID()); - snapshotsCleaner.addSubsumedSnapshot(completedSnapshot); - } - }); - - // Check if any previously still-in-use snapshots can now be released - // (lease expired). - removeUnusedSnapshots(snapshotsCleaner); - - // SST file cleanup: compute effective lowest from retained (non-leased) - // snapshots only, and protect files referenced by still-in-use snapshots. - Set stillInUseIds = new HashSet<>(stillInUseSnapshots.keySet()); - findLowest(completedSnapshots) - .ifPresent( - id -> - sharedKvFileRegistry.unregisterUnusedKvFile( - id, stillInUseIds)); - - // 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 - // metadata, not shared SST files registered in SharedKvFileRegistry. - snapshotsCleaner.cleanSubsumedSnapshots( - snapshot.getSnapshotID() + 1, stillInUseIds, postCleanup, ioExecutor); + adoptConfirmedSnapshot(snapshot, snapshotsCleaner, postCleanup); + } + + /** + * Makes a snapshot whose persistent node is already confirmed visible: shared handles are + * exposed only now, never before the node exists, and retention is applied. + */ + private void adoptConfirmedSnapshot( + CompletedSnapshot snapshot, SnapshotsCleaner snapshotsCleaner, Runnable postCleanup) + throws Exception { + snapshot.registerSharedKvFilesAfterRestored(sharedKvFileRegistry); + completedSnapshots.addLast(snapshot); + + // Remove completed snapshot from queue and snapshotStateHandleStore, not + // discard. + subsume( + completedSnapshots, + maxNumberOfSnapshotsToRetain, + completedSnapshot -> { + if (snapshotInUseChecker.isInUse(completedSnapshot)) { + LOG.debug( + "Snapshot {} is still in use, move it to stillInUseSnapshots", + completedSnapshot.getSnapshotID()); + stillInUseSnapshots.put( + completedSnapshot.getSnapshotID(), completedSnapshot); + } else { + remove( + completedSnapshot.getTableBucket(), + completedSnapshot.getSnapshotID()); + snapshotsCleaner.addSubsumedSnapshot(completedSnapshot); + } }); + + // Check if any previously still-in-use snapshots can now be released + // (lease expired). + removeUnusedSnapshots(snapshotsCleaner); + + // SST file cleanup: compute effective lowest from retained (non-leased) + // snapshots only, and protect files referenced by still-in-use snapshots. + Set stillInUseIds = new HashSet<>(stillInUseSnapshots.keySet()); + findLowest(completedSnapshots) + .ifPresent(id -> sharedKvFileRegistry.unregisterUnusedKvFile(id, stillInUseIds)); + + // 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 + // metadata, not shared SST files registered in SharedKvFileRegistry. + snapshotsCleaner.cleanSubsumedSnapshots( + snapshot.getSnapshotID() + 1, stillInUseIds, postCleanup, ioExecutor); } private void removeUnusedSnapshots(SnapshotsCleaner snapshotsCleaner) throws Exception { diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java b/fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java index 4c1a84dd1fa..50b609c3d8f 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java @@ -421,6 +421,22 @@ public void truncateFullyAndStartAt(TableBucket tableBucket, long newOffset) { } } + /** Durably initializes an empty local log at the given snapshot offset. */ + public void initializeEmptyLocalTail(TableBucket tableBucket, long endOffset) { + LogTablet logTablet = currentLogs.get(tableBucket); + if (logTablet == null) { + throw new LogStorageException("Log tablet does not exist for " + tableBucket + "."); + } + logTablet.initializeEmptyLocalTail(endOffset); + try { + logTablet.flush(true); + } catch (IOException e) { + throw new LogStorageException( + "Failed to durably initialize the local tail for " + tableBucket + ".", e); + } + checkpointRecoveryOffsets(logTablet.getDataDir()); + } + private LogTablet loadLog( File dataDir, File tabletDir, diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java b/fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java index cb37ae40e1c..24f040dc891 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java @@ -70,6 +70,7 @@ import static org.apache.fluss.utils.Preconditions.checkArgument; import static org.apache.fluss.utils.Preconditions.checkNotNull; +import static org.apache.fluss.utils.Preconditions.checkState; /* This file is based on source code of Apache Kafka Project (https://kafka.apache.org/), licensed by the Apache * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for @@ -1166,6 +1167,25 @@ boolean truncateTo(long targetOffset) throws LogStorageException { } } + /** Initializes an unused local log, or accepts a retry at the same empty boundary. */ + void initializeEmptyLocalTail(long endOffset) { + synchronized (lock) { + checkArgument(endOffset >= 0L, "Invalid initial log offset %s.", endOffset); + long startOffset = localLogStartOffset(); + long currentEndOffset = localLogEndOffset(); + checkState( + startOffset == currentEndOffset + && getHighWatermark() == currentEndOffset + && (currentEndOffset == 0L || currentEndOffset == endOffset), + "Cannot initialize nonempty or conflicting local log for %s at offset %s.", + getTableBucket(), + endOffset); + if (currentEndOffset != endOffset) { + truncateFullyAndStartAt(endOffset); + } + } + } + /** Delete all data in the log and start at the new offset. */ void truncateFullyAndStartAt(long newOffset) throws LogStorageException { LOG.debug("Truncate and start at offset {} for bucket {}", newOffset, getTableBucket()); 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 079e1f96502..b9748350096 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 @@ -611,8 +611,6 @@ private void onBecomeNewLeader() { // Clear standby flag — a leader is never a standby replica. isStandbyReplica = false; - updateLeaderEndOffsetSnapshot(); - if (isDataLakeEnabled()) { registerLakeTieringMetrics(); } @@ -625,6 +623,8 @@ private void onBecomeNewLeader() { // now, we can create a new kv tablet createKv(); } + + updateLeaderEndOffsetSnapshot(); } private void registerLakeTieringMetrics() { @@ -748,6 +748,15 @@ private void createKv() { break; } catch (Exception e) { lastError = e; + if (kvTablet != null) { + try { + checkNotNull(kvManager).dropKv(tableBucket); + kvTablet = null; + } catch (Exception cleanupError) { + e.addSuppressed(cleanupError); + break; + } + } LOG.warn( "Failed to init kv tablet for bucket {} on attempt {}/{}.", tableBucket, @@ -870,6 +879,20 @@ private Optional initKvTablet() { checkNotNull(kvTablet, "kv tablet should not be null."); restoreStartOffset = completedSnapshot.getLogOffset(); + if (restoreStartOffset > 0L + && !snapshotContext + .getZooKeeperClient() + .getRemoteLogManifestHandle(tableBucket) + .isPresent()) { + if (logTablet.localLogEndOffset() == 0L) { + logManager.initializeEmptyLocalTail(tableBucket, restoreStartOffset); + } + checkState( + logTablet.localLogEndOffset() >= restoreStartOffset, + "Local log ends before snapshot offset %s for %s without remote logs.", + restoreStartOffset, + tableBucket); + } rowCount = supportsExactRowCount(tableConfig) ? completedSnapshot.getRowCount() : null; // currently, we only support one auto-increment column. @@ -987,13 +1010,9 @@ private Optional getLatestSnapshot(TableBucket tableBucket) { return Optional.ofNullable( snapshotContext.getLatestCompletedSnapshotProvider().apply(tableBucket)); } catch (Exception e) { - LOG.warn( - "Get latest completed snapshot for {} of table {} failed.", - tableBucket, - physicalPath, - e); + throw new KvStorageException( + "Failed to get the latest completed snapshot for " + tableBucket + '.', e); } - return Optional.empty(); } private void recoverKvTablet( diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java index 1911196095f..1cf090e8345 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java @@ -132,6 +132,7 @@ import static java.util.stream.Collectors.toMap; import static org.apache.fluss.metadata.ResolvedPartitionSpec.fromPartitionName; import static org.apache.fluss.server.zk.ZooKeeperOp.multiRequest; +import static org.apache.fluss.utils.Preconditions.checkArgument; import static org.apache.fluss.utils.Preconditions.checkNotNull; /** @@ -1176,6 +1177,43 @@ public void registerTableBucketSnapshot(TableBucket tableBucket, BucketSnapshot .forPath(path, BucketSnapshotIdZNode.encode(snapshot)); } + /** + * Registers an externally produced snapshot under the active coordinator epoch. Retrying an + * identical registration is safe; a conflicting handle is never overwritten. An uncertain + * result leaves the metadata file intact so registration can be retried. + */ + public void registerExternalTableBucketSnapshot( + TableBucket tableBucket, BucketSnapshot snapshot, int coordinatorZkVersion) + throws Exception { + checkArgument(coordinatorZkVersion >= 0, "A coordinator epoch version is required."); + String path = BucketSnapshotIdZNode.path(tableBucket, snapshot.getSnapshotId()); + createRecursiveWithEpochCheck( + BucketSnapshotsZNode.path(tableBucket), null, coordinatorZkVersion, false); + try { + zkClient.transaction() + .forOperations( + wrapRequestWithEpochCheck( + zkOp.createOp( + path, + BucketSnapshotIdZNode.encode(snapshot), + CreateMode.PERSISTENT), + coordinatorZkVersion)); + } catch (KeeperException.NodeExistsException e) { + Stat stat = new Stat(); + BucketSnapshot existing = + BucketSnapshotIdZNode.decode( + zkClient.getData().storingStatIn(stat).forPath(path)); + checkArgument( + existing.equals(snapshot), + "Conflicting snapshot registration for %s.", + tableBucket); + zkClient.transaction() + .forOperations( + wrapRequestWithEpochCheck( + zkOp.checkOp(path, stat.getVersion()), coordinatorZkVersion)); + } + } + public void deleteTableBucketSnapshot(TableBucket tableBucket, long snapshotId) throws Exception { String path = BucketSnapshotIdZNode.path(tableBucket, snapshotId); 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 0f646726972..86418492691 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,19 +17,25 @@ package org.apache.fluss.server.coordinator; +import org.apache.fluss.fs.FSDataOutputStream; +import org.apache.fluss.fs.FileSystem; import org.apache.fluss.fs.FsPath; 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.CompletedSnapshotJsonSerde; import org.apache.fluss.server.kv.snapshot.CompletedSnapshotStore; import org.apache.fluss.server.kv.snapshot.TestingCompletedSnapshotHandle; import org.apache.fluss.server.kv.snapshot.ZooKeeperCompletedSnapshotHandleStore; import org.apache.fluss.server.metrics.group.TestingMetricGroups; import org.apache.fluss.server.testutils.KvTestUtils; import org.apache.fluss.server.zk.NOPErrorHandler; +import org.apache.fluss.server.zk.ZkSequenceIDCounter; import org.apache.fluss.server.zk.ZooKeeperClient; import org.apache.fluss.server.zk.ZooKeeperExtension; +import org.apache.fluss.server.zk.data.ZkData; +import org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.KeeperException; import org.apache.fluss.testutils.common.AllCallbackWrapper; import org.junit.jupiter.api.AfterAll; @@ -60,6 +66,11 @@ import static org.apache.fluss.record.TestData.DATA1_TABLE_PATH; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doCallRealMethod; +import static org.mockito.Mockito.spy; /** Test for {@link CompletedSnapshotStoreManager}. */ class CompletedSnapshotStoreManagerTest { @@ -276,6 +287,156 @@ private void verifyMissingSnapshotMetadataIsCleanedUp(IOException exception) thr assertThat(completedSnapshotHandleStore.get(tableBucket, 2L)).isEmpty(); } + @Test + void testExternalSnapshotRegistrationRetryAndRecovery() throws Exception { + TableBucket bucket = new TableBucket(99, 0); + int epoch = + zookeeperClient + .fenceBecomeCoordinatorLeader("first") + .getCoordinatorEpochZkVersion(); + long id = + new ZkSequenceIDCounter( + zookeeperClient.getCuratorClient(), + ZkData.BucketSnapshotSequenceIdZNode.path(bucket)) + .getAndIncrement(); + CompletedSnapshot snapshot = KvTestUtils.mockCompletedSnapshot(tempDir, bucket, id); + CompletedSnapshotHandle handle = writeExternalSnapshot(snapshot); + CompletedSnapshotStoreManager manager = createCompletedSnapshotStoreManager(1); + CompletedSnapshotStore store = + manager.getOrCreateCompletedSnapshotStore(DATA1_TABLE_PATH, bucket); + manager.registerExternalSnapshot(DATA1_TABLE_PATH, bucket, handle, epoch); + manager.registerExternalSnapshot(DATA1_TABLE_PATH, bucket, handle, epoch); + assertThat(store.getAllSnapshots()).containsExactly(snapshot); + assertThat(handle.retrieveCompleteSnapshot()).isEqualTo(snapshot); + assertThat( + createCompletedSnapshotStoreManager(1) + .getOrCreateCompletedSnapshotStore(DATA1_TABLE_PATH, bucket) + .getAllSnapshots()) + .containsExactly(snapshot); + zookeeperClient.fenceBecomeCoordinatorLeader("second"); + assertThatThrownBy( + () -> + manager.registerExternalSnapshot( + DATA1_TABLE_PATH, bucket, handle, epoch)) + .isInstanceOf(KeeperException.BadVersionException.class); + assertThat(handle.retrieveCompleteSnapshot()).isEqualTo(snapshot); + } + + @Test + void testExternalSnapshotRejectsMismatchedAndConflictingIdentity() throws Exception { + TableBucket bucket = new TableBucket(99, 1); + int epoch = + zookeeperClient + .fenceBecomeCoordinatorLeader("coordinator") + .getCoordinatorEpochZkVersion(); + CompletedSnapshot snapshot = KvTestUtils.mockCompletedSnapshot(tempDir, bucket, 0); + CompletedSnapshotHandle handle = writeExternalSnapshot(snapshot); + CompletedSnapshotStoreManager manager = createCompletedSnapshotStoreManager(1); + assertThatThrownBy( + () -> + manager.registerExternalSnapshot( + DATA1_TABLE_PATH, new TableBucket(99, 2), handle, epoch)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("bucket"); + assertThatThrownBy( + () -> + manager.registerExternalSnapshot( + DATA1_TABLE_PATH, bucket, handle, epoch)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("reserved"); + new ZkSequenceIDCounter( + zookeeperClient.getCuratorClient(), + ZkData.BucketSnapshotSequenceIdZNode.path(bucket)) + .getAndIncrement(); + assertThatThrownBy( + () -> + manager.registerExternalSnapshot( + DATA1_TABLE_PATH, + bucket, + new CompletedSnapshotHandle( + 0, handle.getMetadataFilePath(), 1), + epoch)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("handle"); + assertThat(zookeeperClient.getTableBucketSnapshot(bucket, 0)).isEmpty(); + manager.registerExternalSnapshot(DATA1_TABLE_PATH, bucket, handle, epoch); + CompletedSnapshot conflict = + KvTestUtils.mockCompletedSnapshot(tempDir.resolve("conflict"), bucket, 0); + CompletedSnapshotHandle conflictingHandle = writeExternalSnapshot(conflict); + assertThatThrownBy( + () -> + manager.registerExternalSnapshot( + DATA1_TABLE_PATH, bucket, conflictingHandle, epoch)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Conflicting"); + assertThat(completedSnapshotHandleStore.get(bucket, 0).get().retrieveCompleteSnapshot()) + .isEqualTo(snapshot); + assertThat(handle.retrieveCompleteSnapshot()).isEqualTo(snapshot); + assertThat(conflictingHandle.retrieveCompleteSnapshot()).isEqualTo(conflict); + assertThat( + manager.getOrCreateCompletedSnapshotStore(DATA1_TABLE_PATH, bucket) + .getAllSnapshots()) + .containsExactly(snapshot); + } + + @Test + void testExternalSnapshotRemainsActiveWhenRegistrationResponseIsLost() throws Exception { + TableBucket bucket = new TableBucket(99, 0); + int epoch = + zookeeperClient + .fenceBecomeCoordinatorLeader("coordinator") + .getCoordinatorEpochZkVersion(); + long id = + new ZkSequenceIDCounter( + zookeeperClient.getCuratorClient(), + ZkData.BucketSnapshotSequenceIdZNode.path(bucket)) + .getAndIncrement(); + CompletedSnapshot snapshot = KvTestUtils.mockCompletedSnapshot(tempDir, bucket, id); + CompletedSnapshotHandle handle = writeExternalSnapshot(snapshot); + ZooKeeperClient failingClient = spy(zookeeperClient); + CompletedSnapshotStoreManager manager = + new CompletedSnapshotStoreManager( + 1, + ioExecutor, + failingClient, + TestingMetricGroups.COORDINATOR_METRICS, + ignored -> false); + CompletedSnapshotStore store = + manager.getOrCreateCompletedSnapshotStore(DATA1_TABLE_PATH, bucket); + doAnswer( + invocation -> { + invocation.callRealMethod(); + assertThat(store.getNumSnapshots()).isZero(); + assertThat(manager.getActiveSnapshotIdsByBucket(99, null, 1).get(0)) + .containsExactly(id); + throw new KeeperException.ConnectionLossException(); + }) + .when(failingClient) + .registerExternalTableBucketSnapshot(any(), any(), anyInt()); + assertThatThrownBy( + () -> + manager.registerExternalSnapshot( + DATA1_TABLE_PATH, bucket, handle, epoch)) + .isInstanceOf(KeeperException.ConnectionLossException.class); + assertThat(manager.getActiveSnapshotIdsByBucket(99, null, 1).get(0)).containsExactly(id); + assertThat(handle.retrieveCompleteSnapshot()).isEqualTo(snapshot); + doCallRealMethod() + .when(failingClient) + .registerExternalTableBucketSnapshot(any(), any(), anyInt()); + manager.registerExternalSnapshot(DATA1_TABLE_PATH, bucket, handle, epoch); + assertThat(store.getAllSnapshots()).containsExactly(snapshot); + } + + private static CompletedSnapshotHandle writeExternalSnapshot(CompletedSnapshot snapshot) + throws Exception { + FsPath path = snapshot.getMetadataFilePath(); + try (FSDataOutputStream output = + path.getFileSystem().create(path, FileSystem.WriteMode.NO_OVERWRITE)) { + output.write(CompletedSnapshotJsonSerde.toJson(snapshot)); + } + return new CompletedSnapshotHandle(snapshot.getSnapshotID(), path, snapshot.getLogOffset()); + } + private CompletedSnapshotStoreManager createCompletedSnapshotStoreManager( int maxNumberOfSnapshotsToRetain) { return new CompletedSnapshotStoreManager( diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/ExternalKvSnapshotITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/ExternalKvSnapshotITCase.java new file mode 100644 index 00000000000..99507c5f791 --- /dev/null +++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/ExternalKvSnapshotITCase.java @@ -0,0 +1,335 @@ +/* + * 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.server.coordinator; + +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.fs.FSDataOutputStream; +import org.apache.fluss.fs.FileSystem; +import org.apache.fluss.fs.FsPath; +import org.apache.fluss.metadata.KvSnapshotFileMetadata; +import org.apache.fluss.metadata.KvSnapshotFileMetadataJsonSerde; +import org.apache.fluss.metadata.PhysicalTablePath; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metadata.TableDescriptor; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.record.KvRecord; +import org.apache.fluss.rpc.gateway.TabletServerGateway; +import org.apache.fluss.rpc.messages.NotifyLeaderAndIsrRequest; +import org.apache.fluss.rpc.messages.PutKvResponse; +import org.apache.fluss.server.entity.NotifyLeaderAndIsrData; +import org.apache.fluss.server.kv.rocksdb.RocksDBExtension; +import org.apache.fluss.server.kv.snapshot.CompletedSnapshot; +import org.apache.fluss.server.kv.snapshot.CompletedSnapshotHandle; +import org.apache.fluss.server.kv.snapshot.KvFileHandleAndLocalPath; +import org.apache.fluss.server.kv.snapshot.KvSnapshotDataUploader; +import org.apache.fluss.server.kv.snapshot.KvSnapshotHandle; +import org.apache.fluss.server.kv.snapshot.RocksIncrementalSnapshot; +import org.apache.fluss.server.kv.snapshot.SnapshotLocation; +import org.apache.fluss.server.kv.snapshot.TabletState; +import org.apache.fluss.server.replica.Replica; +import org.apache.fluss.server.replica.ReplicaManager; +import org.apache.fluss.server.testutils.FlussClusterExtension; +import org.apache.fluss.server.utils.ResourceGuard; +import org.apache.fluss.server.zk.ZkSequenceIDCounter; +import org.apache.fluss.server.zk.data.ZkData; +import org.apache.fluss.utils.CloseableRegistry; +import org.apache.fluss.utils.FlussPaths; +import org.apache.fluss.utils.types.Tuple2; + +import org.junit.jupiter.api.extension.RegisterExtension; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.stream.Collectors; + +import static org.apache.fluss.record.TestData.DATA1_SCHEMA_PK; +import static org.apache.fluss.server.testutils.KvTestUtils.assertLookupResponse; +import static org.apache.fluss.server.testutils.RpcMessageTestUtils.createTable; +import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newLookupRequest; +import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newPutKvRequest; +import static org.apache.fluss.server.utils.ServerRpcMessageUtils.makeNotifyBucketLeaderAndIsr; +import static org.apache.fluss.server.utils.ServerRpcMessageUtils.makeNotifyLeaderAndIsrRequest; +import static org.apache.fluss.testutils.DataTestUtils.genKvRecords; +import static org.apache.fluss.testutils.DataTestUtils.getKeyValuePairs; +import static org.apache.fluss.testutils.DataTestUtils.toKvRecordBatch; +import static org.apache.fluss.testutils.common.CommonTestUtils.retry; +import static org.assertj.core.api.Assertions.assertThat; + +/** Registration, recovery and replication from an externally produced KV snapshot. */ +class ExternalKvSnapshotITCase { + + @RegisterExtension public final RocksDBExtension rocksDB = new RocksDBExtension(); + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testSnapshotOnlyRecoveryAndOnlineWrites(boolean remoteLogEnabled) throws Exception { + Configuration conf = new Configuration(); + conf.set(ConfigOptions.KV_SNAPSHOT_INTERVAL, Duration.ofHours(1)); + conf.set(ConfigOptions.KV_MAX_RETAINED_SNAPSHOTS, 1); + conf.set( + ConfigOptions.REMOTE_LOG_TASK_INTERVAL_DURATION, + remoteLogEnabled ? Duration.ofHours(1) : Duration.ZERO); + FlussClusterExtension cluster = + FlussClusterExtension.builder() + .setNumOfTabletServers(2) + .setClusterConf(conf) + .build(); + try { + cluster.start(); + TablePath path = TablePath.of("snapshot_db", "snapshot_boundary"); + long tableId = + createTable( + cluster, + path, + TableDescriptor.builder() + .schema(DATA1_SCHEMA_PK) + .distributedBy(1, "a") + .build() + .withReplicationFactor(2)); + TableBucket bucket = new TableBucket(tableId, 0); + cluster.waitUntilAllReplicaReady(bucket); + List records = new ArrayList<>(); + records.addAll(genKvRecords(new Object[] {1, "snapshot-one"})); + records.addAll(genKvRecords(new Object[] {2, "snapshot-two"})); + // The target has no online writes. Keep its replicas inactive while registering + // files produced for this table, then let normal role notifications restore them. + for (int server = 0; server < 2; server++) { + cluster.stopTabletServer(server); + } + long boundary = 10_017L; + CompletedSnapshotHandle external = + produceSnapshot(cluster, path, bucket, records, boundary); + CompletedSnapshotStoreManager manager = + cluster.getCoordinatorServer() + .getCoordinatorEventProcessor() + .completedSnapshotStoreManager(); + int epoch = + cluster.getZooKeeperClient().getCurrentEpoch().getCoordinatorEpochZkVersion(); + manager.registerExternalSnapshot(path, bucket, external, epoch); + manager.registerExternalSnapshot(path, bucket, external, epoch); + assertThat(manager.getOrCreateCompletedSnapshotStore(path, bucket).getNumSnapshots()) + .isEqualTo(1); + cluster.stopCoordinatorServer(); + cluster.startCoordinatorServer(); + for (int server = 0; server < 2; server++) { + cluster.startTabletServer(server); + } + cluster.waitUntilAllReplicaReady(bucket); + assertReplicaOffsets(cluster, bucket, boundary, boundary); + assertRows(cluster, bucket, records); + + List firstTail = genKvRecords(new Object[] {3, "first-online-write"}); + putRecords(cluster, bucket, firstTail); + records.addAll(firstTail); + assertReplicaOffsets(cluster, bucket, boundary, boundary + 1L); + assertRows(cluster, bucket, records); + + NotifyLeaderAndIsrRequest repeatedNotify = + makeNotifyLeaderAndIsrRequest( + cluster.getZooKeeperClient().getCurrentEpoch().getCoordinatorEpoch(), + Collections.singletonList( + makeNotifyBucketLeaderAndIsr( + new NotifyLeaderAndIsrData( + PhysicalTablePath.of(path), + bucket, + Arrays.asList(0, 1), + cluster.waitLeaderAndIsrReady(bucket))))); + for (int server = 0; server < 2; server++) { + assertThat( + cluster.newTabletServerClientForNode(server) + .notifyLeaderAndIsr(repeatedNotify) + .get() + .getNotifyBucketsLeaderRespAt(0) + .hasErrorCode()) + .isFalse(); + } + assertReplicaOffsets(cluster, bucket, boundary, boundary + 1L); + assertRows(cluster, bucket, records); + + int oldLeader = cluster.waitAndGetLeader(bucket); + cluster.stopTabletServer(oldLeader); + retry( + Duration.ofMinutes(1), + () -> assertThat(cluster.waitAndGetLeader(bucket)).isNotEqualTo(oldLeader)); + assertRows(cluster, bucket, records); + List nextTail = genKvRecords(new Object[] {4, "write-after-failover"}); + putRecords(cluster, bucket, nextTail); + records.addAll(nextTail); + assertRows(cluster, bucket, records); + cluster.startTabletServer(oldLeader); + cluster.waitUntilAllReplicaReady(bucket); + assertReplicaOffsets(cluster, bucket, boundary, boundary + 2L); + CompletedSnapshot ordinary = cluster.triggerAndWaitSnapshot(bucket); + assertThat(ordinary.getSnapshotID()).isGreaterThan(external.getSnapshotId()); + assertThat(ordinary.getLogOffset()).isEqualTo(boundary + 2L); + retry( + Duration.ofMinutes(1), + () -> + assertThat( + external.getMetadataFilePath() + .getFileSystem() + .exists(external.getMetadataFilePath())) + .isFalse()); + assertRows(cluster, bucket, records); + assertThat(cluster.getZooKeeperClient().getRemoteLogManifestHandle(bucket)).isEmpty(); + } finally { + cluster.close(); + } + } + + private CompletedSnapshotHandle produceSnapshot( + FlussClusterExtension cluster, + TablePath tablePath, + TableBucket bucket, + List records, + long offset) + throws Exception { + long snapshotId = + new ZkSequenceIDCounter( + cluster.getZooKeeperClient().getCuratorClient(), + ZkData.BucketSnapshotSequenceIdZNode.path(bucket)) + .getAndIncrement(); + FsPath tabletDir = + FlussPaths.remoteKvTabletDir( + new FsPath(cluster.getRemoteDataDir(), "kv"), + PhysicalTablePath.of(tablePath), + bucket); + FsPath location = FlussPaths.remoteKvSnapshotDir(tabletDir, snapshotId); + SnapshotLocation snapshotLocation = + new SnapshotLocation( + location.getFileSystem(), + location, + FlussPaths.remoteKvSharedDir(tabletDir), + 1024); + for (Tuple2 entry : getKeyValuePairs(records)) { + rocksDB.getRocksDb().put(entry.f0, entry.f1); + } + ExecutorService uploader = Executors.newSingleThreadExecutor(); + KvSnapshotHandle files; + try (ResourceGuard guard = new ResourceGuard(); + CloseableRegistry registry = new CloseableRegistry(); + RocksIncrementalSnapshot snapshot = + new RocksIncrementalSnapshot( + new HashMap<>(), + rocksDB.getRocksDb(), + guard, + new KvSnapshotDataUploader(uploader), + rocksDB.getRockDbDir(), + -1L)) { + files = + snapshot.asyncSnapshot( + snapshot.syncPrepareResources(snapshotId), + snapshotId, + new TabletState(offset, (long) records.size(), null), + snapshotLocation) + .get(registry) + .getKvSnapshotHandle(); + } finally { + uploader.shutdownNow(); + } + KvSnapshotFileMetadata metadata = + new KvSnapshotFileMetadata( + bucket, + snapshotId, + location.toString(), + fileMetadata(files.getSharedKvFileHandles()), + fileMetadata(files.getPrivateFileHandles()), + files.getIncrementalSize(), + offset, + (long) records.size(), + Collections.emptyList()); + FsPath metadataPath = CompletedSnapshot.getMetadataFilePath(location); + try (FSDataOutputStream output = + metadataPath + .getFileSystem() + .create(metadataPath, FileSystem.WriteMode.NO_OVERWRITE)) { + output.write(KvSnapshotFileMetadataJsonSerde.toJson(metadata)); + } + return new CompletedSnapshotHandle(snapshotId, metadataPath, offset); + } + + private static List fileMetadata( + List files) { + return files.stream() + .map( + file -> + new KvSnapshotFileMetadata.FileHandle( + file.getKvFileHandle().getFilePath(), + file.getKvFileHandle().getSize(), + file.getLocalPath())) + .collect(Collectors.toList()); + } + + private static void putRecords( + FlussClusterExtension cluster, TableBucket bucket, List records) + throws Exception { + PutKvResponse response = + cluster.newTabletServerClientForNode(cluster.waitAndGetLeader(bucket)) + .putKv( + newPutKvRequest( + bucket.getTableId(), + bucket.getBucket(), + -1, + toKvRecordBatch(records))) + .get(); + assertThat(response.getBucketsRespAt(0).hasErrorCode()).isFalse(); + } + + private static void assertReplicaOffsets( + FlussClusterExtension cluster, TableBucket bucket, long start, long end) + throws Exception { + retry( + Duration.ofMinutes(1), + () -> { + for (int server = 0; server < 2; server++) { + ReplicaManager replicaManager = + cluster.getTabletServerById(server).getReplicaManager(); + assertThat(replicaManager.getReplica(bucket)) + .isInstanceOf(ReplicaManager.OnlineReplica.class); + Replica replica = replicaManager.getReplicaOrException(bucket); + assertThat(replica.getLogTablet().localLogStartOffset()).isEqualTo(start); + assertThat(replica.getLocalLogEndOffset()).isEqualTo(end); + assertThat(replica.getLogHighWatermark()).isEqualTo(end); + } + }); + } + + private static void assertRows( + FlussClusterExtension cluster, TableBucket bucket, List records) + throws Exception { + TabletServerGateway gateway = + cluster.newTabletServerClientForNode(cluster.waitAndGetLeader(bucket)); + for (Tuple2 keyValue : getKeyValuePairs(records)) { + assertLookupResponse( + gateway.lookup( + newLookupRequest( + bucket.getTableId(), bucket.getBucket(), keyValue.f0)) + .get(), + keyValue.f1); + } + } +} diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/LogTabletTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/LogTabletTest.java index 691a7af479c..82971ad9781 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/log/LogTabletTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/log/LogTabletTest.java @@ -402,6 +402,23 @@ void testWriterStateTruncateToWithNoSnapshots() throws Exception { assertThat(lastBatchSeq).isEqualTo(0); } + @Test + void testInitializeEmptyLocalTailProtectsExistingLog() throws Exception { + logTablet.initializeEmptyLocalTail(17L); + logTablet.initializeEmptyLocalTail(17L); + assertThat(logTablet.localLogStartOffset()).isEqualTo(17L); + assertThat(logTablet.localLogEndOffset()).isEqualTo(17L); + assertThat(logTablet.getHighWatermark()).isEqualTo(17L); + assertThatThrownBy(() -> logTablet.initializeEmptyLocalTail(18L)) + .isInstanceOf(IllegalStateException.class); + logTablet.appendAsLeader( + genMemoryLogRecordsByObject(Collections.singletonList(new Object[] {1, "a"}))); + assertThatThrownBy(() -> logTablet.initializeEmptyLocalTail(17L)) + .isInstanceOf(IllegalStateException.class); + assertThat(logTablet.localLogStartOffset()).isEqualTo(17L); + assertThat(logTablet.localLogEndOffset()).isEqualTo(18L); + } + @Test void testWriterStateTruncateFullyAndStartAt() throws Exception { MemoryLogRecords records = 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 b1e98b9d30f..1c43a47beca 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.config.Configuration; +import org.apache.fluss.exception.KvStorageException; import org.apache.fluss.exception.OutOfOrderSequenceException; import org.apache.fluss.fs.FsPath; import org.apache.fluss.metadata.LogFormat; @@ -51,6 +52,7 @@ import org.apache.fluss.server.log.LogAppendInfo; import org.apache.fluss.server.log.LogReadInfo; import org.apache.fluss.server.testutils.KvTestUtils; +import org.apache.fluss.server.zk.ZooKeeperClient; import org.apache.fluss.server.zk.data.LeaderAndIsr; import org.apache.fluss.testutils.DataTestUtils; import org.apache.fluss.testutils.common.ManuallyTriggeredScheduledExecutorService; @@ -111,6 +113,8 @@ import static org.apache.fluss.utils.Preconditions.checkNotNull; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.spy; /** Test for {@link Replica}. */ final class ReplicaTest extends ReplicaTestBase { @@ -669,6 +673,103 @@ void testKvReplicaSnapshot(@TempDir File snapshotKvTabletDir) throws Exception { KvTestUtils.checkSnapshot(completedSnapshot2, expectedKeyValues, expectedLogOffset); } + @Test + void testSnapshotLookupFailurePreventsEmptyRecovery(@TempDir File snapshotDir) + throws Exception { + TestSnapshotContext context = + new TestSnapshotContext(snapshotDir.getPath()) { + @Override + public FunctionWithException + getLatestCompletedSnapshotProvider() { + return bucket -> { + throw new IOException("Snapshot metadata unavailable"); + }; + } + }; + Replica replica = + makeKvReplica( + DATA1_PHYSICAL_TABLE_PATH_PK, + new TableBucket(DATA1_TABLE_ID_PK, 1), + context); + assertThatThrownBy(() -> makeKvReplicaAsLeader(replica, 0)) + .isInstanceOf(KvStorageException.class) + .hasRootCauseMessage("Snapshot metadata unavailable"); + assertThat(replica.getKvTablet()).isNull(); + assertThat(replica.getLocalLogEndOffset()).isZero(); + } + + @Test + void testSnapshotOnlyRecoveryInitializesLogAfterDownload(@TempDir File snapshotDir) + throws Exception { + TableBucket bucket = new TableBucket(DATA1_TABLE_ID_PK, 1); + TestSnapshotContext source = new TestSnapshotContext(snapshotDir.getPath()); + Replica producer = makeKvReplica(DATA1_PHYSICAL_TABLE_PATH_PK, bucket, source); + makeKvReplicaAsLeader(producer); + putRecordsToLeader(producer, genKvRecordBatch(Tuple2.of("k1", new Object[] {1, "a"}))); + source.scheduledExecutorService.triggerAllNonPeriodicTasks(); + CompletedSnapshot snapshot = + source.testKvSnapshotStore.waitUntilSnapshotComplete(bucket, 0); + CompletedSnapshot external = + new CompletedSnapshot( + bucket, + snapshot.getSnapshotID(), + snapshot.getSnapshotLocation(), + snapshot.getKvSnapshotHandle(), + 10_017L, + snapshot.getRowCount(), + snapshot.getAutoIncIDRanges()); + makeKvReplicaAsFollower(producer, 1); + producer.truncateFullyAndStartAt(0L); + AtomicBoolean failDownload = new AtomicBoolean(true); + ZooKeeperClient recoveringZk = spy(zkClient); + doThrow(new IOException("Remote manifest metadata unavailable")) + .doCallRealMethod() + .when(recoveringZk) + .getRemoteLogManifestHandle(bucket); + TestSnapshotContext context = + new TestSnapshotContext(snapshotDir.getPath()) { + @Override + public ZooKeeperClient getZooKeeperClient() { + return recoveringZk; + } + + @Override + public FunctionWithException + getLatestCompletedSnapshotProvider() { + return ignored -> external; + } + + @Override + public KvSnapshotDataDownloader getSnapshotDataDownloader() { + return new KvSnapshotDataDownloader(executorService) { + @Override + public void transferAllDataToDirectory( + KvSnapshotDownloadSpec spec, CloseableRegistry registry) + throws Exception { + if (failDownload.get()) { + throw new IOException("Snapshot download unavailable"); + } + super.transferAllDataToDirectory(spec, registry); + } + }; + } + }; + Replica replica = makeKvReplica(DATA1_PHYSICAL_TABLE_PATH_PK, bucket, context); + assertThatThrownBy(() -> makeKvReplicaAsLeader(replica, 2)) + .isInstanceOf(KvStorageException.class); + assertThat(replica.isLeader()).isFalse(); + assertThat(replica.getLocalLogEndOffset()).isZero(); + assertThat(replica.getLogHighWatermark()).isZero(); + failDownload.set(false); + makeKvReplicaAsLeader(replica, 3); + assertThat(replica.getLocalLogEndOffset()).isEqualTo(10_017L); + assertThat(replica.getLogHighWatermark()).isEqualTo(10_017L); + assertThat(replica.getLeaderEndOffsetSnapshot()).isEqualTo(10_017L); + verifyGetKeyValues( + replica.getKvTablet(), + getKeyValuePairs(genKvRecords(Tuple2.of("k1", new Object[] {1, "a"})))); + } + @Test void testSnapshotUseLatestLeaderEpoch(@TempDir File snapshotKvTabletDir) throws Exception { TableBucket tableBucket = new TableBucket(DATA1_TABLE_ID_PK, 1);