diff --git a/fluss-common/src/main/java/org/apache/fluss/remote/RemoteLogManifest.java b/fluss-common/src/main/java/org/apache/fluss/remote/RemoteLogManifest.java index bc856e361d..99c7f05eb2 100644 --- a/fluss-common/src/main/java/org/apache/fluss/remote/RemoteLogManifest.java +++ b/fluss-common/src/main/java/org/apache/fluss/remote/RemoteLogManifest.java @@ -17,6 +17,8 @@ package org.apache.fluss.remote; +import org.apache.fluss.annotation.VisibleForTesting; +import org.apache.fluss.fs.FsPath; import org.apache.fluss.metadata.PhysicalTablePath; import org.apache.fluss.metadata.TableBucket; @@ -38,13 +40,17 @@ public class RemoteLogManifest { private final TableBucket tableBucket; private final List remoteLogSegmentList; + private final FsPath remoteLogDir; + public RemoteLogManifest( PhysicalTablePath physicalTablePath, TableBucket tableBucket, - List remoteLogSegmentList) { + List remoteLogSegmentList, + FsPath remoteLogDir) { this.physicalTablePath = physicalTablePath; this.tableBucket = tableBucket; this.remoteLogSegmentList = Collections.unmodifiableList(remoteLogSegmentList); + this.remoteLogDir = remoteLogDir; // sanity check for (RemoteLogSegment remoteLogSegment : remoteLogSegmentList) { @@ -73,7 +79,7 @@ public RemoteLogManifest trimAndMerge( } newSegments.addAll(addedSegments); newSegments.sort(Comparator.comparingLong(RemoteLogSegment::remoteLogStartOffset)); - return new RemoteLogManifest(physicalTablePath, tableBucket, newSegments); + return new RemoteLogManifest(physicalTablePath, tableBucket, newSegments, remoteLogDir); } public long getRemoteLogStartOffset() { @@ -120,10 +126,36 @@ public TableBucket getTableBucket() { return tableBucket; } + public FsPath getRemoteLogDir() { + return remoteLogDir; + } + + @VisibleForTesting public List getRemoteLogSegmentList() { return remoteLogSegmentList; } + public RemoteLogManifest newManifest(FsPath remoteLogDir) { + List newRemoteLogSegments = new ArrayList<>(remoteLogSegmentList.size()); + for (RemoteLogSegment remoteLogSegment : remoteLogSegmentList) { + newRemoteLogSegments.add( + RemoteLogSegment.Builder.builder() + .physicalTablePath(remoteLogSegment.physicalTablePath()) + .tableBucket(remoteLogSegment.tableBucket()) + .remoteLogSegmentId(remoteLogSegment.remoteLogSegmentId()) + .remoteLogStartOffset(remoteLogSegment.remoteLogStartOffset()) + .remoteLogEndOffset(remoteLogSegment.remoteLogEndOffset()) + .maxTimestamp(remoteLogSegment.maxTimestamp()) + .segmentSizeInBytes(remoteLogSegment.segmentSizeInBytes()) + // We set remoteLogDir manually here, so subsequent usage will be safe + // to directly use it. + .remoteLogDir(remoteLogDir) + .build()); + } + return new RemoteLogManifest( + physicalTablePath, tableBucket, newRemoteLogSegments, remoteLogDir); + } + @Override public boolean equals(Object o) { if (this == o) { diff --git a/fluss-common/src/main/java/org/apache/fluss/remote/RemoteLogManifestJsonSerde.java b/fluss-common/src/main/java/org/apache/fluss/remote/RemoteLogManifestJsonSerde.java index c90a85ea02..833bf8eac3 100644 --- a/fluss-common/src/main/java/org/apache/fluss/remote/RemoteLogManifestJsonSerde.java +++ b/fluss-common/src/main/java/org/apache/fluss/remote/RemoteLogManifestJsonSerde.java @@ -17,6 +17,7 @@ package org.apache.fluss.remote; +import org.apache.fluss.fs.FsPath; import org.apache.fluss.metadata.PhysicalTablePath; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator; @@ -49,6 +50,7 @@ public class RemoteLogManifestJsonSerde private static final String END_OFFSET_FIELD = "end_offset"; private static final String MAX_TIMESTAMP_FIELD = "max_timestamp"; private static final String SEGMENT_SIZE_IN_BYTES_FIELD = "size_in_bytes"; + private static final String REMOTE_LOG_DIR_FIELD = "remote_log_dir"; private static final int SNAPSHOT_VERSION = 1; @Override @@ -84,9 +86,14 @@ public void serialize(RemoteLogManifest manifest, JsonGenerator generator) throw generator.writeNumberField(MAX_TIMESTAMP_FIELD, remoteLogSegment.maxTimestamp()); generator.writeNumberField( SEGMENT_SIZE_IN_BYTES_FIELD, remoteLogSegment.segmentSizeInBytes()); + generator.writeStringField( + REMOTE_LOG_DIR_FIELD, remoteLogSegment.remoteLogDir().toString()); generator.writeEndObject(); } generator.writeEndArray(); + + generator.writeStringField(REMOTE_LOG_DIR_FIELD, manifest.getRemoteLogDir().toString()); + generator.writeEndObject(); } @@ -118,6 +125,11 @@ public RemoteLogManifest deserialize(JsonNode node) { long endOffset = entryJson.get(END_OFFSET_FIELD).asLong(); long maxTimestamp = entryJson.get(MAX_TIMESTAMP_FIELD).asLong(); int segmentSizeInBytes = entryJson.get(SEGMENT_SIZE_IN_BYTES_FIELD).asInt(); + // backward compatibility for existing RemoteLogSegment which does not have remoteLogDir + FsPath remoteLogDir = null; + if (entryJson.has(REMOTE_LOG_DIR_FIELD)) { + remoteLogDir = new FsPath(entryJson.get(REMOTE_LOG_DIR_FIELD).asText()); + } snapshotEntries.add( RemoteLogSegment.Builder.builder() .physicalTablePath(physicalTablePath) @@ -127,10 +139,17 @@ public RemoteLogManifest deserialize(JsonNode node) { .remoteLogEndOffset(endOffset) .maxTimestamp(maxTimestamp) .segmentSizeInBytes(segmentSizeInBytes) + .remoteLogDir(remoteLogDir) .build()); } - return new RemoteLogManifest(physicalTablePath, tableBucket, snapshotEntries); + // backward compatibility for existing RemoteLogManifest which does not have remoteLogDir + FsPath remoteLogDir = null; + if (node.has(REMOTE_LOG_DIR_FIELD)) { + remoteLogDir = new FsPath(node.get(REMOTE_LOG_DIR_FIELD).asText()); + } + + return new RemoteLogManifest(physicalTablePath, tableBucket, snapshotEntries, remoteLogDir); } public static RemoteLogManifest fromJson(byte[] json) { diff --git a/fluss-common/src/main/java/org/apache/fluss/remote/RemoteLogSegment.java b/fluss-common/src/main/java/org/apache/fluss/remote/RemoteLogSegment.java index 39480e4d10..0d30ae47cd 100644 --- a/fluss-common/src/main/java/org/apache/fluss/remote/RemoteLogSegment.java +++ b/fluss-common/src/main/java/org/apache/fluss/remote/RemoteLogSegment.java @@ -18,6 +18,7 @@ package org.apache.fluss.remote; import org.apache.fluss.annotation.Internal; +import org.apache.fluss.fs.FsPath; import org.apache.fluss.metadata.PhysicalTablePath; import org.apache.fluss.metadata.TableBucket; @@ -50,6 +51,8 @@ public class RemoteLogSegment { private final int segmentSizeInBytes; + private final FsPath remoteLogDir; + private RemoteLogSegment( PhysicalTablePath physicalTablePath, TableBucket tableBucket, @@ -57,7 +60,8 @@ private RemoteLogSegment( long remoteLogStartOffset, long remoteLogEndOffset, long maxTimestamp, - int segmentSizeInBytes) { + int segmentSizeInBytes, + FsPath remoteLogDir) { this.physicalTablePath = checkNotNull(physicalTablePath); this.tableBucket = checkNotNull(tableBucket); this.remoteLogSegmentId = checkNotNull(remoteLogSegmentId); @@ -79,6 +83,7 @@ private RemoteLogSegment( this.remoteLogEndOffset = remoteLogEndOffset; this.maxTimestamp = maxTimestamp; this.segmentSizeInBytes = segmentSizeInBytes; + this.remoteLogDir = remoteLogDir; } public PhysicalTablePath physicalTablePath() { @@ -115,6 +120,10 @@ public int segmentSizeInBytes() { return segmentSizeInBytes; } + public FsPath remoteLogDir() { + return remoteLogDir; + } + @Override public boolean equals(Object o) { if (this == o) { @@ -130,7 +139,8 @@ public boolean equals(Object o) { && maxTimestamp == that.maxTimestamp && Objects.equals(remoteLogSegmentId, that.remoteLogSegmentId) && Objects.equals(physicalTablePath, that.physicalTablePath) - && Objects.equals(tableBucket, that.tableBucket); + && Objects.equals(tableBucket, that.tableBucket) + && Objects.equals(remoteLogDir, that.remoteLogDir); } @Override @@ -142,7 +152,8 @@ public int hashCode() { remoteLogStartOffset, remoteLogEndOffset, maxTimestamp, - segmentSizeInBytes); + segmentSizeInBytes, + remoteLogDir); } @Override @@ -162,6 +173,8 @@ public String toString() { + maxTimestamp + ", segmentSizeInBytes=" + segmentSizeInBytes + + ", remoteLogDir=" + + remoteLogDir + '}'; } @@ -174,6 +187,7 @@ public static class Builder { private long remoteLogEndOffset; private long maxTimestamp; private int segmentSizeInBytes; + private FsPath remoteLogDir; public static Builder builder() { return new Builder(); @@ -214,6 +228,11 @@ public Builder tableBucket(TableBucket tableBucket) { return this; } + public Builder remoteLogDir(FsPath remoteLogDir) { + this.remoteLogDir = remoteLogDir; + return this; + } + public RemoteLogSegment build() { return new RemoteLogSegment( physicalTablePath, @@ -222,7 +241,8 @@ public RemoteLogSegment build() { remoteLogStartOffset, remoteLogEndOffset, maxTimestamp, - segmentSizeInBytes); + segmentSizeInBytes, + remoteLogDir); } } } diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/FlussPaths.java b/fluss-common/src/main/java/org/apache/fluss/utils/FlussPaths.java index 1c75663ba3..a7c3945952 100644 --- a/fluss-common/src/main/java/org/apache/fluss/utils/FlussPaths.java +++ b/fluss-common/src/main/java/org/apache/fluss/utils/FlussPaths.java @@ -426,6 +426,19 @@ public static FsPath remoteLogDir(Configuration conf) { return new FsPath(conf.get(ConfigOptions.REMOTE_DATA_DIR) + "/" + REMOTE_LOG_DIR_NAME); } + /** + * Returns the remote root directory path for storing log files. + * + *

The path contract: + * + * 

+     * {$remote.data.dir}/log
+     *
+ */ + public static FsPath remoteLogDir(String remoteDataDir) { + return new FsPath(remoteDataDir, REMOTE_LOG_DIR_NAME); + } + /** * Returns the remote directory path for storing log files for a log tablet. * @@ -592,6 +605,19 @@ public static FsPath remoteKvDir(Configuration conf) { return new FsPath(conf.get(ConfigOptions.REMOTE_DATA_DIR) + "/" + REMOTE_KV_DIR_NAME); } + /** + * Returns the remote root directory path for storing kv snapshot files. + * + *

The path contract: + * + * 

+     * {$remote.data.dir}/kv
+     *
+ */ + public static FsPath remoteKvDir(String remoteDataDir) { + return new FsPath(remoteDataDir, REMOTE_KV_DIR_NAME); + } + /** * Returns the remote directory path for storing kv snapshot files for a kv tablet. * diff --git a/fluss-common/src/test/java/org/apache/fluss/remote/RemoteLogManifestJsonSerdeTest.java b/fluss-common/src/test/java/org/apache/fluss/remote/RemoteLogManifestJsonSerdeTest.java index e095132158..744f5e3ac6 100644 --- a/fluss-common/src/test/java/org/apache/fluss/remote/RemoteLogManifestJsonSerdeTest.java +++ b/fluss-common/src/test/java/org/apache/fluss/remote/RemoteLogManifestJsonSerdeTest.java @@ -17,14 +17,19 @@ package org.apache.fluss.remote; +import org.apache.fluss.fs.FsPath; import org.apache.fluss.metadata.PhysicalTablePath; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TablePath; import org.apache.fluss.utils.json.JsonSerdeTestBase; +import org.junit.jupiter.api.Test; + import java.util.Arrays; import java.util.UUID; +import static org.assertj.core.api.Assertions.assertThat; + /** Tests of {@link RemoteLogManifestJsonSerde}. */ class RemoteLogManifestJsonSerdeTest extends JsonSerdeTestBase { private static final PhysicalTablePath TABLE_PATH1 = @@ -35,6 +40,8 @@ class RemoteLogManifestJsonSerdeTest extends JsonSerdeTestBase + new LogStorageException( + String.format( + "Failed to load table '%s': table info not found in zookeeper metadata.", + tablePath))); + } + + public static SchemaInfo getSchemaInfo(ZooKeeperClient zkClient, TablePath tablePath) + throws Exception { int schemaId = zkClient.getCurrentSchemaId(tablePath); Optional schemaInfoOpt = zkClient.getSchemaById(tablePath, schemaId); SchemaInfo schemaInfo; @@ -239,17 +261,20 @@ public static TableInfo getTableInfo(ZooKeeperClient zkClient, TablePath tablePa } else { schemaInfo = schemaInfoOpt.get(); } + return schemaInfo; + } - TableRegistration tableRegistration = - zkClient.getTable(tablePath) - .orElseThrow( - () -> - new LogStorageException( - String.format( - "Failed to load table '%s': table info not found in zookeeper metadata.", - tablePath))); - - return tableRegistration.toTableInfo(tablePath, schemaInfo); + public static PartitionRegistration getPartitionRegistration( + ZooKeeperClient zkClient, PhysicalTablePath physicalTablePath) throws Exception { + return zkClient.getPartition( + physicalTablePath.getTablePath(), physicalTablePath.getPartitionName()) + .orElseThrow( + () -> + new PartitionNotExistException( + String.format( + "Failed to load partition '%s' for table %s: partition info not found in zookeeper metadata.", + physicalTablePath.getPartitionName(), + physicalTablePath.getTablePath()))); } /** Create a tablet directory in the given dir. */ diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java index 6a5c6079c9..a66d2f10db 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java @@ -25,7 +25,6 @@ import org.apache.fluss.config.cluster.ServerReconfigurable; import org.apache.fluss.exception.ConfigException; import org.apache.fluss.exception.KvStorageException; -import org.apache.fluss.fs.FileSystem; import org.apache.fluss.fs.FsPath; import org.apache.fluss.memory.LazyMemorySegmentPool; import org.apache.fluss.memory.MemorySegmentPool; @@ -128,10 +127,6 @@ public static RateLimiter getDefaultRateLimiter() { /** The memory segment pool to allocate memorySegment. */ private final MemorySegmentPool memorySegmentPool; - private final FsPath remoteKvDir; - - private final FileSystem remoteFileSystem; - /** * The shared rate limiter for all RocksDB instances to control flush and compaction write rate. */ @@ -148,16 +143,13 @@ private KvManager( ZooKeeperClient zkClient, int recoveryThreadsPerDataDir, LogManager logManager, - TabletServerMetricGroup tabletServerMetricGroup) - throws IOException { + TabletServerMetricGroup tabletServerMetricGroup) { super(TabletType.KV, localDiskManager.dataDirs(), conf, recoveryThreadsPerDataDir); this.localDiskManager = localDiskManager; this.logManager = logManager; this.arrowBufferAllocator = BufferAllocatorUtil.createBufferAllocator(null); this.memorySegmentPool = LazyMemorySegmentPool.createServerBufferPool(conf); this.zkClient = zkClient; - this.remoteKvDir = FlussPaths.remoteKvDir(conf); - this.remoteFileSystem = remoteKvDir.getFileSystem(); this.serverMetricGroup = tabletServerMetricGroup; this.sharedRocksDBRateLimiter = createSharedRateLimiter(conf); this.currentSharedRateLimitBytesPerSec = @@ -415,12 +407,13 @@ public KvTablet loadKv(File tabletDir, SchemaGetter schemaGetter) throws Excepti } public void deleteRemoteKvSnapshot( - PhysicalTablePath physicalTablePath, TableBucket tableBucket) { + String remoteDataDir, PhysicalTablePath physicalTablePath, TableBucket tableBucket) { FsPath remoteKvTabletDir = - FlussPaths.remoteKvTabletDir(remoteKvDir, physicalTablePath, tableBucket); + FlussPaths.remoteKvTabletDir( + FlussPaths.remoteKvDir(remoteDataDir), physicalTablePath, tableBucket); try { - if (remoteFileSystem.exists(remoteKvTabletDir)) { - remoteFileSystem.delete(remoteKvTabletDir, true); + if (remoteKvTabletDir.getFileSystem().exists(remoteKvTabletDir)) { + remoteKvTabletDir.getFileSystem().delete(remoteKvTabletDir, true); LOG.info("Delete table's remote bucket snapshot dir of {} success.", tableBucket); } } catch (Exception e) { diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/DefaultSnapshotContext.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/DefaultSnapshotContext.java index 492d8aca41..8272c6b6a3 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/DefaultSnapshotContext.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/DefaultSnapshotContext.java @@ -21,11 +21,9 @@ import org.apache.fluss.config.Configuration; import org.apache.fluss.config.cluster.ServerReconfigurable; import org.apache.fluss.exception.ConfigException; -import org.apache.fluss.fs.FsPath; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.server.kv.KvSnapshotResource; import org.apache.fluss.server.zk.ZooKeeperClient; -import org.apache.fluss.utils.FlussPaths; import org.apache.fluss.utils.function.FunctionWithException; import org.slf4j.Logger; @@ -61,8 +59,6 @@ public class DefaultSnapshotContext implements SnapshotContext, ServerReconfigur private final int remoteLogDownloadThreadsInRecoverKv; - private final FsPath remoteKvDir; - private DefaultSnapshotContext( ZooKeeperClient zooKeeperClient, CompletedKvSnapshotCommitter completedKvSnapshotCommitter, @@ -72,7 +68,6 @@ private DefaultSnapshotContext( KvSnapshotDataDownloader kvSnapshotDataDownloader, long kvSnapshotIntervalMs, int writeBufferSizeInBytes, - FsPath remoteKvDir, CompletedSnapshotHandleStore completedSnapshotHandleStore, int maxFetchLogSizeInRecoverKv, int remoteLogPrefetchNumInRecoverKv, @@ -85,7 +80,6 @@ private DefaultSnapshotContext( this.kvSnapshotDataDownloader = kvSnapshotDataDownloader; this.kvSnapshotIntervalMs = kvSnapshotIntervalMs; this.writeBufferSizeInBytes = writeBufferSizeInBytes; - this.remoteKvDir = remoteKvDir; this.completedSnapshotHandleStore = completedSnapshotHandleStore; this.maxFetchLogSizeInRecoverKv = maxFetchLogSizeInRecoverKv; @@ -107,21 +101,23 @@ public static DefaultSnapshotContext create( kvSnapshotResource.getKvSnapshotDataDownloader(), conf.get(ConfigOptions.KV_SNAPSHOT_INTERVAL).toMillis(), (int) conf.get(ConfigOptions.REMOTE_FS_WRITE_BUFFER_SIZE).getBytes(), - FlussPaths.remoteKvDir(conf), new ZooKeeperCompletedSnapshotHandleStore(zkClient), (int) conf.get(ConfigOptions.KV_RECOVER_LOG_RECORD_BATCH_MAX_SIZE).getBytes(), conf.get(ConfigOptions.KV_RECOVERY_REMOTE_LOG_PREFETCH_NUM), conf.get(ConfigOptions.KV_RECOVERY_REMOTE_LOG_DOWNLOAD_THREADS)); } + @Override public ZooKeeperClient getZooKeeperClient() { return zooKeeperClient; } + @Override public ExecutorService getAsyncOperationsThreadPool() { return asyncOperationsThreadPool; } + @Override public KvSnapshotDataUploader getSnapshotDataUploader() { return kvSnapshotDataUploader; } @@ -131,6 +127,7 @@ public KvSnapshotDataDownloader getSnapshotDataDownloader() { return kvSnapshotDataDownloader; } + @Override public ScheduledExecutorService getSnapshotScheduler() { return snapshotScheduler; } @@ -150,10 +147,6 @@ public int getSnapshotFsWriteBufferSize() { return writeBufferSizeInBytes; } - public FsPath getRemoteKvDir() { - return remoteKvDir; - } - @Override public FunctionWithException getLatestCompletedSnapshotProvider() { diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotContext.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotContext.java index 18c32d5312..1093c5e11b 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotContext.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotContext.java @@ -18,7 +18,6 @@ package org.apache.fluss.server.kv.snapshot; import org.apache.fluss.config.ConfigOptions; -import org.apache.fluss.fs.FsPath; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.server.zk.ZooKeeperClient; import org.apache.fluss.utils.function.FunctionWithException; @@ -55,9 +54,6 @@ public interface SnapshotContext { /** Get the size of the write buffer for writing the kv snapshot file to remote filesystem. */ int getSnapshotFsWriteBufferSize(); - /** Get the remote root path to store kv snapshot files. */ - FsPath getRemoteKvDir(); - /** * Get the provider of latest CompletedSnapshot for a table bucket. When no completed snapshot * exists, the CompletedSnapshot provided will be null. diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/DefaultRemoteLogStorage.java b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/DefaultRemoteLogStorage.java index 56e8e24091..f63109016b 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/DefaultRemoteLogStorage.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/DefaultRemoteLogStorage.java @@ -66,24 +66,14 @@ public class DefaultRemoteLogStorage implements RemoteLogStorage { private static final int READ_BUFFER_SIZE = 16 * 1024; - private final FsPath remoteLogDir; - private final FileSystem fileSystem; private final ExecutorService ioExecutor; private final int writeBufferSize; - public DefaultRemoteLogStorage(Configuration conf, ExecutorService ioExecutor) - throws IOException { - this.remoteLogDir = FlussPaths.remoteLogDir(conf); - this.fileSystem = remoteLogDir.getFileSystem(); + public DefaultRemoteLogStorage(Configuration conf, ExecutorService ioExecutor) { this.writeBufferSize = (int) conf.get(ConfigOptions.REMOTE_FS_WRITE_BUFFER_SIZE).getBytes(); this.ioExecutor = ioExecutor; } - @Override - public FsPath getRemoteLogDir() { - return remoteLogDir; - } - /** * Copy log segments to remote path. * @@ -142,6 +132,7 @@ public void deleteLogSegmentFiles(RemoteLogSegment remoteLogSegment) throws RemoteStorageException { LOG.debug("Deleting log segment and indexes for : {}", remoteLogSegment); try { + FsPath remoteLogDir = remoteLogSegment.remoteLogDir(); FsPath segmentDir = remoteLogSegmentDir(remoteLogDir, remoteLogSegment); long baseOffset = remoteLogSegment.remoteLogStartOffset(); FsPath logFile = remoteLogSegmentFile(segmentDir, baseOffset); @@ -155,7 +146,7 @@ public void deleteLogSegmentFiles(RemoteLogSegment remoteLogSegment) // delete dir at last for (FsPath path : Arrays.asList(logFile, offsetIndex, timeIndex, writerSnapshot, segmentDir)) { - fileSystem.delete(path, false); + remoteLogDir.getFileSystem().delete(path, false); } LOG.debug("Successful delete log segment and indexes for : {}", remoteLogSegment); } catch (IOException e) { @@ -167,6 +158,8 @@ public void deleteLogSegmentFiles(RemoteLogSegment remoteLogSegment) @Override public InputStream fetchIndex(RemoteLogSegment remoteLogSegment, IndexType indexType) throws RemoteStorageException { + FsPath remoteLogDir = remoteLogSegment.remoteLogDir(); + FsPath remoteLogSegmentIndexFile; if (indexType == IndexType.WRITER_ID_SNAPSHOT) { remoteLogSegmentIndexFile = @@ -179,7 +172,7 @@ public InputStream fetchIndex(RemoteLogSegment remoteLogSegment, IndexType index } try { - return fileSystem.open(remoteLogSegmentIndexFile); + return remoteLogDir.getFileSystem().open(remoteLogSegmentIndexFile); } catch (IOException e) { throw new RemoteStorageException( "Failed to fetch index file type: " @@ -193,10 +186,11 @@ public InputStream fetchIndex(RemoteLogSegment remoteLogSegment, IndexType index @Override public InputStream fetchLogData(RemoteLogSegment remoteLogSegment) throws RemoteStorageException { + FsPath remoteLogDir = remoteLogSegment.remoteLogDir(); FsPath segmentDir = remoteLogSegmentDir(remoteLogDir, remoteLogSegment); FsPath logFile = remoteLogSegmentFile(segmentDir, remoteLogSegment.remoteLogStartOffset()); try { - return fileSystem.open(logFile); + return remoteLogDir.getFileSystem().open(logFile); } catch (IOException e) { throw new RemoteStorageException("Failed to fetch log data from path: " + logFile, e); } @@ -208,7 +202,7 @@ public RemoteLogManifest readRemoteLogManifestSnapshot(FsPath remoteLogManifestP FSDataInputStream inputStream = null; ByteArrayOutputStream outputStream = null; try { - inputStream = fileSystem.open(remoteLogManifestPath); + inputStream = remoteLogManifestPath.getFileSystem().open(remoteLogManifestPath); outputStream = new ByteArrayOutputStream(); IOUtils.copyBytes(inputStream, outputStream, false); return RemoteLogManifest.fromJsonBytes(outputStream.toByteArray()); @@ -228,7 +222,7 @@ public void deleteRemoteLogManifestSnapshot(FsPath remoteLogManifestPath) throws RemoteStorageException { LOG.debug("Deleting remote log segment manifest: {}", remoteLogManifestPath); try { - fileSystem.delete(remoteLogManifestPath, false); + remoteLogManifestPath.getFileSystem().delete(remoteLogManifestPath, false); LOG.debug("Successful delete log segment manifest: {}", remoteLogManifestPath); } catch (IOException e) { throw new RemoteStorageException( @@ -240,6 +234,7 @@ public void deleteRemoteLogManifestSnapshot(FsPath remoteLogManifestPath) @Override public FsPath writeRemoteLogManifestSnapshot(RemoteLogManifest manifest) throws RemoteStorageException { + FsPath remoteLogDir = manifest.getRemoteLogDir(); FsPath manifestFile = FlussPaths.remoteLogManifestFile( FlussPaths.remoteLogTabletDir( @@ -262,11 +257,13 @@ public FsPath writeRemoteLogManifestSnapshot(RemoteLogManifest manifest) } @Override - public void deleteTableBucket(PhysicalTablePath physicalTablePath, TableBucket tableBucket) + public void deleteTableBucket( + FsPath remoteLogDir, PhysicalTablePath physicalTablePath, TableBucket tableBucket) throws RemoteStorageException { FsPath remoteLogTabletDir = FlussPaths.remoteLogTabletDir(remoteLogDir, physicalTablePath, tableBucket); try { + FileSystem fileSystem = remoteLogDir.getFileSystem(); if (fileSystem.exists(remoteLogTabletDir)) { fileSystem.delete(remoteLogTabletDir, true); } @@ -335,8 +332,9 @@ private List> createUploadFutures( } private FsPath createRemoteLogSegmentDir(RemoteLogSegment remoteLogSegment) throws IOException { + FsPath remoteLogDir = remoteLogSegment.remoteLogDir(); FsPath remoteLogSegmentDir = remoteLogSegmentDir(remoteLogDir, remoteLogSegment); - fileSystem.mkdirs(remoteLogSegmentDir); + remoteLogDir.getFileSystem().mkdirs(remoteLogSegmentDir); return remoteLogSegmentDir; } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/LogTieringTask.java b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/LogTieringTask.java index c84aff7188..26535557b2 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/LogTieringTask.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/LogTieringTask.java @@ -291,6 +291,7 @@ private long copyLogSegmentFilesToRemote( .remoteLogEndOffset(segmentEndOffset) .maxTimestamp(segment.maxTimestampSoFar()) .segmentSizeInBytes(sizeInBytes) + .remoteLogDir(remoteLog.getRemoteLogDir()) .build(); try { remoteLogStorage.copyLogSegmentFiles(copyRemoteLogSegment, logSegmentFiles); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogManager.java b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogManager.java index 4f27c2872a..c96a37e681 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogManager.java @@ -34,6 +34,7 @@ import org.apache.fluss.server.storage.LocalDiskManager; import org.apache.fluss.server.zk.ZooKeeperClient; import org.apache.fluss.server.zk.data.RemoteLogManifestHandle; +import org.apache.fluss.utils.FlussPaths; import org.apache.fluss.utils.IOUtils; import org.apache.fluss.utils.clock.Clock; import org.apache.fluss.utils.concurrent.ExecutorThreadFactory; @@ -140,10 +141,6 @@ public RemoteLogStorage getRemoteLogStorage() { return remoteLogStorage; } - public FsPath remoteLogDir() { - return remoteLogStorage.getRemoteLogDir(); - } - /** Register the replica to the remote log manager. */ public void registerReplica(Replica replica) throws Exception { if (remoteDisabled()) { @@ -152,8 +149,10 @@ public void registerReplica(Replica replica) throws Exception { TableBucket tableBucket = replica.getTableBucket(); PhysicalTablePath physicalTablePath = replica.getPhysicalTablePath(); LogTablet log = replica.getLogTablet(); + FsPath remoteLogDir = FlussPaths.remoteLogDir(replica.getRemoteDataDir()); RemoteLogTablet remoteLog = - new RemoteLogTablet(physicalTablePath, tableBucket, replica.getLogTTLMs()); + new RemoteLogTablet( + physicalTablePath, tableBucket, replica.getLogTTLMs(), remoteLogDir); Optional remoteLogManifestHandleOpt = zkClient.getRemoteLogManifestHandle(tableBucket); if (remoteLogManifestHandleOpt.isPresent()) { @@ -162,6 +161,20 @@ public void registerReplica(Replica replica) throws Exception { RemoteLogManifest manifest = remoteLogStorage.readRemoteLogManifestSnapshot( remoteLogManifestHandleOpt.get().getRemoteLogManifestPath()); + + // If the RemoteLogManifest does not include remoteLogDir, it means the manifest was + // generated by an old version that does not support remote.data.dirs. + // We set remoteLogDir manually here, so subsequent usage will be safe to directly use + // it. + if (manifest.getRemoteLogDir() == null) { + LOG.info( + "RemoteLogManifest loaded from old version without remoteLogDir, " + + "setting remoteLogDir to {} for bucket {}", + remoteLogDir, + manifest.getTableBucket()); + manifest = manifest.newManifest(remoteLogDir); + } + remoteLog.loadRemoteLogManifest(manifest); } remoteLog.getRemoteLogEndOffset().ifPresent(log::updateRemoteLogEndOffset); @@ -233,8 +246,9 @@ public void stopReplica(Replica replica, boolean deleteRemote) { if (deleteRemote) { LOG.info("Deleting the remote log segments for table-bucket: {}", tb); + FsPath remoteLogDir = FlussPaths.remoteLogDir(replica.getRemoteDataDir()); // delete the remote log of the table bucket. - deleteRemoteLog(physicalTablePath, tb); + deleteRemoteLog(remoteLogDir, physicalTablePath, tb); } } @@ -287,11 +301,12 @@ private boolean remoteDisabled() { *

Note: the zk path for {@link RemoteLogManifestHandle} will be deleted by coordinator while * table delete. */ - private void deleteRemoteLog(PhysicalTablePath physicalTablePath, TableBucket tableBucket) { + private void deleteRemoteLog( + FsPath remoteLogDir, PhysicalTablePath physicalTablePath, TableBucket tableBucket) { // delete the file in remote storage. try { // TODO: maybe need to optimize to delete on specific file path - remoteLogStorage.deleteTableBucket(physicalTablePath, tableBucket); + remoteLogStorage.deleteTableBucket(remoteLogDir, physicalTablePath, tableBucket); } catch (RemoteStorageException e) { LOG.error( "Error occurred while deleting remote log for table-bucket: {}", diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogStorage.java b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogStorage.java index 6e1de16cf3..74de4f8564 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogStorage.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogStorage.java @@ -67,13 +67,6 @@ public static String getFileSuffix(IndexType indexType) { } } - /** - * Returns the remote log directory. - * - * @return the remote log directory. - */ - FsPath getRemoteLogDir(); - /** * Copies the given {@link LogSegmentFiles} provided for the given {@link RemoteLogSegment}. * This includes log segment and its auxiliary indexes like offset index and writer id snapshot @@ -165,11 +158,13 @@ void deleteRemoteLogManifestSnapshot(FsPath remoteLogManifestPath) * Deletes the remote log data and metadata from remote storage for the input table bucket as * this table have been deleted. * + * @param remoteLogDir the remote log directory for the table bucket. * @param physicalTablePath the physical table path. * @param tableBucket the table bucket. * @throws RemoteStorageException if there are any errors while delete remote log data and * metadata. */ - void deleteTableBucket(PhysicalTablePath physicalTablePath, TableBucket tableBucket) + void deleteTableBucket( + FsPath remoteLogDir, PhysicalTablePath physicalTablePath, TableBucket tableBucket) throws RemoteStorageException; } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogTablet.java b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogTablet.java index 9f0ae6e949..3310edf76e 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogTablet.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogTablet.java @@ -18,6 +18,7 @@ package org.apache.fluss.server.log.remote; import org.apache.fluss.annotation.VisibleForTesting; +import org.apache.fluss.fs.FsPath; import org.apache.fluss.metadata.PhysicalTablePath; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metrics.MetricNames; @@ -57,6 +58,8 @@ public class RemoteLogTablet { private final PhysicalTablePath physicalTablePath; + private final FsPath remoteLogDir; + /** * It contains all the segment-id to {@link RemoteLogSegment} mappings which did not delete in * remote storage. @@ -103,12 +106,17 @@ public class RemoteLogTablet { private volatile boolean closed = false; public RemoteLogTablet( - PhysicalTablePath physicalTablePath, TableBucket tableBucket, long ttlMs) { + PhysicalTablePath physicalTablePath, + TableBucket tableBucket, + long ttlMs, + FsPath remoteLogDir) { this.tableBucket = tableBucket; this.physicalTablePath = physicalTablePath; + this.remoteLogDir = remoteLogDir; this.ttlMs = ttlMs; this.currentManifest = - new RemoteLogManifest(physicalTablePath, tableBucket, new ArrayList<>()); + new RemoteLogManifest( + physicalTablePath, tableBucket, new ArrayList<>(), remoteLogDir); reset(); } @@ -267,6 +275,10 @@ public OptionalLong getRemoteLogEndOffset() { : OptionalLong.of(remoteLogEndOffset); } + public FsPath getRemoteLogDir() { + return remoteLogDir; + } + /** * Gets the snapshot of current remote log segment manifest. The snapshot including the exists * remoteLogSegment already committed. @@ -358,7 +370,8 @@ public void addAndDeleteLogSegments( new RemoteLogManifest( physicalTablePath, tableBucket, - new ArrayList<>(idToRemoteLogSegment.values())); + new ArrayList<>(idToRemoteLogSegment.values()), + remoteLogDir); }); } 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 f3d1ff76af..c1934fd8aa 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 @@ -161,6 +161,7 @@ public final class Replica { private final PhysicalTablePath physicalPath; private final TableBucket tableBucket; + private final String remoteDataDir; private final LogManager logManager; private final LogTablet logTablet; @@ -247,6 +248,7 @@ public Replica( FatalErrorHandler fatalErrorHandler, BucketMetricGroup bucketMetricGroup, TableInfo tableInfo, + String remoteDataDir, Clock clock, RemoteLogManager remoteLogManager, ScannerManager scannerManager) @@ -280,6 +282,7 @@ public Replica( this.logTablet = createLog(dataDir, lazyHighWatermarkCheckpoint); this.logTablet.updateIsDataLakeEnabled(tableConfig.isDataLakeEnabled()); + this.remoteDataDir = remoteDataDir; this.clock = clock; this.remoteLogManager = remoteLogManager; this.scannerManager = checkNotNull(scannerManager, "scannerManager"); @@ -416,6 +419,10 @@ public LogFormat getLogFormat() { return logFormat; } + public String getRemoteDataDir() { + return remoteDataDir; + } + public void makeLeader(NotifyLeaderAndIsrData data) throws IOException { boolean leaderHWIncremented = inWriteLock( @@ -989,9 +996,10 @@ private void startPeriodicKvSnapshot(@Nullable CompletedSnapshot completedSnapsh // instead of a separate class Supplier bucketLeaderEpochSupplier = () -> leaderEpoch; Supplier coordinatorEpochSupplier = () -> coordinatorEpoch; + + FsPath remoteKvDir = FlussPaths.remoteKvDir(remoteDataDir); FsPath remoteKvTabletDir = - FlussPaths.remoteKvTabletDir( - snapshotContext.getRemoteKvDir(), physicalPath, tableBucket); + FlussPaths.remoteKvTabletDir(remoteKvDir, physicalPath, tableBucket); kvTabletSnapshotTarget = new KvTabletSnapshotTarget( diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java index 88d5840e5a..a0cd92bc7d 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java @@ -38,6 +38,7 @@ import org.apache.fluss.metadata.LogFormat; import org.apache.fluss.metadata.PhysicalTablePath; import org.apache.fluss.metadata.Schema; +import org.apache.fluss.metadata.SchemaInfo; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TableInfo; import org.apache.fluss.metadata.TablePath; @@ -112,6 +113,8 @@ import org.apache.fluss.server.utils.FatalErrorHandler; import org.apache.fluss.server.zk.ZooKeeperClient; import org.apache.fluss.server.zk.data.LeaderAndIsr; +import org.apache.fluss.server.zk.data.PartitionRegistration; +import org.apache.fluss.server.zk.data.TableRegistration; import org.apache.fluss.server.zk.data.lake.LakeTableSnapshot; import org.apache.fluss.utils.FileUtils; import org.apache.fluss.utils.FlussPaths; @@ -148,7 +151,9 @@ import java.util.stream.Stream; import static org.apache.fluss.config.ConfigOptions.KV_FORMAT_VERSION_2; -import static org.apache.fluss.server.TabletManagerBase.getTableInfo; +import static org.apache.fluss.server.TabletManagerBase.getPartitionRegistration; +import static org.apache.fluss.server.TabletManagerBase.getSchemaInfo; +import static org.apache.fluss.server.TabletManagerBase.getTableRegistration; import static org.apache.fluss.utils.Preconditions.checkArgument; import static org.apache.fluss.utils.Preconditions.checkNotNull; import static org.apache.fluss.utils.Preconditions.checkState; @@ -1644,11 +1649,10 @@ private boolean canFetchFromRemoteLog(Replica replica, long fetchOffset) { remoteLogManager.lookupPositionForOffset( remoteLogSegmentList.get(0), fetchOffset); PhysicalTablePath physicalTablePath = replica.getPhysicalTablePath(); + FsPath remoteLogDir = FlussPaths.remoteLogDir(replica.getRemoteDataDir()); FsPath remoteLogTabletDir = FlussPaths.remoteLogTabletDir( - remoteLogManager.remoteLogDir(), - physicalTablePath, - replica.getTableBucket()); + remoteLogDir, physicalTablePath, replica.getTableBucket()); return new RemoteLogFetchInfo( remoteLogTabletDir.toString(), physicalTablePath.getPartitionName(), @@ -2005,7 +2009,9 @@ private StopReplicaResultForBucket stopReplica( replicaToDelete, deleteRemote && replicaToDelete.isLeader()); if (deleteRemote && replicaToDelete.isLeader()) { kvManager.deleteRemoteKvSnapshot( - replicaToDelete.getPhysicalTablePath(), replicaToDelete.getTableBucket()); + replicaToDelete.getRemoteDataDir(), + replicaToDelete.getPhysicalTablePath(), + replicaToDelete.getTableBucket()); } } @@ -2119,6 +2125,20 @@ private void dropEmptyTableOrPartitionDir(Path dir, long id, String dirType) { } } + private String getRemoteDataDir( + TableBucket tb, + PhysicalTablePath physicalTablePath, + TableRegistration tableRegistration) + throws Exception { + if (tb.getPartitionId() != null) { + PartitionRegistration partitionRegistration = + getPartitionRegistration(zkClient, physicalTablePath); + return partitionRegistration.getRemoteDataDir(); + } else { + return tableRegistration.remoteDataDir; + } + } + protected Optional maybeCreateReplica(NotifyLeaderAndIsrData data) { Optional replicaOpt = Optional.empty(); try { @@ -2127,7 +2147,12 @@ protected Optional maybeCreateReplica(NotifyLeaderAndIsrData data) { if (hostedReplica instanceof NoneReplica) { PhysicalTablePath physicalTablePath = data.getPhysicalTablePath(); TablePath tablePath = physicalTablePath.getTablePath(); - TableInfo tableInfo = getTableInfo(zkClient, tablePath); + + TableRegistration tableRegistration = getTableRegistration(zkClient, tablePath); + SchemaInfo schemaInfo = getSchemaInfo(zkClient, tablePath); + TableInfo tableInfo = tableRegistration.toTableInfo(tablePath, schemaInfo); + + String remoteDataDir = getRemoteDataDir(tb, physicalTablePath, tableRegistration); boolean isKvTable = tableInfo.hasPrimaryKey(); Optional existingLogTabletOpt = logManager.getLog(tb); @@ -2161,6 +2186,7 @@ protected Optional maybeCreateReplica(NotifyLeaderAndIsrData data) { fatalErrorHandler, bucketMetricGroup, tableInfo, + remoteDataDir, clock, remoteLogManager, scannerManager); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvSnapshotMultipleDirsITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvSnapshotMultipleDirsITCase.java new file mode 100644 index 0000000000..ae79c1f97b --- /dev/null +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvSnapshotMultipleDirsITCase.java @@ -0,0 +1,268 @@ +/* + * 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.kv.snapshot; + +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.fs.FileStatus; +import org.apache.fluss.fs.FsPath; +import org.apache.fluss.metadata.PartitionSpec; +import org.apache.fluss.metadata.PhysicalTablePath; +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metadata.TableDescriptor; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.record.KvRecordBatch; +import org.apache.fluss.rpc.gateway.TabletServerGateway; +import org.apache.fluss.rpc.messages.PutKvRequest; +import org.apache.fluss.server.testutils.FlussClusterExtension; +import org.apache.fluss.server.testutils.RpcMessageTestUtils; +import org.apache.fluss.server.zk.ZooKeeperClient; +import org.apache.fluss.server.zk.data.PartitionRegistration; +import org.apache.fluss.server.zk.data.TableRegistration; +import org.apache.fluss.types.DataTypes; +import org.apache.fluss.utils.FlussPaths; +import org.apache.fluss.utils.types.Tuple2; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +import static org.apache.fluss.record.TestData.DATA1_SCHEMA_PK; +import static org.apache.fluss.server.testutils.RpcMessageTestUtils.createPartition; +import static org.apache.fluss.server.testutils.RpcMessageTestUtils.createTable; +import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newPutKvRequest; +import static org.apache.fluss.testutils.DataTestUtils.genKvRecordBatch; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * ITCase for verifying KV snapshot can be uploaded to multiple remote directories with round-robin + * distribution. + */ +public class KvSnapshotMultipleDirsITCase { + + private static final List REMOTE_DIR_NAMES = Arrays.asList("dir1", "dir2", "dir3"); + + @RegisterExtension + public static final FlussClusterExtension FLUSS_CLUSTER_EXTENSION = + FlussClusterExtension.builder() + .setNumOfTabletServers(3) + .setClusterConf(initConfig()) + .setRemoteDirNames(REMOTE_DIR_NAMES) + .build(); + + private ZooKeeperClient zkClient; + + @BeforeEach + void setup() { + zkClient = FLUSS_CLUSTER_EXTENSION.getZooKeeperClient(); + } + + @Test + void testKvSnapshotToMultipleDirsForNonPartitionedTable() throws Exception { + // Create multiple tables (more than number of remote dirs) to ensure round-robin + // distribution. Each table's KV snapshot should be uploaded to different remote dirs. + int tableCount = 6; + List tablePaths = new ArrayList<>(); + List tableIds = new ArrayList<>(); + + // Create tables with primary key (KV tables) + TableDescriptor tableDescriptor = + TableDescriptor.builder() + .schema(DATA1_SCHEMA_PK) + .distributedBy(1) // Single bucket for simpler verification + .build(); + + for (int i = 0; i < tableCount; i++) { + TablePath tablePath = TablePath.of("test_db", String.format("kv_snapshot_table_%d", i)); + tablePaths.add(tablePath); + long tableId = + RpcMessageTestUtils.createTable( + FLUSS_CLUSTER_EXTENSION, tablePath, tableDescriptor); + tableIds.add(tableId); + } + + // Write data to each table to trigger KV snapshot + for (int t = 0; t < tableCount; t++) { + TableBucket tb = new TableBucket(tableIds.get(t), 0); + FLUSS_CLUSTER_EXTENSION.waitUntilAllReplicaReady(tb); + + int leaderId = FLUSS_CLUSTER_EXTENSION.waitAndGetLeader(tb); + TabletServerGateway leaderGateway = + FLUSS_CLUSTER_EXTENSION.newTabletServerClientForNode(leaderId); + + // Write KV data to trigger snapshot + KvRecordBatch kvRecordBatch = + genKvRecordBatch( + Tuple2.of("k1", new Object[] {1, "v1"}), + Tuple2.of("k2", new Object[] {2, "v2"})); + + PutKvRequest putKvRequest = newPutKvRequest(tableIds.get(t), 0, 1, kvRecordBatch); + leaderGateway.putKv(putKvRequest).get(); + } + + // Wait for all tables' snapshots to be completed + for (int t = 0; t < tableCount; t++) { + TableBucket tb = new TableBucket(tableIds.get(t), 0); + // Wait for snapshot 0 to be finished + FLUSS_CLUSTER_EXTENSION.waitUntilSnapshotFinished(tb, 0); + } + + // Collect the remote data directories used by each table + Set usedRemoteDataDirs = new HashSet<>(); + for (int t = 0; t < tableCount; t++) { + Optional tableOpt = zkClient.getTable(tablePaths.get(t)); + assertThat(tableOpt).isPresent(); + TableRegistration table = tableOpt.get(); + + assertThat(table.remoteDataDir).isNotNull(); + usedRemoteDataDirs.add(table.remoteDataDir); + + // Verify the remote KV snapshot files actually exist + FsPath remoteKvDir = FlussPaths.remoteKvDir(table.remoteDataDir); + TableBucket tb = new TableBucket(tableIds.get(t), 0); + FsPath remoteKvTabletDir = + FlussPaths.remoteKvTabletDir( + remoteKvDir, PhysicalTablePath.of(tablePaths.get(t)), tb); + assertThat(remoteKvTabletDir.getFileSystem().exists(remoteKvTabletDir)).isTrue(); + FileStatus[] fileStatuses = + remoteKvTabletDir.getFileSystem().listStatus(remoteKvTabletDir); + assertThat(fileStatuses).isNotEmpty(); + } + + // All configured remote dirs should be used due to round-robin distribution + assertThat(usedRemoteDataDirs).hasSameSizeAs(REMOTE_DIR_NAMES); + } + + @Test + void testKvSnapshotToMultipleDirsForPartitionedTable() throws Exception { + // Create a partitioned table and add multiple partitions (more than number of remote dirs) + // to ensure round-robin distribution. Each partition's KV snapshot should be uploaded to + // different remote dirs. + int partitionCount = 6; + TablePath tablePath = TablePath.of("test_db", "partitioned_kv_snapshot_table"); + + // Create partitioned table with primary key + Schema schema = + Schema.newBuilder() + .column("a", DataTypes.INT()) + .column("b", DataTypes.STRING()) + .primaryKey("a", "b") + .build(); + + TableDescriptor tableDescriptor = + TableDescriptor.builder() + .schema(schema) + .distributedBy(1) // Single bucket for simpler verification + .partitionedBy("b") + .build(); + + long tableId = createTable(FLUSS_CLUSTER_EXTENSION, tablePath, tableDescriptor); + + // Create partitions + List partitionNames = new ArrayList<>(); + for (int i = 0; i < partitionCount; i++) { + String partitionName = "p" + i; + partitionNames.add(partitionName); + PartitionSpec partitionSpec = + new PartitionSpec(Collections.singletonMap("b", partitionName)); + createPartition(FLUSS_CLUSTER_EXTENSION, tablePath, partitionSpec, false); + } + + // Get partition IDs from ZK + Map partitionRegistrations = + zkClient.getPartitionRegistrations(tablePath); + assertThat(partitionRegistrations).hasSize(partitionCount); + + // Wait for all partitions to be ready and write data to trigger KV snapshot + for (int p = 0; p < partitionCount; p++) { + String partitionName = partitionNames.get(p); + Long partitionId = partitionRegistrations.get(partitionName).getPartitionId(); + TableBucket tb = new TableBucket(tableId, partitionId, 0); + FLUSS_CLUSTER_EXTENSION.waitUntilAllReplicaReady(tb); + + int leaderId = FLUSS_CLUSTER_EXTENSION.waitAndGetLeader(tb); + TabletServerGateway leaderGateway = + FLUSS_CLUSTER_EXTENSION.newTabletServerClientForNode(leaderId); + + // Write KV data to trigger snapshot + KvRecordBatch kvRecordBatch = + genKvRecordBatch( + Tuple2.of("k1", new Object[] {1, "v1"}), + Tuple2.of("k2", new Object[] {2, "v2"})); + + PutKvRequest putKvRequest = newPutKvRequest(tableId, partitionId, 0, 1, kvRecordBatch); + leaderGateway.putKv(putKvRequest).get(); + } + + // Wait for all partitions' snapshots to be completed + for (int p = 0; p < partitionCount; p++) { + Long partitionId = partitionRegistrations.get(partitionNames.get(p)).getPartitionId(); + TableBucket tb = new TableBucket(tableId, partitionId, 0); + // Wait for snapshot 0 to be finished + FLUSS_CLUSTER_EXTENSION.waitUntilSnapshotFinished(tb, 0); + } + + // Collect the remote data directories used by each partition + Set usedRemoteDataDirs = new HashSet<>(); + for (int p = 0; p < partitionCount; p++) { + String partitionName = partitionNames.get(p); + Long partitionId = partitionRegistrations.get(partitionName).getPartitionId(); + Optional partitionOpt = + zkClient.getPartition(tablePath, partitionName); + assertThat(partitionOpt).isPresent(); + PartitionRegistration partition = partitionOpt.get(); + + assertThat(partition.getRemoteDataDir()).isNotNull(); + usedRemoteDataDirs.add(partition.getRemoteDataDir()); + + // Verify the remote KV snapshot files actually exist + FsPath remoteKvDir = FlussPaths.remoteKvDir(partition.getRemoteDataDir()); + TableBucket tb = new TableBucket(tableId, partitionId, 0); + FsPath remoteKvTabletDir = + FlussPaths.remoteKvTabletDir( + remoteKvDir, PhysicalTablePath.of(tablePath, partitionName), tb); + assertThat(remoteKvTabletDir.getFileSystem().exists(remoteKvTabletDir)).isTrue(); + FileStatus[] fileStatuses = + remoteKvTabletDir.getFileSystem().listStatus(remoteKvTabletDir); + assertThat(fileStatuses).isNotEmpty(); + } + + // All configured remote dirs should be used due to round-robin distribution + assertThat(usedRemoteDataDirs).hasSameSizeAs(REMOTE_DIR_NAMES); + } + + private static Configuration initConfig() { + Configuration conf = new Configuration(); + conf.setInt(ConfigOptions.DEFAULT_BUCKET_NUMBER, 1); + conf.setInt(ConfigOptions.DEFAULT_REPLICATION_FACTOR, 3); + // Set a shorter interval for testing purpose + conf.set(ConfigOptions.KV_SNAPSHOT_INTERVAL, Duration.ofSeconds(1)); + return conf; + } +} diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/DefaultRemoteLogStorageTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/DefaultRemoteLogStorageTest.java index a8ec544510..9a9be1576b 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/DefaultRemoteLogStorageTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/DefaultRemoteLogStorageTest.java @@ -185,13 +185,14 @@ void testDeleteTable(boolean partitionTable) throws Exception { File remoteDirForBucket = new File( FlussPaths.remoteLogTabletDir( - remoteLogStorageManager.getRemoteLogDir(), + remoteLogSegment.remoteLogDir(), physicalTablePath, tableBucket) .toString()); assertThat(remoteDirForBucket.exists()).isTrue(); - remoteLogStorageManager.deleteTableBucket(physicalTablePath, tableBucket); + remoteLogStorageManager.deleteTableBucket( + remoteLogSegment.remoteLogDir(), physicalTablePath, tableBucket); assertThat(remoteDirForBucket.exists()).isFalse(); assertThatThrownBy( () -> @@ -205,7 +206,7 @@ private File getTestingRemoteLogSegmentDir(RemoteLogSegment remoteLogSegment) { return new File( FlussPaths.remoteLogSegmentDir( FlussPaths.remoteLogTabletDir( - remoteLogStorageManager.getRemoteLogDir(), + remoteLogSegment.remoteLogDir(), remoteLogSegment.physicalTablePath(), remoteLogSegment.tableBucket()), remoteLogSegment.remoteLogSegmentId()) diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogITCase.java index 872682ad26..253b60335f 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogITCase.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogITCase.java @@ -20,10 +20,13 @@ import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; import org.apache.fluss.config.MemorySize; +import org.apache.fluss.fs.FileStatus; import org.apache.fluss.fs.FileSystem; import org.apache.fluss.fs.FsPath; import org.apache.fluss.metadata.DataLakeFormat; +import org.apache.fluss.metadata.PartitionSpec; import org.apache.fluss.metadata.PhysicalTablePath; +import org.apache.fluss.metadata.Schema; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TableDescriptor; import org.apache.fluss.metadata.TablePath; @@ -39,27 +42,39 @@ import org.apache.fluss.server.replica.Replica; import org.apache.fluss.server.tablet.TabletServer; import org.apache.fluss.server.testutils.FlussClusterExtension; +import org.apache.fluss.server.testutils.RpcMessageTestUtils; +import org.apache.fluss.server.zk.ZooKeeperClient; +import org.apache.fluss.server.zk.data.PartitionRegistration; +import org.apache.fluss.server.zk.data.TableRegistration; import org.apache.fluss.utils.FlussPaths; import org.apache.fluss.utils.clock.ManualClock; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.Comparator; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Optional; +import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.stream.Collectors; import static org.apache.fluss.record.TestData.DATA1; import static org.apache.fluss.record.TestData.DATA1_SCHEMA; +import static org.apache.fluss.record.TestData.DATA1_SCHEMA_PK; import static org.apache.fluss.record.TestData.DATA1_TABLE_DESCRIPTOR; import static org.apache.fluss.record.TestData.DATA1_TABLE_PATH; import static org.apache.fluss.server.testutils.RpcMessageTestUtils.assertProduceLogResponse; +import static org.apache.fluss.server.testutils.RpcMessageTestUtils.createPartition; import static org.apache.fluss.server.testutils.RpcMessageTestUtils.createTable; import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newAlterTableRequest; import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newDropTableRequest; @@ -74,14 +89,24 @@ public class RemoteLogITCase { private static final ManualClock MANUAL_CLOCK = new ManualClock(System.currentTimeMillis()); + private static final List REMOTE_DIR_NAMES = Arrays.asList("dir1", "dir2", "dir3"); + @RegisterExtension public static final FlussClusterExtension FLUSS_CLUSTER_EXTENSION = FlussClusterExtension.builder() .setNumOfTabletServers(3) .setClusterConf(initConfig()) .setClock(MANUAL_CLOCK) + .setRemoteDirNames(REMOTE_DIR_NAMES) .build(); + private ZooKeeperClient zkClient; + + @BeforeEach + void setup() { + zkClient = FLUSS_CLUSTER_EXTENSION.getZooKeeperClient(); + } + private TableBucket setupTableBucket() throws Exception { long tableId = createTable(FLUSS_CLUSTER_EXTENSION, DATA1_TABLE_PATH, DATA1_TABLE_DESCRIPTOR); @@ -139,9 +164,7 @@ public void remoteLogMiscTest() throws Exception { // test create: verify remote log created FsPath fsPath = FlussPaths.remoteLogTabletDir( - tabletServer.getReplicaManager().getRemoteLogManager().remoteLogDir(), - PhysicalTablePath.of(DATA1_TABLE_PATH), - tb); + manifest.getRemoteLogDir(), PhysicalTablePath.of(DATA1_TABLE_PATH), tb); FileSystem fileSystem = fsPath.getFileSystem(); assertThat(fileSystem.exists(fsPath)).isTrue(); assertThat(fileSystem.listStatus(fsPath).length).isGreaterThan(0); @@ -224,6 +247,205 @@ void testFollowerFetchAlreadyMoveToRemoteLog(boolean withWriterId) throws Except FLUSS_CLUSTER_EXTENSION.waitUntilReplicaExpandToIsr(tb, follower); } + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testRemoteLogTieredToMultipleDirsForNonPartitionedTable(boolean isPrimaryTable) + throws Exception { + // Create multiple tables (more than number of remote dirs) to ensure round-robin + // distribution. Each table's remote log should be uploaded to different remote dirs. + int tableCount = 6; + List tablePaths = new ArrayList<>(); + List tableIds = new ArrayList<>(); + + // Create tables + Schema schema = isPrimaryTable ? DATA1_SCHEMA_PK : DATA1_SCHEMA; + TableDescriptor tableDescriptor = + TableDescriptor.builder() + .schema(schema) + .distributedBy(1) // Single bucket for simpler verification + .build(); + + for (int i = 0; i < tableCount; i++) { + TablePath tablePath = + TablePath.of( + "test_db", + String.format("remote_%s_table_%d", isPrimaryTable ? "kv" : "log", i)); + tablePaths.add(tablePath); + long tableId = + RpcMessageTestUtils.createTable( + FLUSS_CLUSTER_EXTENSION, tablePath, tableDescriptor); + tableIds.add(tableId); + } + + // Write data to each table to trigger segment rollover and remote log copy + for (int t = 0; t < tableCount; t++) { + TableBucket tb = new TableBucket(tableIds.get(t), 0); + FLUSS_CLUSTER_EXTENSION.waitUntilAllReplicaReady(tb); + + int leaderId = FLUSS_CLUSTER_EXTENSION.waitAndGetLeader(tb); + TabletServerGateway leaderGateway = + FLUSS_CLUSTER_EXTENSION.newTabletServerClientForNode(leaderId); + + // Write enough data to create multiple segments (segment size is 1kb) + int batchCount = 10; + for (int i = 0; i < batchCount; i++) { + assertProduceLogResponse( + leaderGateway + .produceLog( + newProduceLogRequest( + tableIds.get(t), + 0, + 1, + genMemoryLogRecordsByObject(DATA1))) + .get(), + 0, + (long) i * DATA1.size()); + } + } + + // Wait for all tables' log segments to be copied to remote + for (int t = 0; t < tableCount; t++) { + TableBucket tb = new TableBucket(tableIds.get(t), 0); + FLUSS_CLUSTER_EXTENSION.waitUntilSomeLogSegmentsCopyToRemote(tb); + } + + // Collect the remote data directories used by each table + // The remote log dir is derived from table's remoteDataDir: {remoteDataDir}/log + Set usedRemoteDataDirs = new HashSet<>(); + for (int t = 0; t < tableCount; t++) { + Optional tableOpt = zkClient.getTable(tablePaths.get(t)); + assertThat(tableOpt).isPresent(); + TableRegistration table = tableOpt.get(); + + assertThat(table.remoteDataDir).isNotNull(); + usedRemoteDataDirs.add(table.remoteDataDir.toString()); + + // Verify the remote log files actually exist + FsPath remoteLogDir = FlussPaths.remoteLogDir(table.remoteDataDir); + TableBucket tb = new TableBucket(tableIds.get(t), 0); + FsPath remoteLogTabletDir = + FlussPaths.remoteLogTabletDir( + remoteLogDir, PhysicalTablePath.of(tablePaths.get(t)), tb); + assertThat(remoteLogTabletDir.getFileSystem().exists(remoteLogTabletDir)).isTrue(); + FileStatus[] fileStatuses = + remoteLogTabletDir.getFileSystem().listStatus(remoteLogTabletDir); + assertThat(fileStatuses).isNotEmpty(); + } + + assertThat(usedRemoteDataDirs).hasSameSizeAs(REMOTE_DIR_NAMES); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testRemoteLogTieredToMultipleDirsForPartitionedTable(boolean isPrimaryTable) + throws Exception { + // Create a partitioned table and add multiple partitions (more than number of remote dirs) + // to ensure round-robin distribution. Each partition's remote log should be uploaded to + // different remote dirs. + int partitionCount = 6; + String tablePrefix = isPrimaryTable ? "partitioned_pk_" : "partitioned_log_"; + TablePath tablePath = TablePath.of("test_db", tablePrefix + "remote_table"); + + // Create partitioned table + Schema.Builder schemaBuilder = + Schema.newBuilder() + .column("a", org.apache.fluss.types.DataTypes.INT()) + .column("b", org.apache.fluss.types.DataTypes.STRING()) + .column("c", org.apache.fluss.types.DataTypes.STRING()); + if (isPrimaryTable) { + schemaBuilder.primaryKey("a", "c"); + } + Schema schema = schemaBuilder.build(); + + TableDescriptor tableDescriptor = + TableDescriptor.builder() + .schema(schema) + .distributedBy(1) // Single bucket for simpler verification + .partitionedBy("c") + .build(); + + long tableId = createTable(FLUSS_CLUSTER_EXTENSION, tablePath, tableDescriptor); + + // Create partitions + List partitionNames = new ArrayList<>(); + for (int i = 0; i < partitionCount; i++) { + String partitionName = "p" + i; + partitionNames.add(partitionName); + PartitionSpec partitionSpec = + new PartitionSpec(Collections.singletonMap("c", partitionName)); + createPartition(FLUSS_CLUSTER_EXTENSION, tablePath, partitionSpec, false); + } + + // Get partition IDs from ZK + Map partitionRegistrations = + zkClient.getPartitionRegistrations(tablePath); + assertThat(partitionRegistrations).hasSize(partitionCount); + + // Wait for all partitions to be ready and write data to trigger segment rollover + for (int p = 0; p < partitionCount; p++) { + String partitionName = partitionNames.get(p); + Long partitionId = partitionRegistrations.get(partitionName).getPartitionId(); + TableBucket tb = new TableBucket(tableId, partitionId, 0); + FLUSS_CLUSTER_EXTENSION.waitUntilAllReplicaReady(tb); + + int leaderId = FLUSS_CLUSTER_EXTENSION.waitAndGetLeader(tb); + TabletServerGateway leaderGateway = + FLUSS_CLUSTER_EXTENSION.newTabletServerClientForNode(leaderId); + + // Write enough data to create multiple segments (segment size is 1kb) + int batchCount = 10; + for (int i = 0; i < batchCount; i++) { + assertProduceLogResponse( + leaderGateway + .produceLog( + newProduceLogRequest( + tableId, + partitionId, + 0, + 1, + genMemoryLogRecordsByObject(DATA1))) + .get(), + 0, + (long) i * DATA1.size()); + } + } + + // Wait for all partitions' log segments to be copied to remote + for (int p = 0; p < partitionCount; p++) { + Long partitionId = partitionRegistrations.get(partitionNames.get(p)).getPartitionId(); + TableBucket tb = new TableBucket(tableId, partitionId, 0); + FLUSS_CLUSTER_EXTENSION.waitUntilSomeLogSegmentsCopyToRemote(tb); + } + + // Collect the remote data directories used by each partition + Set usedRemoteDataDirs = new HashSet<>(); + for (int p = 0; p < partitionCount; p++) { + String partitionName = partitionNames.get(p); + Long partitionId = partitionRegistrations.get(partitionName).getPartitionId(); + Optional partitionOpt = + zkClient.getPartition(tablePath, partitionName); + assertThat(partitionOpt).isPresent(); + PartitionRegistration partition = partitionOpt.get(); + + assertThat(partition.getRemoteDataDir()).isNotNull(); + usedRemoteDataDirs.add(partition.getRemoteDataDir()); + + // Verify the remote log files actually exist + FsPath remoteLogDir = FlussPaths.remoteLogDir(partition.getRemoteDataDir()); + TableBucket tb = new TableBucket(tableId, partitionId, 0); + FsPath remoteLogTabletDir = + FlussPaths.remoteLogTabletDir( + remoteLogDir, PhysicalTablePath.of(tablePath, partitionName), tb); + assertThat(remoteLogTabletDir.getFileSystem().exists(remoteLogTabletDir)).isTrue(); + FileStatus[] fileStatuses = + remoteLogTabletDir.getFileSystem().listStatus(remoteLogTabletDir); + assertThat(fileStatuses).isNotEmpty(); + } + + // All configured remote dirs should be used due to round-robin distribution + assertThat(usedRemoteDataDirs).hasSameSizeAs(REMOTE_DIR_NAMES); + } + @Test void testRemoteLogTTLWithDynamicLakeToggle() throws Exception { TablePath tablePath = TablePath.of("fluss", "test_remote_log_ttl_dynamic_lake"); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogManagerTest.java index 96e8ae704f..e9e745c7bf 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogManagerTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogManagerTest.java @@ -19,9 +19,12 @@ import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.exception.NotLeaderOrFollowerException; +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.remote.RemoteLogFetchInfo; +import org.apache.fluss.remote.RemoteLogManifest; import org.apache.fluss.remote.RemoteLogSegment; import org.apache.fluss.rpc.entity.FetchLogResultForBucket; import org.apache.fluss.rpc.protocol.ApiError; @@ -37,6 +40,8 @@ import org.apache.fluss.server.replica.ReplicaManager; import org.apache.fluss.server.testutils.ServerTestTags; import org.apache.fluss.server.zk.data.LeaderAndIsr; +import org.apache.fluss.server.zk.data.RemoteLogManifestHandle; +import org.apache.fluss.utils.FlussPaths; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Tag; @@ -47,11 +52,13 @@ import org.junit.jupiter.params.provider.ValueSource; import java.io.File; +import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -468,7 +475,7 @@ void testRemoteFirstFetchRejectsNonLeader(boolean partitionTable) throws Excepti @ParameterizedTest @ValueSource(booleans = {true, false}) void testCleanupLocalSegments(boolean partitionTable) throws Exception { - TableBucket tb = makeTableBucket(partitionTable); + TableBucket tb = makeTableBucket(partitionTable, true); // Need to make leader by ReplicaManager. makeKvTableAsLeader(tb, DATA1_TABLE_PATH_PK, INITIAL_LEADER_EPOCH, partitionTable); LogTablet logTablet = replicaManager.getReplicaOrException(tb).getLogTablet(); @@ -819,16 +826,76 @@ void testCopySegmentPartialFailureCommitsSuccessfulOnes(boolean partitionTable) .collect(Collectors.toSet())); } - private TableBucket makeTableBucket(boolean partitionTable) { - return makeTableBucket(DATA1_TABLE_ID, partitionTable); - } + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testRegisterReplicaWithOldManifestWithoutRemoteLogDir(boolean partitionTable) + throws Exception { + TableBucket tb = makeTableBucket(partitionTable); + makeLogTableAsLeader(tb, partitionTable); + LogTablet logTablet = replicaManager.getReplicaOrException(tb).getLogTablet(); + addMultiSegmentsToLogTablet(logTablet, 5); - private TableBucket makeTableBucket(long tableId, boolean partitionTable) { - if (partitionTable) { - return new TableBucket(tableId, 0L, 0); - } else { - return new TableBucket(tableId, 0); + remoteLogTaskScheduler.triggerPeriodicScheduledTasks(); + List remoteLogSegmentList = + remoteLogManager.relevantRemoteLogSegments(tb, 0L); + assertThat(remoteLogSegmentList).hasSize(4); + + // Read the current manifest from remote storage and strip remote_log_dir fields + // to simulate an old-version manifest. + Optional handleOpt = zkClient.getRemoteLogManifestHandle(tb); + assertThat(handleOpt).isPresent(); + FsPath manifestPath = handleOpt.get().getRemoteLogManifestPath(); + FileSystem fs = manifestPath.getFileSystem(); + + RemoteLogManifest currentManifest = + remoteLogStorage.readRemoteLogManifestSnapshot(manifestPath); + byte[] currentJson = currentManifest.toJsonBytes(); + byte[] oldFormatJson = stripRemoteLogDirFromJson(currentJson); + + // Overwrite the manifest file with the old-format JSON (without remote_log_dir) + fs.delete(manifestPath, false); + try (FSDataOutputStream out = fs.create(manifestPath, FileSystem.WriteMode.NO_OVERWRITE)) { + out.write(oldFormatJson); } + + // Verify the written file indeed has no remote_log_dir + RemoteLogManifest oldManifest = + remoteLogStorage.readRemoteLogManifestSnapshot(manifestPath); + assertThat(oldManifest.getRemoteLogDir()).isNull(); + for (RemoteLogSegment seg : oldManifest.getRemoteLogSegmentList()) { + assertThat(seg.remoteLogDir()).isNull(); + } + + // Rebuild replicaManager to trigger registerReplica with the old manifest + replicaManager.shutdown(); + replicaManager = buildReplicaManager(new TestCoordinatorGateway()); + makeLogTableAsLeader(tb, partitionTable); + + // Verify the manifest was fixed with the correct remoteLogDir + RemoteLogTablet remoteLogTablet = remoteLogManager.remoteLogTablet(tb); + assertThat(remoteLogTablet).isNotNull(); + FsPath expectedRemoteLogDir = FlussPaths.remoteLogDir(conf); + assertThat(remoteLogTablet.getRemoteLogDir()).isEqualTo(expectedRemoteLogDir); + assertThat(remoteLogTablet.currentManifest().getRemoteLogDir()) + .isEqualTo(expectedRemoteLogDir); + + // Verify all segments in the manifest also have remoteLogDir set + List fixedSegments = remoteLogTablet.allRemoteLogSegments(); + assertThat(fixedSegments).hasSize(4); + for (RemoteLogSegment seg : fixedSegments) { + assertThat(seg.remoteLogDir()).isEqualTo(expectedRemoteLogDir); + } + + // Verify subsequent tiering still works after the fix + addMultiSegmentsToLogTablet(replicaManager.getReplicaOrException(tb).getLogTablet(), 3); + remoteLogTaskScheduler.triggerPeriodicScheduledTasks(); + assertThat(remoteLogTablet.allRemoteLogSegments().size()).isGreaterThan(4); + } + + private static byte[] stripRemoteLogDirFromJson(byte[] json) { + String jsonStr = new String(json, StandardCharsets.UTF_8); + return jsonStr.replaceAll(",\"remote_log_dir\":\"[^\"]*\"", "") + .getBytes(StandardCharsets.UTF_8); } private static Stream stopArgs() { diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogManifestTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogManifestTest.java new file mode 100644 index 0000000000..a4204dba34 --- /dev/null +++ b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogManifestTest.java @@ -0,0 +1,102 @@ +/* + * 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.log.remote; + +import org.apache.fluss.fs.FsPath; +import org.apache.fluss.metadata.PhysicalTablePath; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.remote.RemoteLogManifest; +import org.apache.fluss.remote.RemoteLogSegment; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.List; +import java.util.UUID; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link RemoteLogManifest}. */ +class RemoteLogManifestTest { + + private static final PhysicalTablePath PHYSICAL_TABLE_PATH = + PhysicalTablePath.of(TablePath.of("db", "test_table")); + private static final TableBucket TABLE_BUCKET = new TableBucket(1001, 0); + private static final FsPath NEW_REMOTE_LOG_DIR = new FsPath("/new_remote_data_dir/log"); + + @Test + void testNewManifestSetsRemoteLogDirOnAllSegments() { + UUID segmentId1 = UUID.randomUUID(); + UUID segmentId2 = UUID.randomUUID(); + + List segments = + Arrays.asList( + RemoteLogSegment.Builder.builder() + .physicalTablePath(PHYSICAL_TABLE_PATH) + .tableBucket(TABLE_BUCKET) + .remoteLogSegmentId(segmentId1) + .remoteLogStartOffset(0) + .remoteLogEndOffset(9) + .maxTimestamp(1000L) + .segmentSizeInBytes(2850) + .remoteLogDir(null) + .build(), + RemoteLogSegment.Builder.builder() + .physicalTablePath(PHYSICAL_TABLE_PATH) + .tableBucket(TABLE_BUCKET) + .remoteLogSegmentId(segmentId2) + .remoteLogStartOffset(10) + .remoteLogEndOffset(19) + .maxTimestamp(2000L) + .segmentSizeInBytes(3200) + .remoteLogDir(null) + .build()); + + RemoteLogManifest oldManifest = + new RemoteLogManifest(PHYSICAL_TABLE_PATH, TABLE_BUCKET, segments, null); + assertThat(oldManifest.getRemoteLogDir()).isNull(); + + RemoteLogManifest newManifest = oldManifest.newManifest(NEW_REMOTE_LOG_DIR); + + assertThat(newManifest.getRemoteLogDir()).isEqualTo(NEW_REMOTE_LOG_DIR); + assertThat(newManifest.getPhysicalTablePath()).isEqualTo(PHYSICAL_TABLE_PATH); + assertThat(newManifest.getTableBucket()).isEqualTo(TABLE_BUCKET); + + List newSegments = newManifest.getRemoteLogSegmentList(); + assertThat(newSegments).hasSize(2); + + RemoteLogSegment seg1 = newSegments.get(0); + assertThat(seg1.remoteLogDir()).isEqualTo(NEW_REMOTE_LOG_DIR); + assertThat(seg1.physicalTablePath()).isEqualTo(PHYSICAL_TABLE_PATH); + assertThat(seg1.tableBucket()).isEqualTo(TABLE_BUCKET); + assertThat(seg1.remoteLogSegmentId()).isEqualTo(segmentId1); + assertThat(seg1.remoteLogStartOffset()).isEqualTo(0); + assertThat(seg1.remoteLogEndOffset()).isEqualTo(9); + assertThat(seg1.maxTimestamp()).isEqualTo(1000L); + assertThat(seg1.segmentSizeInBytes()).isEqualTo(2850); + + RemoteLogSegment seg2 = newSegments.get(1); + assertThat(seg2.remoteLogDir()).isEqualTo(NEW_REMOTE_LOG_DIR); + assertThat(seg2.remoteLogSegmentId()).isEqualTo(segmentId2); + assertThat(seg2.remoteLogStartOffset()).isEqualTo(10); + assertThat(seg2.remoteLogEndOffset()).isEqualTo(19); + assertThat(seg2.maxTimestamp()).isEqualTo(2000L); + assertThat(seg2.segmentSizeInBytes()).isEqualTo(3200); + } +} diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogMaxUploadSegmentsTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogMaxUploadSegmentsTest.java index 5144fc7030..8fe1d7a035 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogMaxUploadSegmentsTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogMaxUploadSegmentsTest.java @@ -30,7 +30,6 @@ import java.util.List; import java.util.stream.Collectors; -import static org.apache.fluss.record.TestData.DATA1_TABLE_ID; import static org.assertj.core.api.Assertions.assertThat; /** Test for {@link LogTieringTask} max upload segments per task limit. */ @@ -81,12 +80,4 @@ void testMaxUploadSegmentsPerTaskLimit(boolean partitionTable) throws Exception .map(s -> s.remoteLogSegmentId().toString()) .collect(Collectors.toSet())); } - - private TableBucket makeTableBucket(boolean partitionTable) { - if (partitionTable) { - return new TableBucket(DATA1_TABLE_ID, 0L, 0); - } else { - return new TableBucket(DATA1_TABLE_ID, 0); - } - } } diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTTLTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTTLTest.java index 6206a2310b..b0d5098c38 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTTLTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTTLTest.java @@ -35,7 +35,6 @@ import java.util.Map; import java.util.concurrent.CompletableFuture; -import static org.apache.fluss.record.TestData.DATA1_TABLE_ID; import static org.assertj.core.api.Assertions.assertThat; /** Test for remote log ttl in {@link RemoteLogManager}. */ @@ -49,12 +48,7 @@ public void setup() throws Exception { @ParameterizedTest @ValueSource(booleans = {true, false}) void testRemoteLogTTL(boolean partitionTable) throws Exception { - TableBucket tb; - if (partitionTable) { - tb = new TableBucket(DATA1_TABLE_ID, 0L, 0); - } else { - tb = new TableBucket(DATA1_TABLE_ID, 0); - } + TableBucket tb = makeTableBucket(partitionTable); // Need to make leader by ReplicaManager. makeLogTableAsLeader(tb, partitionTable); LogTablet logTablet = replicaManager.getReplicaOrException(tb).getLogTablet(); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTestBase.java b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTestBase.java index ef6b2875d9..6819c09089 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTestBase.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTestBase.java @@ -20,6 +20,7 @@ import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; import org.apache.fluss.config.MemorySize; +import org.apache.fluss.fs.FsPath; import org.apache.fluss.metadata.PhysicalTablePath; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.remote.RemoteLogSegment; @@ -29,6 +30,7 @@ import org.apache.fluss.server.replica.Replica; import org.apache.fluss.server.replica.ReplicaTestBase; import org.apache.fluss.server.zk.data.LeaderAndIsr; +import org.apache.fluss.utils.FlussPaths; import org.junit.jupiter.api.BeforeEach; @@ -43,6 +45,7 @@ import static org.apache.fluss.record.TestData.DATA1_PHYSICAL_TABLE_PATH; import static org.apache.fluss.record.TestData.DATA1_PHYSICAL_TABLE_PATH_PA_2024; import static org.apache.fluss.record.TestData.DATA1_TABLE_ID; +import static org.apache.fluss.record.TestData.DEFAULT_REMOTE_DATA_DIR; import static org.assertj.core.api.Assertions.assertThat; /** Test base for remote log. */ @@ -99,7 +102,7 @@ private Replica makeReplicaAndAddSegments( } protected static RemoteLogSegment copyLogSegmentToRemote( - LogTablet logTablet, RemoteLogStorage remoteLogStorage, int segmentIndex) + LogTablet logTablet, DefaultRemoteLogStorage remoteLogStorage, int segmentIndex) throws Exception { PhysicalTablePath tp = logTablet.getPhysicalTablePath(); TableBucket tb = logTablet.getTableBucket(); @@ -126,6 +129,7 @@ protected static RemoteLogSegment copyLogSegmentToRemote( .segmentSizeInBytes(segment.getFileLogRecords().sizeInBytes()) .tableBucket(tb) .physicalTablePath(tp) + .remoteLogDir(new FsPath(DEFAULT_REMOTE_DATA_DIR)) .build(); remoteLogStorage.copyLogSegmentFiles(remoteLogSegment, logSegmentFiles); @@ -136,10 +140,11 @@ protected RemoteLogTablet buildRemoteLogTablet(LogTablet logTablet) { return new RemoteLogTablet( logTablet.getPhysicalTablePath(), logTablet.getTableBucket(), - conf.get(ConfigOptions.TABLE_LOG_TTL).toMillis()); + conf.get(ConfigOptions.TABLE_LOG_TTL).toMillis(), + FlussPaths.remoteLogDir(conf)); } - protected static List createRemoteLogSegmentList(LogTablet logTablet) { + protected List createRemoteLogSegmentList(LogTablet logTablet) { return logTablet.getSegments().stream() .map( segment -> { @@ -153,6 +158,7 @@ protected static List createRemoteLogSegmentList(LogTablet log segment.getFileLogRecords().sizeInBytes()) .tableBucket(logTablet.getTableBucket()) .physicalTablePath(logTablet.getPhysicalTablePath()) + .remoteLogDir(FlussPaths.remoteLogDir(conf)) .build(); } catch (IOException e) { throw new RuntimeException(e); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/NotifyReplicaLakeTableOffsetTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/NotifyReplicaLakeTableOffsetTest.java index 8b5795644e..08e255b89c 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/replica/NotifyReplicaLakeTableOffsetTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/NotifyReplicaLakeTableOffsetTest.java @@ -35,7 +35,6 @@ import static org.apache.fluss.record.LogRecordBatch.CURRENT_LOG_MAGIC_VALUE; import static org.apache.fluss.record.TestData.DATA1; import static org.apache.fluss.record.TestData.DATA1_ROW_TYPE; -import static org.apache.fluss.record.TestData.DATA1_TABLE_ID; import static org.apache.fluss.record.TestData.DEFAULT_SCHEMA_ID; import static org.apache.fluss.testutils.DataTestUtils.createRecordsWithoutBaseLogOffset; import static org.assertj.core.api.Assertions.assertThat; @@ -116,18 +115,6 @@ private void verifyLakeTableOffset( .isEqualTo(maxTimestamp); } - private TableBucket makeTableBucket(boolean partitionTable) { - return makeTableBucket(DATA1_TABLE_ID, partitionTable); - } - - private TableBucket makeTableBucket(long tableId, boolean partitionTable) { - if (partitionTable) { - return new TableBucket(tableId, 0L, 0); - } else { - return new TableBucket(tableId, 0); - } - } - private NotifyLakeTableOffsetData getNotifyLakeTableOffset( TableBucket tableBucket, long snapshotId, diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java index b738c5a859..3bde5b4028 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java @@ -1381,7 +1381,7 @@ void testLimitScanLogTable() throws Exception { @ParameterizedTest @ValueSource(booleans = {true, false}) void testListOffsets(boolean isPartitioned) throws Exception { - TableBucket tb = new TableBucket(DATA1_TABLE_ID, isPartitioned ? 10L : null, 1); + TableBucket tb = makeTableBucket(DATA1_TABLE_ID, isPartitioned ? 10L : null, false); makeLogTableAsLeader(tb, isPartitioned); // produce one batch to this bucket. diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java index bad818a9e9..12848a0607 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java @@ -60,6 +60,7 @@ import org.apache.fluss.server.zk.ZooKeeperClient; import org.apache.fluss.server.zk.ZooKeeperExtension; import org.apache.fluss.server.zk.data.LeaderAndIsr; +import org.apache.fluss.server.zk.data.PartitionAssignment; import org.apache.fluss.server.zk.data.TableRegistration; import org.apache.fluss.testutils.common.AllCallbackWrapper; import org.apache.fluss.testutils.common.ManuallyTriggeredScheduledExecutorService; @@ -136,6 +137,7 @@ public class ReplicaTestBase { protected static final int TABLET_SERVER_ID = 1; private static final String TABLET_SERVER_RACK = "rack1"; protected static ZooKeeperClient zkClient; + protected static String remoteDataDir; // to register all should be closed after each test private final CloseableRegistry closeableRegistry = new CloseableRegistry(); @@ -172,6 +174,7 @@ static void baseBeforeAll() { ZOO_KEEPER_EXTENSION_WRAPPER .getCustomExtension() .getZooKeeperClient(NOPErrorHandler.INSTANCE); + remoteDataDir = zkClient.getDefaultRemoteDataDir(); } @BeforeEach @@ -287,25 +290,31 @@ private void registerTableInZkClient() throws Exception { zkClient.registerTable( DATA1_TABLE_PATH, TableRegistration.newTable( - DATA1_TABLE_ID, DEFAULT_REMOTE_DATA_DIR, data1NonPkTableDescriptor)); + DATA1_TABLE_ID, + conf.get(ConfigOptions.REMOTE_DATA_DIR), + data1NonPkTableDescriptor)); zkClient.registerFirstSchema(DATA1_TABLE_PATH, DATA1_SCHEMA); zkClient.registerTable( DATA1_TABLE_PATH_PK, TableRegistration.newTable( - DATA1_TABLE_ID_PK, DEFAULT_REMOTE_DATA_DIR, DATA1_TABLE_DESCRIPTOR_PK)); + DATA1_TABLE_ID_PK, + conf.get(ConfigOptions.REMOTE_DATA_DIR), + DATA1_TABLE_DESCRIPTOR_PK)); zkClient.registerFirstSchema(DATA1_TABLE_PATH_PK, DATA1_SCHEMA_PK); zkClient.registerTable( DATA2_TABLE_PATH, TableRegistration.newTable( - DATA2_TABLE_ID, DEFAULT_REMOTE_DATA_DIR, DATA2_TABLE_DESCRIPTOR)); + DATA2_TABLE_ID, + conf.get(ConfigOptions.REMOTE_DATA_DIR), + DATA2_TABLE_DESCRIPTOR)); zkClient.registerFirstSchema(DATA2_TABLE_PATH, DATA2_SCHEMA); zkClient.registerTable( DATA3_TABLE_PATH_PK_AUTO_INC, TableRegistration.newTable( DATA3_TABLE_ID_PK_AUTO_INC, - DEFAULT_REMOTE_DATA_DIR, + conf.get(ConfigOptions.REMOTE_DATA_DIR), DATA3_TABLE_DESCRIPTOR_PK_AUTO_INC)); zkClient.registerFirstSchema(DATA3_TABLE_PATH_PK_AUTO_INC, DATA3_SCHEMA_PK_AUTO_INC); } @@ -541,6 +550,7 @@ private Replica makeReplica( NOPErrorHandler.INSTANCE, metricGroup, DATA1_TABLE_INFO, + remoteDataDir, manualClock, remoteLogManager, scannerManager); @@ -620,6 +630,50 @@ protected Set listRemoteLogFiles(TableBucket tableBucket) throws IOExcep .collect(Collectors.toSet()); } + protected TableBucket makeTableBucket(boolean partitionTable) throws Exception { + return makeTableBucket(DATA1_TABLE_ID, partitionTable); + } + + protected TableBucket makeTableBucket(boolean partitionTable, boolean kvTable) + throws Exception { + long tableId = kvTable ? DATA1_TABLE_ID_PK : DATA1_TABLE_ID; + Long partitionId = partitionTable ? 0L : null; + return makeTableBucket(tableId, partitionId, kvTable); + } + + protected TableBucket makeTableBucket(long tableId, boolean partitionTable) throws Exception { + Long partitionId = partitionTable ? 0L : null; + return makeTableBucket(tableId, partitionId, false); + } + + protected TableBucket makeTableBucket(long tableId, Long partitionId, boolean kvTable) + throws Exception { + int bucketId = 0; + boolean partitionTable = partitionId != null; + if (partitionTable) { + if (kvTable) { + zkClient.registerPartitionAssignmentAndMetadata( + partitionId, + DATA1_PHYSICAL_TABLE_PATH_PK_PA_2024.getPartitionName(), + new PartitionAssignment(tableId, Collections.emptyMap()), + conf.get(ConfigOptions.REMOTE_DATA_DIR), + DATA1_TABLE_PATH_PK, + tableId); + } else { + zkClient.registerPartitionAssignmentAndMetadata( + partitionId, + DATA1_PHYSICAL_TABLE_PATH_PA_2024.getPartitionName(), + new PartitionAssignment(tableId, Collections.emptyMap()), + conf.get(ConfigOptions.REMOTE_DATA_DIR), + DATA1_TABLE_PATH, + tableId); + } + return new TableBucket(tableId, partitionId, bucketId); + } else { + return new TableBucket(tableId, bucketId); + } + } + /** An implementation of {@link SnapshotContext} for test purpose. */ protected class TestSnapshotContext implements SnapshotContext { @@ -707,11 +761,6 @@ public int getSnapshotFsWriteBufferSize() { return 1024; } - @Override - public FsPath getRemoteKvDir() { - return remoteKvTabletDir; - } - @Override public FunctionWithException getLatestCompletedSnapshotProvider() { diff --git a/fluss-server/src/test/java/org/apache/fluss/server/testutils/RpcMessageTestUtils.java b/fluss-server/src/test/java/org/apache/fluss/server/testutils/RpcMessageTestUtils.java index 8562163827..e82963e54e 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/testutils/RpcMessageTestUtils.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/testutils/RpcMessageTestUtils.java @@ -193,6 +193,11 @@ public static MetadataRequest newMetadataRequest(List tablePaths) { public static ProduceLogRequest newProduceLogRequest( long tableId, int bucketId, int acks, MemoryLogRecords records) { + return newProduceLogRequest(tableId, null, bucketId, acks, records); + } + + public static ProduceLogRequest newProduceLogRequest( + long tableId, Long partitionId, int bucketId, int acks, MemoryLogRecords records) { ProduceLogRequest produceRequest = new ProduceLogRequest(); produceRequest.setTableId(tableId).setAcks(acks).setTimeoutMs(10000); PbProduceLogReqForBucket pbProduceLogReqForBucket = new PbProduceLogReqForBucket(); @@ -203,16 +208,27 @@ public static ProduceLogRequest newProduceLogRequest( records.getMemorySegment(), records.getPosition(), records.sizeInBytes())); + if (partitionId != null) { + pbProduceLogReqForBucket.setPartitionId(partitionId); + } produceRequest.addAllBucketsReqs(Collections.singletonList(pbProduceLogReqForBucket)); return produceRequest; } public static PutKvRequest newPutKvRequest( long tableId, int bucketId, int acks, KvRecordBatch kvRecordBatch) { + return newPutKvRequest(tableId, null, bucketId, acks, kvRecordBatch); + } + + public static PutKvRequest newPutKvRequest( + long tableId, Long partitionId, int bucketId, int acks, KvRecordBatch kvRecordBatch) { PutKvRequest putKvRequest = new PutKvRequest(); putKvRequest.setTableId(tableId).setAcks(acks).setTimeoutMs(10000); PbPutKvReqForBucket pbPutKvReqForBucket = new PbPutKvReqForBucket(); pbPutKvReqForBucket.setBucketId(bucketId); + if (partitionId != null) { + pbPutKvReqForBucket.setPartitionId(partitionId); + } if (kvRecordBatch instanceof DefaultKvRecordBatch) { DefaultKvRecordBatch batch = (DefaultKvRecordBatch) kvRecordBatch; pbPutKvReqForBucket.setRecords(