From c9273951d917eae0867327dce60ab1b93628af75 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=99=BD=E9=B5=BA?= Date: Mon, 7 Sep 2026 16:08:17 +0800 Subject: [PATCH 1/5] [server] Register external KV snapshots and restore replicas Share KV snapshot file metadata across modules and register externally produced target-bucket snapshots with coordinator fencing and idempotent retries. Adopt confirmed snapshots into retention and protect persistent references when registration responses are lost. Initialize the local log boundary from snapshot-only history through ordinary replica role notifications. Cover external snapshot production, registration, recovery, repeated notifications, online writes, failover and subsequent snapshot retention. --- .../metadata/KvSnapshotFileMetadata.java | 260 ++++++++++++++ .../KvSnapshotFileMetadataJsonSerde.java | 181 ++++++++++ .../KvSnapshotFileMetadataJsonSerdeTest.java | 67 ++++ .../CompletedSnapshotStoreManager.java | 73 +++- .../snapshot/CompletedSnapshotJsonSerde.java | 270 ++++---------- .../kv/snapshot/CompletedSnapshotStore.java | 140 +++++--- .../apache/fluss/server/log/LogManager.java | 16 + .../fluss/server/replica/ReplicaManager.java | 20 ++ .../fluss/server/zk/ZooKeeperClient.java | 38 ++ .../CompletedSnapshotStoreManagerTest.java | 161 +++++++++ .../coordinator/ExternalKvSnapshotITCase.java | 335 ++++++++++++++++++ 11 files changed, 1311 insertions(+), 250 deletions(-) create mode 100644 fluss-common/src/main/java/org/apache/fluss/metadata/KvSnapshotFileMetadata.java create mode 100644 fluss-common/src/main/java/org/apache/fluss/metadata/KvSnapshotFileMetadataJsonSerde.java create mode 100644 fluss-common/src/test/java/org/apache/fluss/metadata/KvSnapshotFileMetadataJsonSerdeTest.java create mode 100644 fluss-server/src/test/java/org/apache/fluss/server/coordinator/ExternalKvSnapshotITCase.java 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..375d6833d3e 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 replaces local history with an empty active tail whose first offset is {@code E}. */ + 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.truncateFullyAndStartAt(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/replica/ReplicaManager.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java index 142facb80f5..cf607a10bb2 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java @@ -121,6 +121,7 @@ import org.apache.fluss.server.storage.LocalDiskManager; import org.apache.fluss.server.utils.FatalErrorHandler; import org.apache.fluss.server.zk.ZooKeeperClient; +import org.apache.fluss.server.zk.data.BucketSnapshot; import org.apache.fluss.server.zk.data.LeaderAndIsr; import org.apache.fluss.server.zk.data.lake.LakeTableSnapshot; import org.apache.fluss.utils.ByteArraySlice; @@ -1436,6 +1437,7 @@ private void makeLeaders( TableBucket tb = data.getTableBucket(); try { Replica replica = getReplicaOrException(tb); + initializeSnapshotOnlyLocalTail(replica); // register replica to remote log manager first. remoteLogManager.registerReplica(replica); @@ -1553,6 +1555,24 @@ private void makeFollowers( addFetcherForReplicas(replicasBecomeFollower, result); } + /** Initializes a new local log from a snapshot when no remote log prefix exists. */ + private void initializeSnapshotOnlyLocalTail(Replica replica) throws Exception { + if (!replica.isKvTable() + || replica.isHistoricalPartition() + || replica.getLocalLogEndOffset() != 0L + || replica.getLogHighWatermark() != 0L) { + return; + } + Optional snapshot = + zkClient.getTableBucketLatestSnapshot(replica.getTableBucket()); + if (snapshot.isPresent() + && snapshot.get().getLogOffset() > 0L + && !zkClient.getRemoteLogManifestHandle(replica.getTableBucket()).isPresent()) { + logManager.initializeEmptyLocalTail( + replica.getTableBucket(), snapshot.get().getLogOffset()); + } + } + private void addFetcherForReplicas( List replicas, Map result) { Map bucketAndStatus = new HashMap<>(); 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); + } + } +} From 958e7f2a2f4d1f8f4a80d068ec1f0aa651110efb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=99=BD=E9=B5=BA?= Date: Tue, 8 Sep 2026 10:06:30 +0800 Subject: [PATCH 2/5] [server] Align snapshot recovery and local log initialization Use the selected KV snapshot to initialize an empty local log after its files have been loaded, and capture the leader log boundary after recovery. Reject conflicting log initialization and propagate snapshot lookup errors. Release partially recovered KV state before retrying, and publish leader identity only after recovery succeeds so failed elections can be retried. Cover snapshot lookup and download failures, metadata retry after loading, and protection of existing local log records. --- .../apache/fluss/server/log/LogManager.java | 4 +- .../apache/fluss/server/log/LogTablet.java | 20 ++++ .../apache/fluss/server/replica/Replica.java | 44 +++++-- .../fluss/server/replica/ReplicaManager.java | 20 ---- .../fluss/server/log/LogTabletTest.java | 17 +++ .../fluss/server/replica/ReplicaTest.java | 110 ++++++++++++++++++ 6 files changed, 184 insertions(+), 31 deletions(-) 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 375d6833d3e..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,13 +421,13 @@ public void truncateFullyAndStartAt(TableBucket tableBucket, long newOffset) { } } - /** Durably replaces local history with an empty active tail whose first offset is {@code E}. */ + /** 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.truncateFullyAndStartAt(endOffset); + logTablet.initializeEmptyLocalTail(endOffset); try { logTablet.flush(true); } catch (IOException e) { 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..b5790269cee 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 @@ -471,8 +471,15 @@ public void makeLeader(NotifyLeaderAndIsrData data) throws IOException { int requestLeaderEpoch = data.getLeaderEpoch(); if (requestLeaderEpoch > leaderEpoch) { boolean isNewLeader = !isLeader(); + int previousLeaderEpoch = leaderEpoch; leaderEpoch = requestLeaderEpoch; - onBecomeNewLeader(); + leaderReplicaIdOpt.set(null); + try { + onBecomeNewLeader(); + } catch (RuntimeException e) { + leaderEpoch = previousLeaderEpoch; + throw e; + } leaderReplicaIdOpt.set(localTabletServerId); // onBecomeNewLeader may recover a KV snapshot, so start the ISR lag // grace period after it completes. @@ -611,8 +618,6 @@ private void onBecomeNewLeader() { // Clear standby flag — a leader is never a standby replica. isStandbyReplica = false; - updateLeaderEndOffsetSnapshot(); - if (isDataLakeEnabled()) { registerLakeTieringMetrics(); } @@ -625,6 +630,8 @@ private void onBecomeNewLeader() { // now, we can create a new kv tablet createKv(); } + + updateLeaderEndOffsetSnapshot(); } private void registerLakeTieringMetrics() { @@ -748,6 +755,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 +886,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 +1017,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/replica/ReplicaManager.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java index cf607a10bb2..142facb80f5 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java @@ -121,7 +121,6 @@ import org.apache.fluss.server.storage.LocalDiskManager; import org.apache.fluss.server.utils.FatalErrorHandler; import org.apache.fluss.server.zk.ZooKeeperClient; -import org.apache.fluss.server.zk.data.BucketSnapshot; import org.apache.fluss.server.zk.data.LeaderAndIsr; import org.apache.fluss.server.zk.data.lake.LakeTableSnapshot; import org.apache.fluss.utils.ByteArraySlice; @@ -1437,7 +1436,6 @@ private void makeLeaders( TableBucket tb = data.getTableBucket(); try { Replica replica = getReplicaOrException(tb); - initializeSnapshotOnlyLocalTail(replica); // register replica to remote log manager first. remoteLogManager.registerReplica(replica); @@ -1555,24 +1553,6 @@ private void makeFollowers( addFetcherForReplicas(replicasBecomeFollower, result); } - /** Initializes a new local log from a snapshot when no remote log prefix exists. */ - private void initializeSnapshotOnlyLocalTail(Replica replica) throws Exception { - if (!replica.isKvTable() - || replica.isHistoricalPartition() - || replica.getLocalLogEndOffset() != 0L - || replica.getLogHighWatermark() != 0L) { - return; - } - Optional snapshot = - zkClient.getTableBucketLatestSnapshot(replica.getTableBucket()); - if (snapshot.isPresent() - && snapshot.get().getLogOffset() > 0L - && !zkClient.getRemoteLogManifestHandle(replica.getTableBucket()).isPresent()) { - logManager.initializeEmptyLocalTail( - replica.getTableBucket(), snapshot.get().getLogOffset()); - } - } - private void addFetcherForReplicas( List replicas, Map result) { Map bucketAndStatus = new HashMap<>(); 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..c018cf39994 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,112 @@ void testKvReplicaSnapshot(@TempDir File snapshotKvTabletDir) throws Exception { KvTestUtils.checkSnapshot(completedSnapshot2, expectedKeyValues, expectedLogOffset); } + @Test + void testSnapshotLookupFailureAllowsSameLeaderEpochRetry(@TempDir File snapshotDir) + throws Exception { + AtomicBoolean failLookup = new AtomicBoolean(false); + TestSnapshotContext context = + new TestSnapshotContext(snapshotDir.getPath()) { + @Override + public FunctionWithException + getLatestCompletedSnapshotProvider() { + return bucket -> { + if (failLookup.get()) { + throw new IOException("Snapshot metadata unavailable"); + } + return null; + }; + } + }; + Replica replica = + makeKvReplica( + DATA1_PHYSICAL_TABLE_PATH_PK, + new TableBucket(DATA1_TABLE_ID_PK, 1), + context); + makeKvReplicaAsLeader(replica, 0); + failLookup.set(true); + assertThatThrownBy(() -> makeKvReplicaAsLeader(replica, 1)) + .isInstanceOf(KvStorageException.class); + assertThat(replica.isLeader()).isFalse(); + assertThat(replica.getLocalLogEndOffset()).isZero(); + failLookup.set(false); + makeKvReplicaAsLeader(replica, 1); + assertThat(replica.isLeader()).isTrue(); + assertThat(replica.getKvTablet()).isNotNull(); + } + + @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, 2); + 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); From 52d5f632307550ab74da101a34a19f9a7d94800a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=99=BD=E9=B5=BA?= Date: Tue, 8 Sep 2026 14:46:19 +0800 Subject: [PATCH 3/5] [server] Preserve leader epoch semantics during snapshot recovery Keep the existing leader and follower transition behavior while coordinating KV snapshot recovery with local log initialization. Test snapshot lookup failures independently of role transition retries, and recover from a failed download under a subsequent leader epoch. --- .../apache/fluss/server/replica/Replica.java | 9 +------- .../fluss/server/replica/ReplicaTest.java | 23 ++++++------------- 2 files changed, 8 insertions(+), 24 deletions(-) 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 b5790269cee..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 @@ -471,15 +471,8 @@ public void makeLeader(NotifyLeaderAndIsrData data) throws IOException { int requestLeaderEpoch = data.getLeaderEpoch(); if (requestLeaderEpoch > leaderEpoch) { boolean isNewLeader = !isLeader(); - int previousLeaderEpoch = leaderEpoch; leaderEpoch = requestLeaderEpoch; - leaderReplicaIdOpt.set(null); - try { - onBecomeNewLeader(); - } catch (RuntimeException e) { - leaderEpoch = previousLeaderEpoch; - throw e; - } + onBecomeNewLeader(); leaderReplicaIdOpt.set(localTabletServerId); // onBecomeNewLeader may recover a KV snapshot, so start the ISR lag // grace period after it completes. 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 c018cf39994..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 @@ -674,19 +674,15 @@ void testKvReplicaSnapshot(@TempDir File snapshotKvTabletDir) throws Exception { } @Test - void testSnapshotLookupFailureAllowsSameLeaderEpochRetry(@TempDir File snapshotDir) + void testSnapshotLookupFailurePreventsEmptyRecovery(@TempDir File snapshotDir) throws Exception { - AtomicBoolean failLookup = new AtomicBoolean(false); TestSnapshotContext context = new TestSnapshotContext(snapshotDir.getPath()) { @Override public FunctionWithException getLatestCompletedSnapshotProvider() { return bucket -> { - if (failLookup.get()) { - throw new IOException("Snapshot metadata unavailable"); - } - return null; + throw new IOException("Snapshot metadata unavailable"); }; } }; @@ -695,16 +691,11 @@ void testSnapshotLookupFailureAllowsSameLeaderEpochRetry(@TempDir File snapshotD DATA1_PHYSICAL_TABLE_PATH_PK, new TableBucket(DATA1_TABLE_ID_PK, 1), context); - makeKvReplicaAsLeader(replica, 0); - failLookup.set(true); - assertThatThrownBy(() -> makeKvReplicaAsLeader(replica, 1)) - .isInstanceOf(KvStorageException.class); - assertThat(replica.isLeader()).isFalse(); + assertThatThrownBy(() -> makeKvReplicaAsLeader(replica, 0)) + .isInstanceOf(KvStorageException.class) + .hasRootCauseMessage("Snapshot metadata unavailable"); + assertThat(replica.getKvTablet()).isNull(); assertThat(replica.getLocalLogEndOffset()).isZero(); - failLookup.set(false); - makeKvReplicaAsLeader(replica, 1); - assertThat(replica.isLeader()).isTrue(); - assertThat(replica.getKvTablet()).isNotNull(); } @Test @@ -770,7 +761,7 @@ public void transferAllDataToDirectory( assertThat(replica.getLocalLogEndOffset()).isZero(); assertThat(replica.getLogHighWatermark()).isZero(); failDownload.set(false); - makeKvReplicaAsLeader(replica, 2); + makeKvReplicaAsLeader(replica, 3); assertThat(replica.getLocalLogEndOffset()).isEqualTo(10_017L); assertThat(replica.getLogHighWatermark()).isEqualTo(10_017L); assertThat(replica.getLeaderEndOffsetSnapshot()).isEqualTo(10_017L); From 7a1bd7efc14a32048dd221f31bbff0c268bbb813 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=99=BD=E9=B5=BA?= Date: Thu, 10 Sep 2026 11:38:12 +0800 Subject: [PATCH 4/5] [server] Encapsulate external snapshot registration in the store Serialize external snapshot identity and ordering checks, persistent handle confirmation and retention updates in CompletedSnapshotStore. Delegate coordinator fencing and idempotent persistence to the handle store while preserving ordinary snapshot writes and cleanup behavior. Describe the snapshot counter check as an allocation bound; reserving the ID remains the caller's responsibility. --- .../CompletedSnapshotStoreManager.java | 20 +------ .../CompletedSnapshotHandleStore.java | 17 ++++++ .../kv/snapshot/CompletedSnapshotStore.java | 57 ++++++++++--------- ...ZooKeeperCompletedSnapshotHandleStore.java | 16 ++++++ .../CompletedSnapshotStoreManagerTest.java | 11 +++- .../CoordinatorServiceOrphanRpcsITCase.java | 9 +++ .../TestCompletedSnapshotHandleStore.java | 8 +++ 7 files changed, 94 insertions(+), 44 deletions(-) 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 8b07baca22e..cd4c0cd2733 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 @@ -33,7 +33,6 @@ 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; @@ -203,22 +202,9 @@ public void registerExternalSnapshot( .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); + "External snapshot ID must be below the target bucket counter."); + getOrCreateCompletedSnapshotStore(tablePath, tableBucket) + .registerExternalSnapshot(snapshot, coordinatorZkVersion); } public void removeCompletedSnapshotStoreByTableBuckets(Set tableBuckets) { diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotHandleStore.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotHandleStore.java index 31f2424e3c4..7a1447172d3 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotHandleStore.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotHandleStore.java @@ -42,6 +42,23 @@ void add( CompletedSnapshotHandle completedSnapshotHandle) throws Exception; + /** + * Confirms an existing immutable snapshot handle under the active coordinator epoch. + * + *

An identical registration can be retried; a conflicting handle is rejected. Failures, + * including uncertain results, must leave the snapshot metadata file intact. + * + * @param tableBucket the target bucket + * @param snapshotHandle the handle of the already written snapshot metadata + * @param coordinatorZkVersion the expected coordinator epoch node version + * @throws Exception if the handle cannot be confirmed + */ + void registerExternal( + TableBucket tableBucket, + CompletedSnapshotHandle snapshotHandle, + int coordinatorZkVersion) + throws Exception; + /** * Remove the snapshot handle for the given snapshot id of the given table bucket. * 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 4c0d5ffe030..88521e6ace6 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 @@ -42,8 +42,8 @@ import java.util.concurrent.Executor; import java.util.concurrent.locks.ReentrantLock; +import static org.apache.fluss.utils.Preconditions.checkArgument; 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 @@ -118,43 +118,48 @@ public void add(final CompletedSnapshot completedSnapshot) throws Exception { } /** - * Adopts a snapshot whose persistent snapshot node has already been confirmed. + * Registers an immutable external snapshot and includes it in ordinary snapshot retention. * - *

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. + *

Identity and ordering checks, persistent handle confirmation and retention updates are + * serialized with ordinary snapshot additions. Even an already retained snapshot must have its + * handle confirmed under the supplied coordinator epoch before a retry succeeds. */ - public void adoptAfterNodeConfirmed(final CompletedSnapshot snapshot) throws Exception { + public void registerExternalSnapshot(final CompletedSnapshot snapshot, int coordinatorZkVersion) + 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 existing = stillInUseSnapshots.get(snapshot.getSnapshotID()); + for (CompletedSnapshot retained : completedSnapshots) { + if (retained.getSnapshotID() == snapshot.getSnapshotID()) { + existing = retained; + break; } } - CompletedSnapshot stillInUse = - stillInUseSnapshots.get(snapshot.getSnapshotID()); - if (stillInUse != null) { - checkState( - stillInUse.equals(snapshot), + if (existing != null) { + checkArgument( + existing.equals(snapshot), "Conflicting snapshot identity for %s snapshot %s.", snapshot.getTableBucket(), snapshot.getSnapshotID()); - return; + } else { + checkArgument( + completedSnapshots.isEmpty() + || completedSnapshots.peekLast().getSnapshotID() + < snapshot.getSnapshotID(), + "Cannot register an older snapshot that has already been subsumed."); + } + completedSnapshotHandleStore.registerExternal( + snapshot.getTableBucket(), + new CompletedSnapshotHandle( + snapshot.getSnapshotID(), + snapshot.getMetadataFilePath(), + snapshot.getLogOffset()), + coordinatorZkVersion); + if (existing == null) { + adoptConfirmedSnapshot(snapshot, snapshotsCleaner, () -> {}); } - checkState( - completedSnapshots.isEmpty() - || completedSnapshots.peekLast().getSnapshotID() - < snapshot.getSnapshotID(), - "Cannot adopt an older snapshot %s.", - snapshot.getSnapshotID()); - adoptConfirmedSnapshot(snapshot, snapshotsCleaner, () -> {}); }); } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/ZooKeeperCompletedSnapshotHandleStore.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/ZooKeeperCompletedSnapshotHandleStore.java index 9753209ad64..593842c5f56 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/ZooKeeperCompletedSnapshotHandleStore.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/ZooKeeperCompletedSnapshotHandleStore.java @@ -85,6 +85,22 @@ public void add( } } + @Override + public void registerExternal( + TableBucket tableBucket, + CompletedSnapshotHandle snapshotHandle, + int coordinatorZkVersion) + throws Exception { + checkNotNull(snapshotHandle, "completed snapshot handle"); + client.registerExternalTableBucketSnapshot( + tableBucket, + new BucketSnapshot( + snapshotHandle.getSnapshotId(), + snapshotHandle.getLogOffset(), + snapshotHandle.getMetadataFilePath().toString()), + coordinatorZkVersion); + } + @Override public void remove(TableBucket tableBucket, long snapshotId) throws Exception { // TODO: it may bring concurrent delete operations when lost leadership and a new leadership 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 86418492691..b578827d0e3 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 @@ -343,7 +343,7 @@ DATA1_TABLE_PATH, new TableBucket(99, 2), handle, epoch)) manager.registerExternalSnapshot( DATA1_TABLE_PATH, bucket, handle, epoch)) .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining("reserved"); + .hasMessageContaining("below the target bucket counter"); new ZkSequenceIDCounter( zookeeperClient.getCuratorClient(), ZkData.BucketSnapshotSequenceIdZNode.path(bucket)) @@ -533,6 +533,15 @@ public void add( .put(snapshotId, completedSnapshotHandle); } + @Override + public void registerExternal( + TableBucket tableBucket, + CompletedSnapshotHandle snapshotHandle, + int coordinatorZkVersion) { + throw new UnsupportedOperationException( + "External snapshot registration is not supported."); + } + @Override public void remove(TableBucket tableBucket, long snapshotId) throws Exception { snapshotHandleMap.computeIfAbsent(tableBucket, k -> new HashMap<>()).remove(snapshotId); 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 00e6ea5e498..cd47e909309 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 @@ -236,6 +236,15 @@ public void add( long snapshotId, CompletedSnapshotHandle completedSnapshotHandle) {} + @Override + public void registerExternal( + TableBucket tableBucket, + CompletedSnapshotHandle snapshotHandle, + int coordinatorZkVersion) { + throw new UnsupportedOperationException( + "External snapshot registration is not supported."); + } + @Override public void remove(TableBucket tableBucket, long snapshotId) {} diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/TestCompletedSnapshotHandleStore.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/TestCompletedSnapshotHandleStore.java index def7adcf3ee..a1a979ca64e 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/TestCompletedSnapshotHandleStore.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/TestCompletedSnapshotHandleStore.java @@ -47,6 +47,14 @@ public void add( addFunction.apply(snapshotHandle); } + @Override + public void registerExternal( + TableBucket tableBucket, + CompletedSnapshotHandle snapshotHandle, + int coordinatorZkVersion) { + throw new UnsupportedOperationException("External snapshot registration is not supported."); + } + @Override public void remove(TableBucket tableBucket, long snapshotId) throws Exception {} From 097aacc7c26560d3ac7bc197a17feb91b6e5c806 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=99=BD=E9=B5=BA?= Date: Thu, 10 Sep 2026 12:10:34 +0800 Subject: [PATCH 5/5] [server] Resolve snapshot metadata before leader activation Resolve the recovery snapshot and log prerequisites before changing role state or discarding the current KV. Reuse that selection for the first recovery attempt while preserving retries through older healthy snapshots. Avoid remote manifest lookups when the local log already covers the snapshot. Cover metadata lookup failures with a live KV and recovery after metadata becomes available. Clarify that obsolete snapshot cleanup may be deferred. --- .../kv/snapshot/CompletedSnapshotStore.java | 3 +- .../apache/fluss/server/replica/Replica.java | 76 ++++++++++++++----- .../fluss/server/replica/ReplicaTest.java | 63 ++++++++++++--- 3 files changed, 110 insertions(+), 32 deletions(-) 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 88521e6ace6..1ede85ba5c4 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 @@ -122,7 +122,8 @@ public void add(final CompletedSnapshot completedSnapshot) throws Exception { * *

Identity and ordering checks, persistent handle confirmation and retention updates are * serialized with ordinary snapshot additions. Even an already retained snapshot must have its - * handle confirmed under the supplied coordinator epoch before a retry succeeds. + * handle confirmed under the supplied coordinator epoch before a retry succeeds. Obsolete + * snapshot cleanup follows ordinary retention semantics and may be deferred on failure. */ public void registerExternalSnapshot(final CompletedSnapshot snapshot, int coordinatorZkVersion) throws Exception { 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 b9748350096..afe063646bc 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 @@ -457,6 +457,12 @@ public void makeLeader(NotifyLeaderAndIsrData data) throws IOException { int requestBucketEpoch = data.getBucketEpoch(); validateBucketEpoch(requestBucketEpoch); + // Resolve recovery metadata before changing role state or dropping KV. + KvRecovery recovery = + data.getLeaderEpoch() > leaderEpoch && isKvTable() + ? prepareKvRecovery() + : null; + coordinatorEpoch = data.getCoordinatorEpoch(); long currentTimeMs = clock.milliseconds(); @@ -472,7 +478,7 @@ public void makeLeader(NotifyLeaderAndIsrData data) throws IOException { if (requestLeaderEpoch > leaderEpoch) { boolean isNewLeader = !isLeader(); leaderEpoch = requestLeaderEpoch; - onBecomeNewLeader(); + onBecomeNewLeader(recovery); leaderReplicaIdOpt.set(localTabletServerId); // onBecomeNewLeader may recover a KV snapshot, so start the ISR lag // grace period after it completes. @@ -607,7 +613,7 @@ public LogOffsetSnapshot fetchOffsetSnapshot(boolean fetchOnlyFromLeader) throws // ------------------------------------------------------------------------------------------- - private void onBecomeNewLeader() { + private void onBecomeNewLeader(@Nullable KvRecovery recovery) { // Clear standby flag — a leader is never a standby replica. isStandbyReplica = false; @@ -621,7 +627,7 @@ private void onBecomeNewLeader() { // if exist. Otherwise, it'll use still the old kv tablet which will cause data loss dropKv(); // now, we can create a new kv tablet - createKv(); + createKv(checkNotNull(recovery)); } updateLeaderEndOffsetSnapshot(); @@ -725,7 +731,7 @@ private void logTableConfigChanges(TableInfo oldTableInfo, TableInfo newTableInf } } - private void createKv() { + private void createKv(KvRecovery recovery) { try { // create a closeable registry for the closable related to kv closeableRegistryForKv = new CloseableRegistry(); @@ -743,7 +749,10 @@ private void createKv() { Exception lastError = null; for (int i = 1; i <= INIT_KV_TABLET_MAX_RETRY_TIMES; i++) { try { - snapshotUsed = initKvTablet(); + if (i > 1) { + recovery = prepareKvRecovery(); + } + snapshotUsed = initKvTablet(recovery); lastError = null; break; } catch (Exception e) { @@ -831,7 +840,7 @@ private void onKvFlushComplete() { * * @return the snapshot used to init kv tablet, empty if no any snapshot. */ - private Optional initKvTablet() { + private Optional initKvTablet(KvRecovery recovery) { checkNotNull(kvManager); TableConfig tableConfig = getTableConfig(); long startTime = clock.milliseconds(); @@ -855,8 +864,7 @@ private Optional initKvTablet() { long restoreStartOffset = isHistoricalPartition() ? historicalRecoveryStartOffset() : 0; // Lake is the durable base for local historical KV state. Historical replicas therefore // never restore a normal KV snapshot, even if one exists from older code. - Optional optCompletedSnapshot = - isHistoricalPartition() ? Optional.empty() : getLatestSnapshot(tableBucket); + Optional optCompletedSnapshot = recovery.snapshot; try { Long rowCount; AutoIncIDRange autoIncIDRange; @@ -879,19 +887,8 @@ 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); + if (recovery.initializeLocalLog) { + logManager.initializeEmptyLocalTail(tableBucket, restoreStartOffset); } rowCount = supportsExactRowCount(tableConfig) ? completedSnapshot.getRowCount() : null; @@ -1005,6 +1002,43 @@ private static boolean isSnapshotDataNotExists( return false; } + private KvRecovery prepareKvRecovery() { + Optional snapshot = + isHistoricalPartition() ? Optional.empty() : getLatestSnapshot(tableBucket); + boolean initializeLocalLog = false; + if (snapshot.isPresent() && snapshot.get().getLogOffset() > logTablet.localLogEndOffset()) { + try { + if (!snapshotContext + .getZooKeeperClient() + .getRemoteLogManifestHandle(tableBucket) + .isPresent()) { + checkState( + logTablet.localLogEndOffset() == 0L + && logTablet.getHighWatermark() == 0L, + "Local log ends before snapshot offset %s for %s without remote logs.", + snapshot.get().getLogOffset(), + tableBucket); + initializeLocalLog = true; + } + } catch (Exception e) { + throw new KvStorageException( + "Failed to prepare snapshot recovery for " + tableBucket + '.', e); + } + } + return new KvRecovery(snapshot, initializeLocalLog); + } + + /** Snapshot selection and log initialization required for one KV recovery attempt. */ + private static final class KvRecovery { + private final Optional snapshot; + private final boolean initializeLocalLog; + + private KvRecovery(Optional snapshot, boolean initializeLocalLog) { + this.snapshot = snapshot; + this.initializeLocalLog = initializeLocalLog; + } + } + private Optional getLatestSnapshot(TableBucket tableBucket) { try { return Optional.ofNullable( 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 1c43a47beca..1e151f18a1f 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 @@ -698,6 +698,59 @@ void testSnapshotLookupFailurePreventsEmptyRecovery(@TempDir File snapshotDir) assertThat(replica.getLocalLogEndOffset()).isZero(); } + @Test + void testRecoveryMetadataFailurePreservesLiveKv(@TempDir File snapshotDir) throws Exception { + TableBucket bucket = new TableBucket(DATA1_TABLE_ID_PK, 1); + AtomicBoolean failLookup = new AtomicBoolean(false); + ZooKeeperClient failingRemoteMetadata = spy(zkClient); + doThrow(new IOException("Remote manifest metadata unavailable")) + .when(failingRemoteMetadata) + .getRemoteLogManifestHandle(bucket); + TestSnapshotContext context = + new TestSnapshotContext(snapshotDir.getPath()) { + @Override + public ZooKeeperClient getZooKeeperClient() { + return failingRemoteMetadata; + } + + @Override + public FunctionWithException + getLatestCompletedSnapshotProvider() { + FunctionWithException provider = + super.getLatestCompletedSnapshotProvider(); + return target -> { + if (failLookup.get()) { + throw new IOException("Snapshot metadata unavailable"); + } + return provider.apply(target); + }; + } + }; + Replica replica = makeKvReplica(DATA1_PHYSICAL_TABLE_PATH_PK, bucket, context); + makeKvReplicaAsLeader(replica, 0); + putRecordsToLeader(replica, genKvRecordBatch(Tuple2.of("k1", new Object[] {1, "a"}))); + context.scheduledExecutorService.triggerAllNonPeriodicTasks(); + context.testKvSnapshotStore.waitUntilSnapshotComplete(bucket, 0); + KvTablet liveKv = replica.getKvTablet(); + long logEndOffset = replica.getLocalLogEndOffset(); + failLookup.set(true); + assertThatThrownBy(() -> makeKvReplicaAsLeader(replica, 1)) + .isInstanceOf(KvStorageException.class); + assertThat(replica.getLeaderEpoch()).isZero(); + assertThat(replica.isLeader()).isTrue(); + assertThat(replica.getKvTablet()).isSameAs(liveKv); + assertThat(replica.getLocalLogEndOffset()).isEqualTo(logEndOffset); + verifyGetKeyValues( + replica.getKvTablet(), + getKeyValuePairs(genKvRecords(Tuple2.of("k1", new Object[] {1, "a"})))); + failLookup.set(false); + makeKvReplicaAsLeader(replica, 1); + assertThat(replica.getLeaderEpoch()).isEqualTo(1); + verifyGetKeyValues( + replica.getKvTablet(), + getKeyValuePairs(genKvRecords(Tuple2.of("k1", new Object[] {1, "a"})))); + } + @Test void testSnapshotOnlyRecoveryInitializesLogAfterDownload(@TempDir File snapshotDir) throws Exception { @@ -721,18 +774,8 @@ void testSnapshotOnlyRecoveryInitializesLogAfterDownload(@TempDir File snapshotD 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() {