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..1762662d561 --- /dev/null +++ b/fluss-common/src/main/java/org/apache/fluss/metadata/KvSnapshotFileMetadata.java @@ -0,0 +1,279 @@ +/* + * 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.checkArgument; +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; + private final @Nullable String sha256; + + /** Creates an immutable file reference without a content digest. */ + public FileHandle(String path, long size, String localPath) { + this(path, size, localPath, null); + } + + /** Creates an immutable file reference with an optional SHA-256 content digest. */ + public FileHandle(String path, long size, String localPath, @Nullable String sha256) { + this.path = checkNotNull(path, "File path must not be null."); + checkArgument(size >= 0L, "File size must not be negative."); + this.size = size; + this.localPath = checkNotNull(localPath, "File local path must not be null."); + checkArgument( + sha256 == null || sha256.matches("[0-9a-f]{64}"), + "File SHA-256 must contain exactly 64 lowercase hexadecimal characters."); + this.sha256 = sha256; + } + + /** 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; + } + + /** Returns the lowercase hexadecimal SHA-256 when present. */ + @Nullable + public String getSha256() { + return sha256; + } + + @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) + && Objects.equals(sha256, that.sha256); + } + + @Override + public int hashCode() { + return Objects.hash(path, size, localPath, sha256); + } + } + + /** 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..cfd48ada7bc --- /dev/null +++ b/fluss-common/src/main/java/org/apache/fluss/metadata/KvSnapshotFileMetadataJsonSerde.java @@ -0,0 +1,197 @@ +/* + * 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; + +import static org.apache.fluss.utils.Preconditions.checkArgument; + +/** 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_SHA256 = "sha256"; + 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 versionNode = node.get(VERSION_KEY); + checkArgument( + versionNode != null + && versionNode.isIntegralNumber() + && versionNode.canConvertToInt() + && versionNode.intValue() == VERSION, + "Unsupported KV snapshot metadata version."); + 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()); + if (fileHandle.getSha256() != null) { + generator.writeStringField(KV_FILE_SHA256, fileHandle.getSha256()); + } + 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(), + handleNode.has(KV_FILE_SHA256) + ? handleNode.get(KV_FILE_SHA256).asText() + : null)); + } + 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..1387e867fba --- /dev/null +++ b/fluss-common/src/test/java/org/apache/fluss/metadata/KvSnapshotFileMetadataJsonSerdeTest.java @@ -0,0 +1,133 @@ +/* + * 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 org.junit.jupiter.api.Test; + +import java.nio.charset.StandardCharsets; +import java.util.Collections; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** 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); + } + + @Test + void testOptionalFileDigestRoundTrip() { + String sha256 = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; + KvSnapshotFileMetadata metadata = + new KvSnapshotFileMetadata( + new TableBucket(1, 1), + 2, + "oss://bucket/snapshot", + Collections.emptyList(), + Collections.singletonList( + new KvSnapshotFileMetadata.FileHandle( + "oss://bucket/snapshot/private.sst", + 10, + "private.sst", + sha256)), + 10, + 20, + null, + null); + + byte[] json = KvSnapshotFileMetadataJsonSerde.toJson(metadata); + + assertThat(new String(json, StandardCharsets.UTF_8)) + .contains("\"sha256\":\"" + sha256 + "\""); + assertThat(KvSnapshotFileMetadataJsonSerde.fromJson(json)).isEqualTo(metadata); + } + + @Test + void testRejectMissingVersion() { + byte[] json = GOLDEN_JSON.replace("{\"version\":1,", "{").getBytes(StandardCharsets.UTF_8); + + assertThatThrownBy(() -> KvSnapshotFileMetadataJsonSerde.fromJson(json)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("version"); + } + + @Test + void testRejectUnsupportedVersion() { + byte[] json = + GOLDEN_JSON + .replace("\"version\":1", "\"version\":2") + .getBytes(StandardCharsets.UTF_8); + + assertThatThrownBy(() -> KvSnapshotFileMetadataJsonSerde.fromJson(json)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("version"); + } + + @Test + void testRejectVersionOutsideIntegerRange() { + byte[] json = + GOLDEN_JSON + .replace("\"version\":1", "\"version\":4294967297") + .getBytes(StandardCharsets.UTF_8); + + assertThatThrownBy(() -> KvSnapshotFileMetadataJsonSerde.fromJson(json)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("version"); + } + + @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-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/action/orphan/OrphanFilesCleanITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/action/orphan/OrphanFilesCleanITCase.java index 39583965c54..84e7c192ba4 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/action/orphan/OrphanFilesCleanITCase.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/action/orphan/OrphanFilesCleanITCase.java @@ -582,7 +582,12 @@ void kvUnitFailureDoesNotBlockLogCleanup() throws Exception { TableInfo tableInfo = admin.getTableInfo(tablePath).get(); TableBucket tableBucket = new TableBucket(tableInfo.getTableId(), 0); - // Seed a valid KV snapshot in ZK so listBucketSnapshots returns a child to decode. + // Seed a log manifest + active segment while the replica still has ordinary empty state. + // Updating the manifest can trigger a leader notification, which must complete before the + // intentionally synthetic snapshot metadata below is installed for the cleaner test. + Path activeLogSegment = seedActiveBucketManifest(tablePath); + + // Seed a KV snapshot in ZK so listBucketSnapshots returns a child to decode. FsPath remoteKvTabletDir = FlussPaths.remoteKvTabletDir( new FsPath(remoteDataRoot().resolve("kv").toUri().toString()), @@ -591,10 +596,6 @@ void kvUnitFailureDoesNotBlockLogCleanup() throws Exception { long activeSnapshotId = 1L; seedKvSnapshots(tableBucket, remoteKvTabletDir, new long[] {activeSnapshotId}); - // Seed a log manifest + active segment so the log bucket reaches RESOLVED in the - // active-file cleanup. - Path activeLogSegment = seedActiveBucketManifest(tablePath); - // ----------------------------------------------------------------- // Step 1 — baseline (no fault injection) // Plant an orphan KV snapshot dir under snap-99 (NOT registered in ZK) plus an diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java index 43764059341..4f457a497d9 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java @@ -214,7 +214,10 @@ void testBoundedPkTableEmitsKvBatchSplits() throws Throwable { @Test void testBoundedPkTableEmitsSnapshotSplitsByDefault() throws Throwable { - createTable(DEFAULT_TABLE_PATH, DEFAULT_PK_TABLE_DESCRIPTOR); + long tableId = createTable(DEFAULT_TABLE_PATH, DEFAULT_PK_TABLE_DESCRIPTOR); + for (int bucket = 0; bucket < DEFAULT_BUCKET_NUM; bucket++) { + FLUSS_CLUSTER_EXTENSION.waitUntilAllReplicaReady(new TableBucket(tableId, bucket)); + } int numSubtasks = DEFAULT_BUCKET_NUM; try (MockSplitEnumeratorContext context = new MockSplitEnumeratorContext<>(numSubtasks)) { 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..fa1fb97f8f2 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManager.java @@ -19,6 +19,7 @@ import org.apache.fluss.annotation.VisibleForTesting; import org.apache.fluss.exception.ApiException; +import org.apache.fluss.fs.FsPath; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TablePath; import org.apache.fluss.metrics.MetricNames; @@ -28,10 +29,15 @@ import org.apache.fluss.server.kv.snapshot.CompletedSnapshotHandleStore; import org.apache.fluss.server.kv.snapshot.CompletedSnapshotStore; import org.apache.fluss.server.kv.snapshot.CompletedSnapshotStore.SnapshotInUseChecker; +import org.apache.fluss.server.kv.snapshot.KvFileHandle; +import org.apache.fluss.server.kv.snapshot.KvFileHandleAndLocalPath; +import org.apache.fluss.server.kv.snapshot.KvSnapshotHandle; 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.ZkData; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -39,6 +45,10 @@ import javax.annotation.Nullable; import javax.annotation.concurrent.ThreadSafe; +import java.io.File; +import java.net.URI; +import java.nio.file.Path; +import java.nio.file.Paths; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; @@ -163,6 +173,115 @@ 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. Every referenced file must + * have a canonical SHA-256 digest, use a safe local path, and private files must remain under + * the declared snapshot location. + * + *

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."); + validateExternalSnapshotFiles(snapshot); + long nextSnapshotId = + new ZkSequenceIDCounter( + zooKeeperClient.getCuratorClient(), + ZkData.BucketSnapshotSequenceIdZNode.path(tableBucket)) + .getCurrent(); + checkArgument( + snapshot.getSnapshotID() < nextSnapshotId, + "External snapshot ID must be below the target bucket counter."); + getOrCreateCompletedSnapshotStore(tablePath, tableBucket) + .registerExternalSnapshot(snapshot, coordinatorZkVersion); + } + + private static void validateExternalSnapshotFiles(CompletedSnapshot snapshot) { + KvSnapshotHandle snapshotHandle = snapshot.getKvSnapshotHandle(); + Set localPaths = new HashSet<>(); + Set remotePaths = new HashSet<>(); + for (KvFileHandleAndLocalPath file : snapshotHandle.getSharedKvFileHandles()) { + validateExternalSnapshotFile(file, localPaths, remotePaths); + } + for (KvFileHandleAndLocalPath file : snapshotHandle.getPrivateFileHandles()) { + validateExternalSnapshotFile(file, localPaths, remotePaths); + requireWithinSnapshot(file.getKvFileHandle(), snapshot.getSnapshotLocation()); + } + checkArgument( + snapshotHandle.getIncrementalSize() >= 0L, + "Snapshot incremental size must not be negative."); + } + + private static void validateExternalSnapshotFile( + KvFileHandleAndLocalPath file, Set localPaths, Set remotePaths) { + KvFileHandle handle = file.getKvFileHandle(); + checkArgument(handle.getSize() >= 0L, "Snapshot file size must not be negative."); + checkArgument( + handle.getSha256() != null && handle.getSha256().matches("[0-9a-f]{64}"), + "External snapshot files must have a canonical SHA-256."); + checkArgument(isSafeLocalPath(file.getLocalPath()), "Snapshot local path is unsafe."); + checkArgument(localPaths.add(file.getLocalPath()), "Duplicate snapshot local path."); + checkArgument( + remotePaths.add(handle.getFilePath()), "Duplicate snapshot remote file path."); + } + + private static boolean isSafeLocalPath(String localPath) { + if (localPath == null + || localPath.isEmpty() + || localPath.startsWith("/") + || localPath.endsWith("/") + || localPath.indexOf('\\') >= 0 + || localPath.contains("//")) { + return false; + } + String[] components = localPath.split("/", -1); + for (String component : components) { + if (component.isEmpty() || ".".equals(component) || "..".equals(component)) { + return false; + } + } + try { + Path normalized = Paths.get(localPath).normalize(); + return !normalized.isAbsolute() + && localPath.equals(normalized.toString().replace(File.separatorChar, '/')); + } catch (RuntimeException invalid) { + return false; + } + } + + private static void requireWithinSnapshot(KvFileHandle file, FsPath snapshotLocation) { + URI child = new FsPath(file.getFilePath()).toUri().normalize(); + URI root = snapshotLocation.toUri().normalize(); + String rootPath = root.getPath().endsWith("/") ? root.getPath() : root.getPath() + '/'; + checkArgument( + java.util.Objects.equals(child.getScheme(), root.getScheme()) + && java.util.Objects.equals(child.getAuthority(), root.getAuthority()) + && child.getPath().startsWith(rootPath), + "Snapshot private file escapes its snapshot location."); + } + public void removeCompletedSnapshotStoreByTableBuckets(Set tableBuckets) { for (TableBucket tableBucket : tableBuckets) { bucketCompletedSnapshotStores.remove(tableBucket); @@ -244,10 +363,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 +375,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/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/CompletedSnapshotJsonSerde.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotJsonSerde.java index 3c9201ff80e..44d1bd8d301 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,110 @@ 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(), + serverHandle.getKvFileHandle().getSha256())); } - 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.getSha256()), + 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..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 @@ -42,6 +42,7 @@ 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.concurrent.LockUtils.inLock; @@ -116,6 +117,53 @@ public void add(final CompletedSnapshot completedSnapshot) throws Exception { completedSnapshot, snapshotsCleaner, () -> {})); } + /** + * Registers an immutable external snapshot and includes it in ordinary snapshot retention. + * + *

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. Obsolete + * snapshot cleanup follows ordinary retention semantics and may be deferred on failure. + */ + public void registerExternalSnapshot(final CompletedSnapshot snapshot, int coordinatorZkVersion) + throws Exception { + checkNotNull(snapshot, "Snapshot"); + inLock( + lock, + () -> { + CompletedSnapshot existing = stillInUseSnapshots.get(snapshot.getSnapshotID()); + for (CompletedSnapshot retained : completedSnapshots) { + if (retained.getSnapshotID() == snapshot.getSnapshotID()) { + existing = retained; + break; + } + } + if (existing != null) { + checkArgument( + existing.equals(snapshot), + "Conflicting snapshot identity for %s snapshot %s.", + snapshot.getTableBucket(), + snapshot.getSnapshotID()); + } 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, () -> {}); + } + }); + } + public long getPhysicalStorageRemoteKvSize() { return sharedKvFileRegistry.getFileSize(); } @@ -159,60 +207,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/kv/snapshot/KvFileHandle.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvFileHandle.java index 7f537815a65..cc5fa47c03f 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvFileHandle.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvFileHandle.java @@ -20,6 +20,8 @@ import org.apache.fluss.fs.FileSystem; import org.apache.fluss.fs.FsPath; +import javax.annotation.Nullable; + import java.io.IOException; import java.io.Serializable; import java.util.Objects; @@ -34,9 +36,17 @@ public class KvFileHandle implements Serializable { private final long size; + private final @Nullable String sha256; + public KvFileHandle(String fileHandle, long size) { + this(fileHandle, size, null); + } + + /** Creates a file handle with an optional lowercase hexadecimal SHA-256 digest. */ + public KvFileHandle(String fileHandle, long size, @Nullable String sha256) { this.filePath = fileHandle; this.size = size; + this.sha256 = sha256; } public String getFilePath() { @@ -47,6 +57,12 @@ public long getSize() { return size; } + /** Returns the expected lowercase hexadecimal SHA-256 when present. */ + @Nullable + public String getSha256() { + return sha256; + } + @Override public boolean equals(Object o) { if (this == o) { @@ -56,17 +72,26 @@ public boolean equals(Object o) { return false; } KvFileHandle that = (KvFileHandle) o; - return size == that.size && Objects.equals(filePath, that.filePath); + return size == that.size + && Objects.equals(filePath, that.filePath) + && Objects.equals(sha256, that.sha256); } @Override public int hashCode() { - return Objects.hash(filePath, size); + return Objects.hash(filePath, size, sha256); } @Override public String toString() { - return "KvFileHandle{" + "filePath=" + filePath + ", size=" + size + '}'; + return "KvFileHandle{" + + "filePath=" + + filePath + + ", size=" + + size + + ", sha256=" + + sha256 + + '}'; } /** diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataDownloader.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataDownloader.java index 48b235300a0..196432a7d2c 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataDownloader.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataDownloader.java @@ -22,7 +22,15 @@ import org.apache.fluss.fs.utils.FileDownloadSpec; import org.apache.fluss.fs.utils.FileDownloadUtils; import org.apache.fluss.utils.CloseableRegistry; +import org.apache.fluss.utils.FileUtils; +import java.io.IOException; +import java.io.InputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; @@ -64,24 +72,111 @@ void transferAllDataToDirectory( List fileDownloadSpecs = new ArrayList<>(); for (KvSnapshotDownloadSpec kvSnapshotDownloadSpec : kvSnapshotDownloadSpecs) { KvSnapshotHandle kvSnapshotHandle = kvSnapshotDownloadSpec.getKvSnapshotHandle(); - List fsPathAndFileNames = - Stream.concat( - kvSnapshotHandle.getSharedKvFileHandles().stream(), - kvSnapshotHandle.getPrivateFileHandles().stream()) - .map( - kvFileHandleAndLocalPath -> - new FsPathAndFileName( - new FsPath( - kvFileHandleAndLocalPath - .getKvFileHandle() - .getFilePath()), - kvFileHandleAndLocalPath.getLocalPath())) - .collect(Collectors.toList()); + List handles = fileHandles(kvSnapshotHandle); + List fsPathAndFileNames = new ArrayList<>(handles.size()); + for (KvFileHandleAndLocalPath handle : handles) { + resolveSafeLocalPath( + kvSnapshotDownloadSpec.getDownloadDestination(), handle.getLocalPath()); + fsPathAndFileNames.add( + new FsPathAndFileName( + new FsPath(handle.getKvFileHandle().getFilePath()), + handle.getLocalPath())); + } fileDownloadSpecs.add( new FileDownloadSpec( fsPathAndFileNames, kvSnapshotDownloadSpec.getDownloadDestination())); } - FileDownloadUtils.transferAllDataToDirectory( - fileDownloadSpecs, closeableRegistry, dataTransferThreadPool); + try { + FileDownloadUtils.transferAllDataToDirectory( + fileDownloadSpecs, closeableRegistry, dataTransferThreadPool); + verifyDownloadedFiles(kvSnapshotDownloadSpecs); + } catch (Exception failure) { + for (KvSnapshotDownloadSpec downloadSpec : kvSnapshotDownloadSpecs) { + FileUtils.deleteDirectoryQuietly(downloadSpec.getDownloadDestination().toFile()); + } + throw failure; + } + } + + private static void verifyDownloadedFiles( + Collection kvSnapshotDownloadSpecs) throws IOException { + for (KvSnapshotDownloadSpec downloadSpec : kvSnapshotDownloadSpecs) { + for (KvFileHandleAndLocalPath handle : + fileHandles(downloadSpec.getKvSnapshotHandle())) { + String expectedSha256 = handle.getKvFileHandle().getSha256(); + if (expectedSha256 == null) { + continue; + } + Path localFile = + resolveSafeLocalPath( + downloadSpec.getDownloadDestination(), handle.getLocalPath()); + if (Files.size(localFile) != handle.getKvFileHandle().getSize()) { + throw new IOException( + "Downloaded KV snapshot file length differs: " + localFile + '.'); + } + if (!expectedSha256.equals(sha256Hex(localFile))) { + throw new IOException( + "Downloaded KV snapshot file SHA-256 differs: " + localFile + '.'); + } + } + } + } + + private static List fileHandles(KvSnapshotHandle handle) { + return Stream.concat( + handle.getSharedKvFileHandles().stream(), + handle.getPrivateFileHandles().stream()) + .collect(Collectors.toList()); + } + + private static Path resolveSafeLocalPath(Path root, String localPath) throws IOException { + if (localPath == null + || localPath.isEmpty() + || localPath.startsWith("/") + || localPath.endsWith("/") + || localPath.indexOf('\\') >= 0 + || localPath.contains("//")) { + throw new IOException("Unsafe KV snapshot local path: " + localPath + '.'); + } + String[] components = localPath.split("/", -1); + for (String component : components) { + if (component.isEmpty() || ".".equals(component) || "..".equals(component)) { + throw new IOException("Unsafe KV snapshot local path: " + localPath + '.'); + } + } + Path normalized = Paths.get(localPath).normalize(); + Path normalizedRoot = root.toAbsolutePath().normalize(); + Path resolved = normalizedRoot.resolve(normalized).normalize(); + if (normalized.isAbsolute() + || !localPath.equals(normalized.toString().replace('\\', '/')) + || !resolved.startsWith(normalizedRoot)) { + throw new IOException("Unsafe KV snapshot local path: " + localPath + '.'); + } + return resolved; + } + + private static String sha256Hex(Path path) throws IOException { + MessageDigest digest; + try { + digest = MessageDigest.getInstance("SHA-256"); + } catch (NoSuchAlgorithmException e) { + throw new IllegalStateException("SHA-256 is unavailable.", e); + } + try (InputStream input = Files.newInputStream(path)) { + byte[] buffer = new byte[64 * 1024]; + int read; + while ((read = input.read(buffer)) != -1) { + digest.update(buffer, 0, read); + } + } + byte[] hash = digest.digest(); + char[] result = new char[hash.length * 2]; + char[] alphabet = "0123456789abcdef".toCharArray(); + for (int i = 0; i < hash.length; i++) { + int value = hash[i] & 0xff; + result[i * 2] = alphabet[value >>> 4]; + result[i * 2 + 1] = alphabet[value & 0xf]; + } + return new String(result); } } 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/main/java/org/apache/fluss/server/log/LogManager.java b/fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java index 4c1a84dd1fa..50b609c3d8f 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java @@ -421,6 +421,22 @@ public void truncateFullyAndStartAt(TableBucket tableBucket, long newOffset) { } } + /** Durably initializes an empty local log at the given snapshot offset. */ + public void initializeEmptyLocalTail(TableBucket tableBucket, long endOffset) { + LogTablet logTablet = currentLogs.get(tableBucket); + if (logTablet == null) { + throw new LogStorageException("Log tablet does not exist for " + tableBucket + "."); + } + logTablet.initializeEmptyLocalTail(endOffset); + try { + logTablet.flush(true); + } catch (IOException e) { + throw new LogStorageException( + "Failed to durably initialize the local tail for " + tableBucket + ".", e); + } + checkpointRecoveryOffsets(logTablet.getDataDir()); + } + private LogTablet loadLog( File dataDir, File tabletDir, diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java b/fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java index 47e0e7eb2c6..e47dbfff9bc 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 @@ -1195,6 +1196,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 e3eda80a82c..acb74f47220 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 @@ -488,6 +488,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(); updateRoutingState(data); @@ -504,7 +510,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. @@ -640,12 +646,10 @@ 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; - updateLeaderEndOffsetSnapshot(); - if (isDataLakeEnabled()) { registerLakeTieringMetrics(); } @@ -656,8 +660,10 @@ 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(); } private void registerLakeTieringMetrics() { @@ -758,7 +764,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(); @@ -776,11 +782,23 @@ 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) { 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, @@ -855,7 +873,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(); @@ -879,8 +897,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; @@ -903,6 +920,9 @@ private Optional initKvTablet() { checkNotNull(kvTablet, "kv tablet should not be null."); restoreStartOffset = completedSnapshot.getLogOffset(); + if (recovery.initializeLocalLog) { + logManager.initializeEmptyLocalTail(tableBucket, restoreStartOffset); + } rowCount = supportsExactRowCount(tableConfig) ? completedSnapshot.getRowCount() : null; // currently, we only support one auto-increment column. @@ -1022,18 +1042,51 @@ 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( snapshotContext.getLatestCompletedSnapshotProvider().apply(tableBucket)); } catch (Exception e) { - LOG.warn( - "Get latest completed snapshot for {} of table {} failed.", - tableBucket, - physicalPath, - e); + throw new KvStorageException( + "Failed to get the latest completed snapshot for " + tableBucket + '.', e); } - return Optional.empty(); } private void recoverKvTablet( diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java index 01f897cff29..12d166aad2f 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; /** @@ -1341,6 +1342,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..e432bbbafac 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,28 @@ 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.KvFileHandle; +import org.apache.fluss.server.kv.snapshot.KvFileHandleAndLocalPath; +import org.apache.fluss.server.kv.snapshot.KvSnapshotHandle; 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; @@ -47,6 +56,7 @@ import java.nio.file.Files; import java.nio.file.Path; import java.util.ArrayList; +import java.util.Collections; import java.util.Comparator; import java.util.HashMap; import java.util.HashSet; @@ -60,6 +70,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 +291,240 @@ private void verifyMissingSnapshotMetadataIsCleanedUp(IOException exception) thr assertThat(completedSnapshotHandleStore.get(tableBucket, 2L)).isEmpty(); } + @Test + void testExternalSnapshotRequiresDigestAndSafeLocalPath() throws Exception { + TableBucket bucket = new TableBucket(99, 0); + int epoch = + zookeeperClient + .fenceBecomeCoordinatorLeader("coordinator") + .getCoordinatorEpochZkVersion(); + FsPath location = new FsPath(tempDir.resolve("snapshot").toString()); + CompletedSnapshotStoreManager manager = createCompletedSnapshotStoreManager(1); + + CompletedSnapshot missingDigest = + externalSnapshotWithFile(bucket, location, "file.sst", null); + assertThatThrownBy( + () -> + manager.registerExternalSnapshot( + DATA1_TABLE_PATH, + bucket, + writeExternalSnapshot(missingDigest), + epoch)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("SHA-256"); + + String sha256 = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; + CompletedSnapshot unsafePath = + externalSnapshotWithFile( + bucket, + new FsPath(tempDir.resolve("unsafe-snapshot").toString()), + "../file.sst", + sha256); + assertThatThrownBy( + () -> + manager.registerExternalSnapshot( + DATA1_TABLE_PATH, + bucket, + writeExternalSnapshot(unsafePath), + epoch)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("local path"); + + FsPath escapedLocation = new FsPath(tempDir.resolve("escaped-snapshot").toString()); + CompletedSnapshot escapedRemotePath = + externalSnapshotWithFile( + bucket, + escapedLocation, + new FsPath(tempDir.resolve("outside.sst").toString()).toString(), + "file.sst", + sha256); + assertThatThrownBy( + () -> + manager.registerExternalSnapshot( + DATA1_TABLE_PATH, + bucket, + writeExternalSnapshot(escapedRemotePath), + epoch)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("escapes"); + } + + @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("below the target bucket counter"); + 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 CompletedSnapshot externalSnapshotWithFile( + TableBucket bucket, FsPath location, String localPath, String sha256) { + return externalSnapshotWithFile( + bucket, location, new FsPath(location, "file.sst").toString(), localPath, sha256); + } + + private static CompletedSnapshot externalSnapshotWithFile( + TableBucket bucket, + FsPath location, + String remotePath, + String localPath, + String sha256) { + KvFileHandle file = new KvFileHandle(remotePath, 1L, sha256); + return new CompletedSnapshot( + bucket, + 0L, + location, + KvSnapshotHandle.create( + Collections.emptyList(), + Collections.singletonList(KvFileHandleAndLocalPath.of(file, localPath)), + 1L), + 0L, + null, + null); + } + + 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( @@ -372,6 +621,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/coordinator/ExternalKvSnapshotITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/ExternalKvSnapshotITCase.java new file mode 100644 index 00000000000..f36e510e1eb --- /dev/null +++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/ExternalKvSnapshotITCase.java @@ -0,0 +1,360 @@ +/* + * 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.FSDataInputStream; +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.metrics.ThreadSafeSimpleCounter; +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.security.MessageDigest; +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 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, + new ThreadSafeSimpleCounter())) { + 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) throws Exception { + List metadata = new ArrayList<>(files.size()); + for (KvFileHandleAndLocalPath file : files) { + metadata.add( + new KvSnapshotFileMetadata.FileHandle( + file.getKvFileHandle().getFilePath(), + file.getKvFileHandle().getSize(), + file.getLocalPath(), + sha256(new FsPath(file.getKvFileHandle().getFilePath())))); + } + return metadata; + } + + private static String sha256(FsPath path) throws Exception { + MessageDigest digest = MessageDigest.getInstance("SHA-256"); + try (FSDataInputStream input = path.getFileSystem().open(path)) { + byte[] buffer = new byte[64 * 1024]; + int read; + while ((read = input.read(buffer)) != -1) { + digest.update(buffer, 0, read); + } + } + byte[] hash = digest.digest(); + char[] result = new char[hash.length * 2]; + char[] alphabet = "0123456789abcdef".toCharArray(); + for (int i = 0; i < hash.length; i++) { + int value = hash[i] & 0xff; + result[i * 2] = alphabet[value >>> 4]; + result[i * 2 + 1] = alphabet[value & 0xf]; + } + return new String(result); + } + + private static void putRecords( + FlussClusterExtension cluster, TableBucket bucket, List records) + throws Exception { + PutKvResponse response = + cluster.newTabletServerClientForNode(cluster.waitAndGetLeader(bucket)) + .putKv( + newPutKvRequest( + bucket.getTableId(), + bucket.getBucket(), + -1, + toKvRecordBatch(records))) + .get(); + assertThat(response.getBucketsRespAt(0).hasErrorCode()).isFalse(); + } + + private static void assertReplicaOffsets( + FlussClusterExtension cluster, TableBucket bucket, long start, long end) + throws Exception { + retry( + Duration.ofMinutes(1), + () -> { + for (int server = 0; server < 2; server++) { + ReplicaManager replicaManager = + cluster.getTabletServerById(server).getReplicaManager(); + assertThat(replicaManager.getReplica(bucket)) + .isInstanceOf(ReplicaManager.OnlineReplica.class); + Replica replica = replicaManager.getReplicaOrException(bucket); + assertThat(replica.getLogTablet().localLogStartOffset()).isEqualTo(start); + assertThat(replica.getLocalLogEndOffset()).isEqualTo(end); + assertThat(replica.getLogHighWatermark()).isEqualTo(end); + } + }); + } + + private static void assertRows( + FlussClusterExtension cluster, TableBucket bucket, List records) + throws Exception { + TabletServerGateway gateway = + cluster.newTabletServerClientForNode(cluster.waitAndGetLeader(bucket)); + for (Tuple2 keyValue : getKeyValuePairs(records)) { + assertLookupResponse( + gateway.lookup( + newLookupRequest( + bucket.getTableId(), bucket.getBucket(), keyValue.f0)) + .get(), + keyValue.f1); + } + } +} diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotJsonSerdeTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotJsonSerdeTest.java index 117fef36349..aaf16657d61 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotJsonSerdeTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotJsonSerdeTest.java @@ -22,10 +22,14 @@ import org.apache.fluss.server.kv.autoinc.AutoIncIDRange; import org.apache.fluss.utils.json.JsonSerdeTestBase; +import org.junit.jupiter.api.Test; + import java.util.Arrays; import java.util.Collections; import java.util.List; +import static org.assertj.core.api.Assertions.assertThat; + /** Test for {@link org.apache.fluss.server.kv.snapshot.CompletedSnapshotJsonSerde}. */ class CompletedSnapshotJsonSerdeTest extends JsonSerdeTestBase { @@ -33,6 +37,32 @@ protected CompletedSnapshotJsonSerdeTest() { super(CompletedSnapshotJsonSerde.INSTANCE); } + @Test + void testFileDigestRoundTrip() { + String sha256 = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; + CompletedSnapshot snapshot = + new CompletedSnapshot( + new TableBucket(1, 1), + 1, + new FsPath("oss://bucket/snapshot"), + KvSnapshotHandle.create( + Collections.emptyList(), + Collections.singletonList( + KvFileHandleAndLocalPath.of( + new KvFileHandle( + "oss://bucket/snapshot/private.sst", + 10, + sha256), + "private.sst")), + 10), + 10, + null, + null); + + assertThat(CompletedSnapshotJsonSerde.fromJson(CompletedSnapshotJsonSerde.toJson(snapshot))) + .isEqualTo(snapshot); + } + @Override protected CompletedSnapshot[] createObjects() { List sharedFileHandles = diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataDownloaderTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataDownloaderTest.java index b97521fca1f..3687636f9fe 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataDownloaderTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataDownloaderTest.java @@ -28,6 +28,7 @@ import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; +import java.security.MessageDigest; import java.util.ArrayList; import java.util.List; import java.util.Random; @@ -82,6 +83,60 @@ void testMultiThreadRestoreCorrectly(@TempDir Path destDir, @TempDir Path srcDir } } + @Test + void testRejectUnsafeLocalPathBeforeDownload(@TempDir Path destDir, @TempDir Path srcDir) + throws Exception { + Path source = srcDir.resolve("state"); + Files.write(source, new byte[] {1}); + KvSnapshotHandle handle = + KvSnapshotHandle.create( + java.util.Collections.singletonList( + KvFileHandleAndLocalPath.of( + new KvFileHandle(source.toString(), 1L), "../state")), + java.util.Collections.emptyList(), + 1L); + + assertThatThrownBy( + () -> + new KvSnapshotDataDownloader(downLoaderThreadPool) + .transferAllDataToDirectory( + new KvSnapshotDownloadSpec(handle, destDir), + new CloseableRegistry())) + .isInstanceOf(IOException.class) + .hasMessageContaining("Unsafe KV snapshot local path"); + assertThat(destDir.resolve("../state").normalize()).doesNotExist(); + } + + @Test + void testDigestMismatchCleansUpDownloadedFiles(@TempDir Path destDir, @TempDir Path srcDir) + throws Exception { + byte[] content = new byte[] {1, 2, 3}; + Path source = srcDir.resolve("state"); + Files.write(source, content); + KvSnapshotHandle handle = + KvSnapshotHandle.create( + java.util.Collections.singletonList( + KvFileHandleAndLocalPath.of( + new KvFileHandle( + source.toString(), + content.length, + "0000000000000000000000000000000000000000000000000000000000000000"), + "state")), + java.util.Collections.emptyList(), + content.length); + Path destination = destDir.resolve("download"); + + assertThatThrownBy( + () -> + new KvSnapshotDataDownloader(downLoaderThreadPool) + .transferAllDataToDirectory( + new KvSnapshotDownloadSpec(handle, destination), + new CloseableRegistry())) + .isInstanceOf(IOException.class) + .hasMessageContaining("SHA-256"); + assertThat(destination).doesNotExist(); + } + /** Tests cleanup on download failures. */ @Test public void testMultiThreadCleanupOnFailure(@TempDir Path destDir, @TempDir Path srcDir) @@ -148,7 +203,8 @@ private KvSnapshotDownloadSpec createDownloadRequestForContent( for (int i = 0; i < numSubHandles; ++i) { Path path = srcPath.resolve(String.format("state-%d-%d", remoteHandleId, i)); Files.write(path, content[i]); - handles.add(new KvFileHandle(path.toString(), content.length)); + handles.add( + new KvFileHandle(path.toString(), content[i].length, sha256Hex(content[i]))); } List sharedStates = new ArrayList<>(numSubHandles); @@ -167,4 +223,20 @@ private KvSnapshotDownloadSpec createDownloadRequestForContent( return new KvSnapshotDownloadSpec(kvSnapshotHandle, dstPath); } + + private static String sha256Hex(byte[] content) { + try { + byte[] hash = MessageDigest.getInstance("SHA-256").digest(content); + char[] result = new char[hash.length * 2]; + char[] alphabet = "0123456789abcdef".toCharArray(); + for (int i = 0; i < hash.length; i++) { + int value = hash[i] & 0xff; + result[i * 2] = alphabet[value >>> 4]; + result[i * 2 + 1] = alphabet[value & 0xf]; + } + return new String(result); + } catch (Exception e) { + throw new IllegalStateException(e); + } + } } 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 {} 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 b133eeccad2..8ce4cc4f17d 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 @@ -405,6 +405,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 50a9dc79ec2..6faec67a589 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.NotLeaderOrFollowerException; import org.apache.fluss.exception.OutOfOrderSequenceException; import org.apache.fluss.fs.FsPath; @@ -53,6 +54,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; @@ -113,6 +115,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 { @@ -698,6 +702,146 @@ void testKvReplicaSnapshot(@TempDir File snapshotKvTabletDir) throws Exception { KvTestUtils.checkSnapshot(completedSnapshot2, expectedKeyValues, expectedLogOffset); } + @Test + void testSnapshotLookupFailurePreventsEmptyRecovery(@TempDir File snapshotDir) + throws Exception { + TestSnapshotContext context = + new TestSnapshotContext(snapshotDir.getPath()) { + @Override + public FunctionWithException + getLatestCompletedSnapshotProvider() { + return bucket -> { + throw new IOException("Snapshot metadata unavailable"); + }; + } + }; + Replica replica = + makeKvReplica( + DATA1_PHYSICAL_TABLE_PATH_PK, + new TableBucket(DATA1_TABLE_ID_PK, 1), + context); + assertThatThrownBy(() -> makeKvReplicaAsLeader(replica, 0)) + .isInstanceOf(KvStorageException.class) + .hasRootCauseMessage("Snapshot metadata unavailable"); + assertThat(replica.getKvTablet()).isNull(); + assertThat(replica.getLocalLogEndOffset()).isZero(); + } + + @Test + void 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 { + 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); + TestSnapshotContext context = + new TestSnapshotContext(snapshotDir.getPath()) { + @Override + public FunctionWithException + getLatestCompletedSnapshotProvider() { + return ignored -> external; + } + + @Override + public KvSnapshotDataDownloader getSnapshotDataDownloader() { + return new KvSnapshotDataDownloader(executorService) { + @Override + public void transferAllDataToDirectory( + KvSnapshotDownloadSpec spec, CloseableRegistry registry) + throws Exception { + if (failDownload.get()) { + throw new IOException("Snapshot download unavailable"); + } + super.transferAllDataToDirectory(spec, registry); + } + }; + } + }; + Replica replica = makeKvReplica(DATA1_PHYSICAL_TABLE_PATH_PK, bucket, context); + assertThatThrownBy(() -> makeKvReplicaAsLeader(replica, 2)) + .isInstanceOf(KvStorageException.class); + assertThat(replica.isLeader()).isFalse(); + assertThat(replica.getLocalLogEndOffset()).isZero(); + assertThat(replica.getLogHighWatermark()).isZero(); + failDownload.set(false); + makeKvReplicaAsLeader(replica, 3); + assertThat(replica.getLocalLogEndOffset()).isEqualTo(10_017L); + assertThat(replica.getLogHighWatermark()).isEqualTo(10_017L); + assertThat(replica.getLeaderEndOffsetSnapshot()).isEqualTo(10_017L); + verifyGetKeyValues( + replica.getKvTablet(), + getKeyValuePairs(genKvRecords(Tuple2.of("k1", new Object[] {1, "a"})))); + } + @Test void testSnapshotUseLatestLeaderEpoch(@TempDir File snapshotKvTabletDir) throws Exception { TableBucket tableBucket = new TableBucket(DATA1_TABLE_ID_PK, 1); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java b/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java index cff3f788c99..0d864818cd2 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java @@ -684,6 +684,9 @@ public void waitUntilAllReplicaReady(TableBucket tableBucket) { List isr = leaderAndIsr.isr(); for (int replicaId : isr) { TabletServer tabletServer = getTabletServerById(replicaId); + assertThat(tabletServer) + .as("Tablet server %s should be available.", replicaId) + .isNotNull(); ReplicaManager replicaManager = tabletServer.getReplicaManager(); assertThat(replicaManager.getReplica(tableBucket)) .isInstanceOf(ReplicaManager.OnlineReplica.class); @@ -696,7 +699,11 @@ public void waitUntilAllReplicaReady(TableBucket tableBucket) { } int leader = leaderAndIsr.leader(); - ReplicaManager replicaManager = getTabletServerById(leader).getReplicaManager(); + TabletServer leaderServer = getTabletServerById(leader); + assertThat(leaderServer) + .as("Leader tablet server %s should be available.", leader) + .isNotNull(); + ReplicaManager replicaManager = leaderServer.getReplicaManager(); assertThat(replicaManager.getReplicaOrException(tableBucket).isLeader()) .isTrue(); });