diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java index 233902cb391..18160b3d076 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java +++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java @@ -1775,6 +1775,15 @@ public class ConfigOptions { + "The partitions earlier than 20241108 will be deleted. " + "The default value is 7."); + public static final ConfigOption TABLE_AUTO_PARTITION_RETENTION_ENSURE_TIERED = + key("table.auto-partition.retention.ensure-tiered") + .booleanType() + .defaultValue(false) + .withDescription( + "Whether an expired auto partition must be fully committed to the data lake before deletion. " + + "When enabled, writes to the partition are frozen first, and the partition is deleted only after the lake contains all frozen log offsets. " + + "The default value is false."); + public static final ConfigOption TABLE_LOG_TTL = key("table.log.ttl") .durationType() diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/AutoPartitionStrategy.java b/fluss-common/src/main/java/org/apache/fluss/utils/AutoPartitionStrategy.java index 3c393415e64..f9bc578dd78 100644 --- a/fluss-common/src/main/java/org/apache/fluss/utils/AutoPartitionStrategy.java +++ b/fluss-common/src/main/java/org/apache/fluss/utils/AutoPartitionStrategy.java @@ -34,6 +34,7 @@ public class AutoPartitionStrategy { private final String timeFormat; private final int numPreCreate; private final int numToRetain; + private final boolean ensureLakeTieredBeforeDrop; private final TimeZone timeZone; private AutoPartitionStrategy( @@ -43,6 +44,7 @@ private AutoPartitionStrategy( String timeFormat, int numPreCreate, int numToRetain, + boolean ensureLakeTieredBeforeDrop, TimeZone timeZone) { this.autoPartitionEnabled = autoPartitionEnabled; this.key = key; @@ -50,6 +52,7 @@ private AutoPartitionStrategy( this.timeFormat = timeFormat; this.numPreCreate = numPreCreate; this.numToRetain = numToRetain; + this.ensureLakeTieredBeforeDrop = ensureLakeTieredBeforeDrop; this.timeZone = timeZone; } @@ -65,6 +68,7 @@ public static AutoPartitionStrategy from(Configuration conf) { conf.getOptional(ConfigOptions.TABLE_AUTO_PARTITION_TIME_FORMAT).orElse(null), conf.getInt(ConfigOptions.TABLE_AUTO_PARTITION_NUM_PRECREATE), conf.getInt(ConfigOptions.TABLE_AUTO_PARTITION_NUM_RETENTION), + conf.getBoolean(ConfigOptions.TABLE_AUTO_PARTITION_RETENTION_ENSURE_TIERED), TimeZone.getTimeZone(conf.getString(ConfigOptions.TABLE_AUTO_PARTITION_TIMEZONE))); } @@ -92,6 +96,10 @@ public int numToRetain() { return numToRetain; } + public boolean ensureLakeTieredBeforeDrop() { + return ensureLakeTieredBeforeDrop; + } + public TimeZone timeZone() { return timeZone; } diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/TabletServerGateway.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/TabletServerGateway.java index 34bcd361ef9..423a1a43f8e 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/TabletServerGateway.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/TabletServerGateway.java @@ -20,6 +20,8 @@ import org.apache.fluss.rpc.RpcGateway; import org.apache.fluss.rpc.messages.FetchLogRequest; import org.apache.fluss.rpc.messages.FetchLogResponse; +import org.apache.fluss.rpc.messages.FreezePartitionRequest; +import org.apache.fluss.rpc.messages.FreezePartitionResponse; import org.apache.fluss.rpc.messages.GetTableStatsRequest; import org.apache.fluss.rpc.messages.GetTableStatsResponse; import org.apache.fluss.rpc.messages.InitWriterRequest; @@ -84,6 +86,10 @@ CompletableFuture notifyLeaderAndIsr( @RPC(api = ApiKeys.STOP_REPLICA) CompletableFuture stopReplica(StopReplicaRequest stopBucketReplicaRequest); + /** Freeze writes to partition bucket leaders for lake-aware retention. */ + @RPC(api = ApiKeys.FREEZE_PARTITION) + CompletableFuture freezePartition(FreezePartitionRequest request); + /** * Produce log data to the specified table bucket. * diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java index b03e1ec63ab..127f37dd2d2 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java @@ -106,7 +106,8 @@ public enum ApiKeys { SCAN_KV(1061, 0, 0, PUBLIC), GET_CLUSTER_HEALTH(1062, 0, 0, PUBLIC), LIST_REMOTE_LOG_MANIFESTS(1063, 0, 0, PUBLIC), - LIST_KV_SNAPSHOTS(1064, 0, 0, PUBLIC); + LIST_KV_SNAPSHOTS(1064, 0, 0, PUBLIC), + FREEZE_PARTITION(1065, 0, 0, PRIVATE); private static final Map ID_TO_TYPE = Arrays.stream(ApiKeys.values()) diff --git a/fluss-rpc/src/main/proto/FlussApi.proto b/fluss-rpc/src/main/proto/FlussApi.proto index 836c642a7aa..0fd14f3ff74 100644 --- a/fluss-rpc/src/main/proto/FlussApi.proto +++ b/fluss-rpc/src/main/proto/FlussApi.proto @@ -375,6 +375,16 @@ message StopReplicaResponse { repeated PbStopReplicaRespForBucket stop_replicas_resp = 1; } +// Freeze writes to partition bucket leaders before lake-aware retention. +message FreezePartitionRequest { + required int32 coordinator_epoch = 1; + repeated PbFreezePartitionReqForBucket buckets_req = 2; +} + +message FreezePartitionResponse { + repeated PbFreezePartitionRespForBucket buckets_resp = 1; +} + // adjust isr request and response message AdjustIsrRequest { required int32 serverId = 1; @@ -1065,6 +1075,19 @@ message PbStopReplicaRespForBucket { optional string error_message = 3; } +message PbFreezePartitionReqForBucket { + required PbTableBucket table_bucket = 1; + required int32 leader_epoch = 2; +} + +message PbFreezePartitionRespForBucket { + required PbTableBucket table_bucket = 1; + optional int32 error_code = 2; + optional string error_message = 3; + optional int64 high_watermark = 4; + optional int64 log_end_offset = 5; +} + message PbKvSnapshot { required int32 bucket_id = 1; // null if there is no snapshot for this bucket diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/TestingTabletGatewayService.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/TestingTabletGatewayService.java index f465bb4a69a..53baee1357d 100644 --- a/fluss-rpc/src/test/java/org/apache/fluss/rpc/TestingTabletGatewayService.java +++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/TestingTabletGatewayService.java @@ -25,6 +25,8 @@ import org.apache.fluss.rpc.messages.DescribeClusterConfigsResponse; import org.apache.fluss.rpc.messages.FetchLogRequest; import org.apache.fluss.rpc.messages.FetchLogResponse; +import org.apache.fluss.rpc.messages.FreezePartitionRequest; +import org.apache.fluss.rpc.messages.FreezePartitionResponse; import org.apache.fluss.rpc.messages.GetClusterHealthRequest; import org.apache.fluss.rpc.messages.GetClusterHealthResponse; import org.apache.fluss.rpc.messages.GetDatabaseInfoRequest; @@ -112,6 +114,12 @@ public CompletableFuture stopReplica( return null; } + @Override + public CompletableFuture freezePartition( + FreezePartitionRequest request) { + return null; + } + @Override public CompletableFuture produceLog(ProduceLogRequest request) { return null; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java index 7f513a47636..0a7ea825a45 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java @@ -93,6 +93,7 @@ public class AutoPartitionManager implements AutoCloseable { private final MetadataManager metadataManager; private final RemoteDirDynamicLoader remoteDirDynamicLoader; private final ReplicaCapacityController replicaCapacityController; + private final @Nullable LakeAwarePartitionRetentionManager lakeAwareRetentionManager; private final Clock clock; private final long periodicInterval; @@ -113,6 +114,7 @@ public class AutoPartitionManager implements AutoCloseable { private final Lock lock = new ReentrantLock(); + @VisibleForTesting public AutoPartitionManager( ServerMetadataCache metadataCache, MetadataManager metadataManager, @@ -129,7 +131,29 @@ public AutoPartitionManager( // TODO: Reuse the CoordinatorServer shared scheduler for this lightweight // coordinator periodic task instead of creating a component-owned scheduler. Executors.newScheduledThreadPool( - 1, new ExecutorThreadFactory("periodic-auto-partition-manager"))); + 1, new ExecutorThreadFactory("periodic-auto-partition-manager")), + null); + } + + public AutoPartitionManager( + ServerMetadataCache metadataCache, + MetadataManager metadataManager, + RemoteDirDynamicLoader remoteDirDynamicLoader, + Configuration conf, + ReplicaCapacityController replicaCapacityController, + LakeAwarePartitionRetentionManager lakeAwareRetentionManager) { + this( + metadataCache, + metadataManager, + remoteDirDynamicLoader, + conf, + replicaCapacityController, + SystemClock.getInstance(), + // TODO: Reuse the CoordinatorServer shared scheduler for this lightweight + // coordinator periodic task instead of creating a component-owned scheduler. + Executors.newScheduledThreadPool( + 1, new ExecutorThreadFactory("periodic-auto-partition-manager")), + checkNotNull(lakeAwareRetentionManager)); } @VisibleForTesting @@ -141,10 +165,32 @@ public AutoPartitionManager( ReplicaCapacityController replicaCapacityController, Clock clock, ScheduledExecutorService periodicExecutor) { + this( + metadataCache, + metadataManager, + remoteDirDynamicLoader, + conf, + replicaCapacityController, + clock, + periodicExecutor, + null); + } + + @VisibleForTesting + AutoPartitionManager( + ServerMetadataCache metadataCache, + MetadataManager metadataManager, + RemoteDirDynamicLoader remoteDirDynamicLoader, + Configuration conf, + ReplicaCapacityController replicaCapacityController, + Clock clock, + ScheduledExecutorService periodicExecutor, + @Nullable LakeAwarePartitionRetentionManager lakeAwareRetentionManager) { this.metadataCache = metadataCache; this.metadataManager = metadataManager; this.remoteDirDynamicLoader = remoteDirDynamicLoader; this.replicaCapacityController = replicaCapacityController; + this.lakeAwareRetentionManager = lakeAwareRetentionManager; this.clock = clock; this.periodicExecutor = periodicExecutor; this.periodicInterval = conf.get(ConfigOptions.AUTO_PARTITION_CHECK_INTERVAL).toMillis(); @@ -388,8 +434,7 @@ private void doAutoPartition(Instant now, Set tableIds, boolean forceDoAut } dropPartitions( - tablePath, - tableInfo.getPartitionKeys(), + tableInfo, now, tableInfo.getTableConfig().getAutoPartitionStrategy(), currentPartitions); @@ -502,11 +547,12 @@ private List partitionNamesToPreCreate( } private void dropPartitions( - TablePath tablePath, - List partitionKeys, + TableInfo tableInfo, Instant currentInstant, AutoPartitionStrategy autoPartitionStrategy, NavigableMap> currentPartitions) { + boolean ensureTiered = autoPartitionStrategy.ensureLakeTieredBeforeDrop(); + int numToRetain = autoPartitionStrategy.numToRetain(); // negative value means not to drop partitions if (numToRetain < 0) { @@ -541,15 +587,17 @@ private void dropPartitions( currentPartitions.headMap(lastRetainPartitionTime).entrySet().iterator(); while (iterator.hasNext()) { Map.Entry> entry = iterator.next(); - dropPartitions(tablePath, partitionKeys, iterator, entry); + dropPartitions(tableInfo, ensureTiered, iterator, entry); } } private void dropPartitions( - TablePath tablePath, - List partitionKeys, + TableInfo tableInfo, + boolean ensureTiered, Iterator>> iterator, Map.Entry> entry) { + TablePath tablePath = tableInfo.getTablePath(); + List partitionKeys = tableInfo.getPartitionKeys(); Iterator dropIterator; if (entry.getValue() == null) { dropIterator = new HashSet<>(Collections.singleton(entry.getKey())).iterator(); @@ -559,6 +607,11 @@ private void dropPartitions( while (dropIterator.hasNext()) { String partitionName = dropIterator.next(); + if (ensureTiered) { + checkNotNull(lakeAwareRetentionManager).startRetention(tableInfo, partitionName); + continue; + } + try { metadataManager.dropPartition( tablePath, @@ -577,7 +630,9 @@ private void dropPartitions( partitionName, tablePath); } - iterator.remove(); + if (!ensureTiered) { + iterator.remove(); + } } @VisibleForTesting @@ -590,6 +645,9 @@ protected Integer getAutoCreateDayDelayMinutes(long tableId) { public void close() throws Exception { if (isClosed.compareAndSet(false, true)) { periodicExecutor.shutdownNow(); + if (lakeAwareRetentionManager != null) { + lakeAwareRetentionManager.close(); + } } } } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorChannelManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorChannelManager.java index e095935ae2b..042ae7fedc4 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorChannelManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorChannelManager.java @@ -20,9 +20,12 @@ import org.apache.fluss.annotation.VisibleForTesting; import org.apache.fluss.cluster.ServerNode; import org.apache.fluss.cluster.ServerType; +import org.apache.fluss.exception.TabletServerNotAvailableException; import org.apache.fluss.rpc.RpcClient; import org.apache.fluss.rpc.gateway.TabletServerGateway; import org.apache.fluss.rpc.messages.ApiMessage; +import org.apache.fluss.rpc.messages.FreezePartitionRequest; +import org.apache.fluss.rpc.messages.FreezePartitionResponse; import org.apache.fluss.rpc.messages.NotifyKvSnapshotOffsetRequest; import org.apache.fluss.rpc.messages.NotifyKvSnapshotOffsetResponse; import org.apache.fluss.rpc.messages.NotifyLakeTableOffsetRequest; @@ -40,7 +43,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import javax.annotation.concurrent.NotThreadSafe; +import javax.annotation.concurrent.ThreadSafe; import java.util.Collection; import java.util.Optional; @@ -53,7 +56,7 @@ * Using by coordinator server. It's a manager to manage the rpc channels to tablet servers and send * request to the servers. */ -@NotThreadSafe +@ThreadSafe public class CoordinatorChannelManager { private static final Logger LOG = LoggerFactory.getLogger(CoordinatorChannelManager.class); @@ -121,6 +124,20 @@ public void sendStopBucketReplicaRequest( responseConsumer); } + /** Send a request to freeze partition writes on a tablet server. */ + public CompletableFuture sendFreezePartitionRequest( + int receiveServerId, FreezePartitionRequest request) { + Optional gateway = getTabletServerGateway(receiveServerId); + if (gateway.isPresent()) { + return gateway.get().freezePartition(request); + } + CompletableFuture response = new CompletableFuture<>(); + response.completeExceptionally( + new TabletServerNotAvailableException( + "Tablet server " + receiveServerId + " is not available.")); + return response; + } + /** Send UpdateMetadataRequest to the server and handle the response. */ public void sendUpdateMetadataRequest( int receiveServerId, diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java index 0f610eb42a3..48ce34e0023 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java @@ -329,14 +329,21 @@ protected void initCoordinatorLeader() throws Exception { this.coordinatorChannelManager = new CoordinatorChannelManager(rpcClient); + LakeAwarePartitionRetentionManager lakeAwarePartitionRetentionManager = + new LakeAwarePartitionRetentionManager( + metadataManager, + zkClient, + coordinatorChannelManager, + zkEpoch.getCoordinatorEpoch(), + ioExecutor); this.autoPartitionManager = new AutoPartitionManager( metadataCache, metadataManager, remoteDirDynamicLoader, conf, - replicaCapacityController); - autoPartitionManager.start(); + replicaCapacityController, + lakeAwarePartitionRetentionManager); // start coordinator event processor after we register coordinator leader to zk // so that the event processor can get the coordinator leader node from zk during @@ -360,6 +367,7 @@ protected void initCoordinatorLeader() throws Exception { scheduler, clock); coordinatorEventProcessor.startup(); + autoPartitionManager.start(); // As the active leader, this server is the sole writer of dynamic configs and holds the // latest values, so stop consuming change notifications to avoid rolling a value back. diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeAwarePartitionRetentionManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeAwarePartitionRetentionManager.java new file mode 100644 index 00000000000..17e2043d66a --- /dev/null +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeAwarePartitionRetentionManager.java @@ -0,0 +1,630 @@ +/* + * 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.exception.FlussRuntimeException; +import org.apache.fluss.metadata.PhysicalTablePath; +import org.apache.fluss.metadata.ResolvedPartitionSpec; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metadata.TableInfo; +import org.apache.fluss.metadata.TablePartition; +import org.apache.fluss.rpc.messages.FreezePartitionRequest; +import org.apache.fluss.rpc.messages.FreezePartitionResponse; +import org.apache.fluss.rpc.messages.PbFreezePartitionRespForBucket; +import org.apache.fluss.server.zk.ZooKeeperClient; +import org.apache.fluss.server.zk.data.LeaderAndIsr; +import org.apache.fluss.server.zk.data.PartitionAssignment; +import org.apache.fluss.server.zk.data.PartitionRegistration; +import org.apache.fluss.server.zk.data.PartitionRegistration.FrozenBucket; +import org.apache.fluss.server.zk.data.lake.LakeTableSnapshot; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.Executor; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.apache.fluss.server.utils.ServerRpcMessageUtils.makeFreezePartitionRequest; +import static org.apache.fluss.server.utils.ServerRpcMessageUtils.toTableBucket; + +/** Coordinates freeze, lake offset verification, and deletion for safe partition retention. */ +public class LakeAwarePartitionRetentionManager implements AutoCloseable { + + private static final Logger LOG = + LoggerFactory.getLogger(LakeAwarePartitionRetentionManager.class); + + private final MetadataManager metadataManager; + private final ZooKeeperClient zooKeeperClient; + private final CoordinatorChannelManager coordinatorChannelManager; + private final int coordinatorEpoch; + private final Executor executor; + private final ConcurrentMap retainingPartitions = + new ConcurrentHashMap<>(); + private final AtomicBoolean closed = new AtomicBoolean(false); + + public LakeAwarePartitionRetentionManager( + MetadataManager metadataManager, + ZooKeeperClient zooKeeperClient, + CoordinatorChannelManager coordinatorChannelManager, + int coordinatorEpoch, + Executor executor) { + this.metadataManager = metadataManager; + this.zooKeeperClient = zooKeeperClient; + this.coordinatorChannelManager = coordinatorChannelManager; + this.coordinatorEpoch = coordinatorEpoch; + this.executor = executor; + } + + /** Start or continue lake-aware retention for an expired partition. */ + public void startRetention(TableInfo tableInfo, String partitionName) { + PhysicalTablePath physicalTablePath = + PhysicalTablePath.of(tableInfo.getTablePath(), partitionName); + if (closed.get()) { + return; + } + + RetentionStatus newStatus = new RetentionStatus(); + RetentionStatus retention = retainingPartitions.putIfAbsent(physicalTablePath, newStatus); + if (retention == null) { + submitFreeze(tableInfo, partitionName, physicalTablePath, newStatus); + } else { + toNext(tableInfo, partitionName, physicalTablePath, retention); + } + } + + private void toNext( + TableInfo tableInfo, + String partitionName, + PhysicalTablePath physicalTablePath, + RetentionStatus retention) { + synchronized (retention) { + if (closed.get() + || retainingPartitions.get(physicalTablePath) != retention + || retention.operationRunning) { + return; + } + retention.operationRunning = true; + if (retention.stage == RetentionStage.FROZEN) { + submitLakeCheck(tableInfo, partitionName, physicalTablePath, retention); + } else if (retention.stage == RetentionStage.LAKE_TIERED) { + submitDrop(tableInfo, partitionName, physicalTablePath, retention); + } + } + } + + private void submitFreeze( + TableInfo tableInfo, + String partitionName, + PhysicalTablePath physicalTablePath, + RetentionStatus retention) { + if (closed.get()) { + retainingPartitions.remove(physicalTablePath, retention); + return; + } + try { + executor.execute( + () -> freezePartition(tableInfo, partitionName, physicalTablePath, retention)); + } catch (RuntimeException e) { + retainingPartitions.remove(physicalTablePath, retention); + LOG.warn( + "Failed to submit lake-aware retention for partition {} of table {}.", + partitionName, + tableInfo.getTablePath(), + e); + } + } + + private void freezePartition( + TableInfo tableInfo, + String partitionName, + PhysicalTablePath physicalTablePath, + RetentionStatus retention) { + Optional registration; + try { + registration = + metadataManager.getPartitionRegistration( + tableInfo.getTablePath(), partitionName); + } catch (Exception e) { + LOG.warn( + "Failed to load partition {} of table {} for lake-aware retention.", + partitionName, + tableInfo.getTablePath(), + e); + retainingPartitions.remove(physicalTablePath, retention); + return; + } + if (!registration.isPresent()) { + retainingPartitions.remove(physicalTablePath, retention); + return; + } + if (registration.get().getTableId() != tableInfo.getTableId()) { + retainingPartitions.remove(physicalTablePath, retention); + return; + } + freezePartition(tableInfo, partitionName, registration.get(), physicalTablePath, retention); + } + + private void freezePartition( + TableInfo tableInfo, + String partitionName, + PartitionRegistration partitionRegistration, + PhysicalTablePath physicalTablePath, + RetentionStatus retention) { + if (closed.get()) { + retainingPartitions.remove(physicalTablePath, retention); + return; + } + + try { + PartitionAssignment partitionAssignment = + zooKeeperClient + .getPartitionAssignment(partitionRegistration.getPartitionId()) + .orElse(null); + if (partitionAssignment == null) { + LOG.info( + "Waiting to retain partition {} of table {} because its assignment is unavailable.", + partitionName, + tableInfo.getTablePath()); + retainingPartitions.remove(physicalTablePath, retention); + return; + } + + List tableBuckets = new ArrayList<>(); + for (Integer bucketId : partitionAssignment.getBucketAssignments().keySet()) { + tableBuckets.add( + new TableBucket( + tableInfo.getTableId(), + partitionRegistration.getPartitionId(), + bucketId)); + } + Map leaders = zooKeeperClient.getLeaderAndIsrs(tableBuckets); + if (leaders.size() != tableBuckets.size() + || leaders.values().stream() + .anyMatch( + leaderAndIsr -> + leaderAndIsr.leader() == LeaderAndIsr.NO_LEADER)) { + LOG.info( + "Waiting to retain partition {} of table {} because not all bucket leaders are available.", + partitionName, + tableInfo.getTablePath()); + retainingPartitions.remove(physicalTablePath, retention); + return; + } + + Map> leaderEpochsByServer = new HashMap<>(); + leaders.forEach( + (tableBucket, leaderAndIsr) -> + leaderEpochsByServer + .computeIfAbsent( + leaderAndIsr.leader(), ignored -> new HashMap<>()) + .put(tableBucket, leaderAndIsr.leaderEpoch())); + + List> freezeFutures = new ArrayList<>(); + for (Map.Entry> entry : + leaderEpochsByServer.entrySet()) { + FreezePartitionRequest request = + makeFreezePartitionRequest(coordinatorEpoch, entry.getValue()); + freezeFutures.add( + coordinatorChannelManager.sendFreezePartitionRequest( + entry.getKey(), request)); + } + + CompletableFuture.allOf( + freezeFutures.toArray(new CompletableFuture[freezeFutures.size()])) + .whenCompleteAsync( + (ignored, error) -> { + try { + if (closed.get()) { + retainingPartitions.remove(physicalTablePath, retention); + return; + } + if (error != null) { + LOG.warn( + "Failed to freeze partition {} of table {} for retention.", + partitionName, + tableInfo.getTablePath(), + error); + retainingPartitions.remove(physicalTablePath, retention); + return; + } + finishFreeze( + tableInfo, + partitionName, + partitionRegistration, + leaders, + freezeFutures, + physicalTablePath, + retention); + } catch (Exception e) { + LOG.warn( + "Failed to finish freezing partition {} of table {} for retention.", + partitionName, + tableInfo.getTablePath(), + e); + retainingPartitions.remove(physicalTablePath, retention); + } + }, + executor); + } catch (Exception e) { + LOG.warn( + "Failed to start lake-aware retention for partition {} of table {}.", + partitionName, + tableInfo.getTablePath(), + e); + retainingPartitions.remove(physicalTablePath, retention); + } + } + + private void finishFreeze( + TableInfo tableInfo, + String partitionName, + PartitionRegistration initialRegistration, + Map expectedLeaders, + List> freezeFutures, + PhysicalTablePath physicalTablePath, + RetentionStatus retention) + throws Exception { + Map frozenBuckets = + collectFrozenBuckets(expectedLeaders, freezeFutures); + if (!leadersUnchanged(expectedLeaders)) { + LOG.info( + "Waiting to retry retention for partition {} of table {} because a bucket leader changed while freezing.", + partitionName, + tableInfo.getTablePath()); + retainingPartitions.remove(physicalTablePath, retention); + return; + } + + Optional currentRegistration = + metadataManager.getPartitionRegistration(tableInfo.getTablePath(), partitionName); + if (!currentRegistration.isPresent() + || currentRegistration.get().getPartitionId() + != initialRegistration.getPartitionId()) { + retainingPartitions.remove(physicalTablePath, retention); + return; + } + + frozenBuckets = mergeFrozenOffsets(currentRegistration.get(), frozenBuckets); + if (!currentRegistration.get().getFrozenBuckets().equals(frozenBuckets)) { + PartitionRegistration updatedRegistration = + currentRegistration.get().withFrozenBuckets(frozenBuckets); + metadataManager.updatePartitionRegistration( + tableInfo.getTablePath(), partitionName, updatedRegistration); + } + completeOperation(physicalTablePath, retention, RetentionStage.FROZEN); + LOG.info( + "Partition {} of table {} is frozen and ready for the lake tiering check.", + partitionName, + tableInfo.getTablePath()); + } + + private void submitLakeCheck( + TableInfo tableInfo, + String partitionName, + PhysicalTablePath physicalTablePath, + RetentionStatus retention) { + if (closed.get()) { + retainingPartitions.remove(physicalTablePath, retention); + return; + } + try { + executor.execute( + () -> checkLakeTiered(tableInfo, partitionName, physicalTablePath, retention)); + } catch (RuntimeException e) { + completeOperation(physicalTablePath, retention, RetentionStage.FROZEN); + LOG.warn( + "Failed to submit lake tiering check for partition {} of table {}.", + partitionName, + tableInfo.getTablePath(), + e); + } + } + + private void checkLakeTiered( + TableInfo tableInfo, + String partitionName, + PhysicalTablePath physicalTablePath, + RetentionStatus retention) { + try { + Optional registration = + metadataManager.getPartitionRegistration( + tableInfo.getTablePath(), partitionName); + if (!registration.isPresent() + || registration.get().getTableId() != tableInfo.getTableId()) { + retainingPartitions.remove(physicalTablePath, retention); + return; + } + + Map frozenBuckets = registration.get().getFrozenBuckets(); + TablePartition tablePartition = registration.get().toTablePartition(); + if (frozenBuckets.isEmpty() || !leadersUnchanged(tablePartition, frozenBuckets)) { + retainingPartitions.remove(physicalTablePath, retention); + return; + } + + if (!isFullyTiered(tablePartition, frozenBuckets)) { + completeOperation(physicalTablePath, retention, RetentionStage.FROZEN); + LOG.info( + "Partition {} of table {} is frozen and waiting for lake tiering.", + partitionName, + tableInfo.getTablePath()); + return; + } + if (!leadersUnchanged(tablePartition, frozenBuckets)) { + retainingPartitions.remove(physicalTablePath, retention); + return; + } + + synchronized (retention) { + if (retainingPartitions.get(physicalTablePath) != retention) { + return; + } + retention.stage = RetentionStage.LAKE_TIERED; + } + dropPartition(tableInfo, partitionName, physicalTablePath, retention); + } catch (Exception e) { + completeOperation(physicalTablePath, retention, RetentionStage.FROZEN); + LOG.warn( + "Failed to check lake tiering for partition {} of table {}.", + partitionName, + tableInfo.getTablePath(), + e); + } + } + + private void submitDrop( + TableInfo tableInfo, + String partitionName, + PhysicalTablePath physicalTablePath, + RetentionStatus retention) { + if (closed.get()) { + retainingPartitions.remove(physicalTablePath, retention); + return; + } + try { + executor.execute( + () -> dropPartition(tableInfo, partitionName, physicalTablePath, retention)); + } catch (RuntimeException e) { + completeOperation(physicalTablePath, retention, RetentionStage.LAKE_TIERED); + LOG.warn( + "Failed to submit drop for retained partition {} of table {}.", + partitionName, + tableInfo.getTablePath(), + e); + } + } + + private void dropPartition( + TableInfo tableInfo, + String partitionName, + PhysicalTablePath physicalTablePath, + RetentionStatus retention) { + try { + Optional registration = + metadataManager.getPartitionRegistration( + tableInfo.getTablePath(), partitionName); + if (!registration.isPresent() + || registration.get().getTableId() != tableInfo.getTableId()) { + retainingPartitions.remove(physicalTablePath, retention); + return; + } + + Map frozenBuckets = registration.get().getFrozenBuckets(); + TablePartition tablePartition = registration.get().toTablePartition(); + if (!leadersUnchanged(tablePartition, frozenBuckets)) { + retainingPartitions.remove(physicalTablePath, retention); + return; + } + if (!isFullyTiered(tablePartition, frozenBuckets)) { + completeOperation(physicalTablePath, retention, RetentionStage.FROZEN); + return; + } + + metadataManager.dropPartition( + tableInfo.getTablePath(), + ResolvedPartitionSpec.fromPartitionName( + tableInfo.getPartitionKeys(), partitionName), + false); + retainingPartitions.remove(physicalTablePath, retention); + LOG.info( + "Deleted partition {} of table {} after all frozen offsets were committed to the lake.", + partitionName, + tableInfo.getTablePath()); + } catch (Exception e) { + completeOperation(physicalTablePath, retention, RetentionStage.LAKE_TIERED); + LOG.warn( + "Failed to drop retained partition {} of table {}.", + partitionName, + tableInfo.getTablePath(), + e); + } + } + + private void completeOperation( + PhysicalTablePath physicalTablePath, + RetentionStatus retention, + RetentionStage nextStage) { + synchronized (retention) { + if (retainingPartitions.get(physicalTablePath) == retention) { + retention.stage = nextStage; + retention.operationRunning = false; + } + } + } + + private Map collectFrozenBuckets( + Map expectedLeaders, + List> freezeFutures) { + Map frozenBuckets = new HashMap<>(); + for (CompletableFuture freezeFuture : freezeFutures) { + FreezePartitionResponse response = freezeFuture.join(); + for (PbFreezePartitionRespForBucket bucketResponse : response.getBucketsRespsList()) { + TableBucket tableBucket = toTableBucket(bucketResponse.getTableBucket()); + LeaderAndIsr leaderAndIsr = expectedLeaders.get(tableBucket); + if (leaderAndIsr == null + || bucketResponse.hasErrorCode() + || !bucketResponse.hasHighWatermark() + || !bucketResponse.hasLogEndOffset()) { + throw new FlussRuntimeException( + "Failed to freeze bucket " + + tableBucket + + ": " + + (bucketResponse.hasErrorMessage() + ? bucketResponse.getErrorMessage() + : "invalid response")); + } + frozenBuckets.put( + tableBucket.getBucket(), + new FrozenBucket( + leaderAndIsr.leader(), + leaderAndIsr.leaderEpoch(), + bucketResponse.getHighWatermark(), + bucketResponse.getLogEndOffset())); + } + } + if (frozenBuckets.size() != expectedLeaders.size()) { + throw new FlussRuntimeException( + String.format( + "Only %s of %s partition buckets were frozen.", + frozenBuckets.size(), expectedLeaders.size())); + } + return frozenBuckets; + } + + private Map mergeFrozenOffsets( + PartitionRegistration registration, Map frozenBuckets) { + Map merged = new HashMap<>(frozenBuckets); + registration + .getFrozenBuckets() + .forEach( + (bucketId, previous) -> { + FrozenBucket current = merged.get(bucketId); + if (current != null) { + merged.put( + bucketId, + new FrozenBucket( + current.getLeaderId(), + current.getLeaderEpoch(), + Math.max( + current.getHighWatermark(), + previous.getHighWatermark()), + Math.max( + current.getLogEndOffset(), + previous.getLogEndOffset()))); + } + }); + return merged; + } + + private boolean leadersUnchanged(Map expectedLeaders) + throws Exception { + Map currentLeaders = + zooKeeperClient.getLeaderAndIsrs(expectedLeaders.keySet()); + if (currentLeaders.size() != expectedLeaders.size()) { + return false; + } + for (Map.Entry entry : expectedLeaders.entrySet()) { + LeaderAndIsr current = currentLeaders.get(entry.getKey()); + LeaderAndIsr expected = entry.getValue(); + if (current == null + || current.leader() != expected.leader() + || current.leaderEpoch() != expected.leaderEpoch()) { + return false; + } + } + return true; + } + + private boolean leadersUnchanged( + TablePartition tablePartition, Map frozenBuckets) + throws Exception { + Map expectedLeaders = new HashMap<>(); + frozenBuckets.forEach( + (bucketId, frozenBucket) -> + expectedLeaders.put( + new TableBucket( + tablePartition.getTableId(), + tablePartition.getPartitionId(), + bucketId), + frozenBucket)); + + Map currentLeaders = + zooKeeperClient.getLeaderAndIsrs(expectedLeaders.keySet()); + if (currentLeaders.size() != expectedLeaders.size()) { + return false; + } + for (Map.Entry entry : expectedLeaders.entrySet()) { + LeaderAndIsr current = currentLeaders.get(entry.getKey()); + FrozenBucket expected = entry.getValue(); + if (current == null + || current.leader() != expected.getLeaderId() + || current.leaderEpoch() != expected.getLeaderEpoch()) { + return false; + } + } + return true; + } + + private boolean isFullyTiered( + TablePartition tablePartition, Map frozenBuckets) + throws Exception { + Optional lakeSnapshot = + zooKeeperClient.getLakeTableSnapshot(tablePartition.getTableId(), null); + if (!lakeSnapshot.isPresent()) { + return false; + } + for (Map.Entry entry : frozenBuckets.entrySet()) { + TableBucket tableBucket = + new TableBucket( + tablePartition.getTableId(), + tablePartition.getPartitionId(), + entry.getKey()); + long tieredOffset = lakeSnapshot.get().getLogEndOffset(tableBucket).orElse(0L); + if (tieredOffset < entry.getValue().getLogEndOffset()) { + return false; + } + } + return true; + } + + @Override + public void close() { + if (closed.compareAndSet(false, true)) { + retainingPartitions.clear(); + } + } + + private enum RetentionStage { + FREEZING, + FROZEN, + LAKE_TIERED + } + + private static class RetentionStatus { + private RetentionStage stage = RetentionStage.FREEZING; + private boolean operationRunning = true; + } +} diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java index 48533ea0d8f..bc4fe369344 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java @@ -923,6 +923,25 @@ Optional getOptionalPartitionRegistration( } } + Optional getPartitionRegistration( + TablePath tablePath, String partitionName) { + return getOptionalPartitionRegistration(tablePath, partitionName); + } + + void updatePartitionRegistration( + TablePath tablePath, + String partitionName, + PartitionRegistration partitionRegistration) { + try { + zookeeperClient.updatePartition(tablePath, partitionName, partitionRegistration); + } catch (Exception e) { + throw new FlussRuntimeException( + String.format( + "Fail to update partition '%s' of table %s.", partitionName, tablePath), + e); + } + } + private void rethrowIfIsNotNoNodeException( ThrowingRunnable throwingRunnable, String exceptionMessage) { try { diff --git a/fluss-server/src/main/java/org/apache/fluss/server/entity/FreezePartitionResultForBucket.java b/fluss-server/src/main/java/org/apache/fluss/server/entity/FreezePartitionResultForBucket.java new file mode 100644 index 00000000000..6ba4c1ceb7e --- /dev/null +++ b/fluss-server/src/main/java/org/apache/fluss/server/entity/FreezePartitionResultForBucket.java @@ -0,0 +1,50 @@ +/* + * 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.entity; + +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.rpc.entity.ResultForBucket; +import org.apache.fluss.rpc.protocol.ApiError; + +/** The offsets of a bucket after its leader has frozen writes for partition retention. */ +public class FreezePartitionResultForBucket extends ResultForBucket { + + private final long highWatermark; + private final long logEndOffset; + + public FreezePartitionResultForBucket( + TableBucket tableBucket, long highWatermark, long logEndOffset) { + super(tableBucket); + this.highWatermark = highWatermark; + this.logEndOffset = logEndOffset; + } + + public FreezePartitionResultForBucket(TableBucket tableBucket, ApiError error) { + super(tableBucket, error); + this.highWatermark = -1L; + this.logEndOffset = -1L; + } + + public long getHighWatermark() { + return highWatermark; + } + + public long getLogEndOffset() { + return logEndOffset; + } +} 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 f3d1ff76af1..396dc3fd82f 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 @@ -23,6 +23,7 @@ import org.apache.fluss.config.TableConfig; import org.apache.fluss.exception.FencedLeaderEpochException; import org.apache.fluss.exception.InvalidColumnProjectionException; +import org.apache.fluss.exception.InvalidPartitionException; import org.apache.fluss.exception.InvalidTableException; import org.apache.fluss.exception.InvalidTimestampException; import org.apache.fluss.exception.InvalidUpdateVersionException; @@ -113,6 +114,7 @@ import org.slf4j.LoggerFactory; import javax.annotation.Nullable; +import javax.annotation.concurrent.GuardedBy; import javax.annotation.concurrent.ThreadSafe; import java.io.Closeable; @@ -210,6 +212,9 @@ public final class Replica { private volatile int coordinatorEpoch = CoordinatorContext.INITIAL_COORDINATOR_EPOCH; private volatile boolean isStandbyReplica = false; + @GuardedBy("leaderIsrUpdateLock") + private boolean frozenForRetention = false; + // null if table without pk or haven't become leader private volatile @Nullable KvTablet kvTablet; private volatile @Nullable CloseableRegistry closeableRegistryForKv; @@ -1046,6 +1051,7 @@ public LogAppendInfo appendRecordsToLeader(MemoryLogRecords memoryLogRecords, in "Leader not local for bucket %s on tabletServer %d", tableBucket, localTabletServerId)); } + checkNotFrozenForRetention(); validateInSyncReplicaSize(requiredAcks); @@ -1085,6 +1091,7 @@ public LogAppendInfo putRecordsToLeader( "Leader not local for bucket %s on tabletServer %d", tableBucket, localTabletServerId)); } + checkNotFrozenForRetention(); validateInSyncReplicaSize(requiredAcks); KvTablet kv = this.kvTablet; @@ -1105,6 +1112,57 @@ public LogAppendInfo putRecordsToLeader( }); } + /** Freeze leader writes and return a stable offset boundary for partition retention. */ + public FrozenOffsets freezeForRetention(int expectedLeaderEpoch) { + return inWriteLock( + leaderIsrUpdateLock, + () -> { + if (!isLeader()) { + throw new NotLeaderOrFollowerException( + String.format( + "Leader not local for bucket %s on tabletServer %d", + tableBucket, localTabletServerId)); + } + if (leaderEpoch != expectedLeaderEpoch) { + throw new FencedLeaderEpochException( + String.format( + "Expected leader epoch %s for bucket %s, but current epoch is %s.", + expectedLeaderEpoch, tableBucket, leaderEpoch)); + } + frozenForRetention = true; + return new FrozenOffsets( + logTablet.getHighWatermark(), logTablet.localLogEndOffset()); + }); + } + + private void checkNotFrozenForRetention() { + if (frozenForRetention) { + throw new InvalidPartitionException( + String.format( + "Partition %s is frozen for retention and no longer accepts writes.", + physicalPath)); + } + } + + /** High watermark and log end offset captured while leader writes are fenced. */ + public static final class FrozenOffsets { + private final long highWatermark; + private final long logEndOffset; + + private FrozenOffsets(long highWatermark, long logEndOffset) { + this.highWatermark = highWatermark; + this.logEndOffset = logEndOffset; + } + + public long getHighWatermark() { + return highWatermark; + } + + public long getLogEndOffset() { + return logEndOffset; + } + } + public LogReadInfo fetchRecords(FetchParams fetchParams) throws IOException { if (fetchParams.projection() != null && logFormat != LogFormat.ARROW) { throw new InvalidColumnProjectionException( 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 88d5840e5a7..79f4544dff7 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 @@ -69,6 +69,7 @@ import org.apache.fluss.rpc.protocol.MergeMode; import org.apache.fluss.server.coordinator.CoordinatorContext; import org.apache.fluss.server.entity.FetchReqInfo; +import org.apache.fluss.server.entity.FreezePartitionResultForBucket; import org.apache.fluss.server.entity.LakeBucketOffset; import org.apache.fluss.server.entity.NotifyKvSnapshotOffsetData; import org.apache.fluss.server.entity.NotifyLakeTableOffsetData; @@ -644,6 +645,37 @@ public void appendRecordsToLog( timeoutMs, requiredAcks, entriesPerBucket.size(), appendResult, responseCallback); } + /** Freeze writes to partition bucket leaders and return their stable offsets. */ + public void freezePartitions( + int requestCoordinatorEpoch, + Map leaderEpochs, + Consumer> responseCallback) { + List results = new ArrayList<>(); + inLock( + replicaStateChangeLock, + () -> { + validateAndApplyCoordinatorEpoch(requestCoordinatorEpoch, "freezePartitions"); + for (Map.Entry entry : leaderEpochs.entrySet()) { + TableBucket tableBucket = entry.getKey(); + try { + Replica.FrozenOffsets frozenOffsets = + getReplicaOrException(tableBucket) + .freezeForRetention(entry.getValue()); + results.add( + new FreezePartitionResultForBucket( + tableBucket, + frozenOffsets.getHighWatermark(), + frozenOffsets.getLogEndOffset())); + } catch (Exception e) { + results.add( + new FreezePartitionResultForBucket( + tableBucket, ApiError.fromThrowable(e))); + } + } + }); + responseCallback.accept(results); + } + /** * Fetch records from a replica. Currently, we will return the fetched records immediately. * diff --git a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java index 0c2058915d6..98a542838ff 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java @@ -39,6 +39,8 @@ import org.apache.fluss.rpc.gateway.TabletServerGateway; import org.apache.fluss.rpc.messages.FetchLogRequest; import org.apache.fluss.rpc.messages.FetchLogResponse; +import org.apache.fluss.rpc.messages.FreezePartitionRequest; +import org.apache.fluss.rpc.messages.FreezePartitionResponse; import org.apache.fluss.rpc.messages.GetClusterHealthRequest; import org.apache.fluss.rpc.messages.GetClusterHealthResponse; import org.apache.fluss.rpc.messages.GetTableStatsRequest; @@ -126,6 +128,7 @@ import static org.apache.fluss.server.coordinator.CoordinatorContext.INITIAL_COORDINATOR_EPOCH; import static org.apache.fluss.server.log.FetchParams.DEFAULT_MAX_WAIT_MS_WHEN_MIN_BYTES_ENABLE; import static org.apache.fluss.server.utils.ServerRpcMessageUtils.getFetchLogData; +import static org.apache.fluss.server.utils.ServerRpcMessageUtils.getFreezePartitionData; import static org.apache.fluss.server.utils.ServerRpcMessageUtils.getListOffsetsData; import static org.apache.fluss.server.utils.ServerRpcMessageUtils.getNotifyLakeTableOffset; import static org.apache.fluss.server.utils.ServerRpcMessageUtils.getNotifyLeaderAndIsrRequestData; @@ -139,6 +142,7 @@ import static org.apache.fluss.server.utils.ServerRpcMessageUtils.getTargetColumns; import static org.apache.fluss.server.utils.ServerRpcMessageUtils.getUpdateMetadataRequestData; import static org.apache.fluss.server.utils.ServerRpcMessageUtils.makeFetchLogResponse; +import static org.apache.fluss.server.utils.ServerRpcMessageUtils.makeFreezePartitionResponse; import static org.apache.fluss.server.utils.ServerRpcMessageUtils.makeGetTableStatsResponse; import static org.apache.fluss.server.utils.ServerRpcMessageUtils.makeInitWriterResponse; import static org.apache.fluss.server.utils.ServerRpcMessageUtils.makeLimitScanResponse; @@ -432,6 +436,19 @@ public CompletableFuture stopReplica( value -> result.complete(makeStopReplicaResponse(value)))); } + @Override + public CompletableFuture freezePartition( + FreezePartitionRequest request) { + CompletableFuture response = new CompletableFuture<>(); + return submitReplicaStateChange( + response, + result -> + replicaManager.freezePartitions( + request.getCoordinatorEpoch(), + getFreezePartitionData(request), + results -> result.complete(makeFreezePartitionResponse(results)))); + } + @Override public CompletableFuture listOffsets(ListOffsetsRequest request) { authorizeTable(DESCRIBE, request.getTableId()); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/RpcGatewayManager.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/RpcGatewayManager.java index c89fca6f765..1d4bf99c59f 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/utils/RpcGatewayManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/RpcGatewayManager.java @@ -26,7 +26,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import javax.annotation.concurrent.NotThreadSafe; +import javax.annotation.concurrent.ThreadSafe; import java.util.HashMap; import java.util.Map; @@ -34,7 +34,7 @@ import java.util.concurrent.CompletableFuture; /** A manager to manage the rpc gateways to the servers. */ -@NotThreadSafe +@ThreadSafe public class RpcGatewayManager implements AutoCloseable { private static final Logger LOG = LoggerFactory.getLogger(RpcGatewayManager.class); @@ -57,7 +57,7 @@ public RpcGatewayManager(RpcClient rpcClient, Class gatewayClass) { * @param serverUid the id of the server * @return the rpc gateway of the server, empty if the server doesn't exist */ - public Optional getRpcGateway(int serverUid) { + public synchronized Optional getRpcGateway(int serverUid) { if (!serverRpcGateways.containsKey(serverUid)) { return Optional.empty(); } @@ -69,7 +69,7 @@ public Optional getRpcGateway(int serverUid) { * manager. If the server has already existed, it'll remove the already existing server before * adding the new one. */ - public void addServer(ServerNode serverNode) { + public synchronized void addServer(ServerNode serverNode) { int serverId = serverNode.id(); if (serverRpcGateways.containsKey(serverId)) { // close the already existing server @@ -95,7 +95,7 @@ public void addServer(ServerNode serverNode) { * @param serverId the id of the server to be removed * @return a future to be completed when the disconnection is complete */ - public CompletableFuture removeServer(int serverId) { + public synchronized CompletableFuture removeServer(int serverId) { ServerRpcGateway serverRpcGateway = serverRpcGateways.remove(serverId); if (serverRpcGateway != null) { return rpcClient.disconnect(serverRpcGateway.serverUid); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java index 1e5e914c0d2..ce89651e078 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java @@ -76,6 +76,8 @@ import org.apache.fluss.rpc.messages.DropAclsResponse; import org.apache.fluss.rpc.messages.FetchLogRequest; import org.apache.fluss.rpc.messages.FetchLogResponse; +import org.apache.fluss.rpc.messages.FreezePartitionRequest; +import org.apache.fluss.rpc.messages.FreezePartitionResponse; import org.apache.fluss.rpc.messages.GetFileSystemSecurityTokenResponse; import org.apache.fluss.rpc.messages.GetKvSnapshotMetadataResponse; import org.apache.fluss.rpc.messages.GetLakeSnapshotResponse; @@ -119,6 +121,8 @@ import org.apache.fluss.rpc.messages.PbFetchLogReqForTable; import org.apache.fluss.rpc.messages.PbFetchLogRespForBucket; import org.apache.fluss.rpc.messages.PbFetchLogRespForTable; +import org.apache.fluss.rpc.messages.PbFreezePartitionReqForBucket; +import org.apache.fluss.rpc.messages.PbFreezePartitionRespForBucket; import org.apache.fluss.rpc.messages.PbKeyValue; import org.apache.fluss.rpc.messages.PbKvSnapshot; import org.apache.fluss.rpc.messages.PbKvSnapshotLeaseForBucket; @@ -181,6 +185,7 @@ import org.apache.fluss.server.entity.CommitLakeTableSnapshotsData; import org.apache.fluss.server.entity.CommitRemoteLogManifestData; import org.apache.fluss.server.entity.FetchReqInfo; +import org.apache.fluss.server.entity.FreezePartitionResultForBucket; import org.apache.fluss.server.entity.LakeBucketOffset; import org.apache.fluss.server.entity.NotifyKvSnapshotOffsetData; import org.apache.fluss.server.entity.NotifyLakeTableOffsetData; @@ -890,6 +895,49 @@ public static StopReplicaResponse makeStopReplicaResponse( return stopReplicaResponse; } + public static FreezePartitionRequest makeFreezePartitionRequest( + int coordinatorEpoch, Map leaderEpochs) { + FreezePartitionRequest request = + new FreezePartitionRequest().setCoordinatorEpoch(coordinatorEpoch); + List bucketRequests = new ArrayList<>(); + leaderEpochs.forEach( + (tableBucket, leaderEpoch) -> + bucketRequests.add( + new PbFreezePartitionReqForBucket() + .setTableBucket(fromTableBucket(tableBucket)) + .setLeaderEpoch(leaderEpoch))); + return request.addAllBucketsReqs(bucketRequests); + } + + public static Map getFreezePartitionData(FreezePartitionRequest request) { + Map leaderEpochs = new HashMap<>(); + for (PbFreezePartitionReqForBucket bucketRequest : request.getBucketsReqsList()) { + leaderEpochs.put( + toTableBucket(bucketRequest.getTableBucket()), bucketRequest.getLeaderEpoch()); + } + return leaderEpochs; + } + + public static FreezePartitionResponse makeFreezePartitionResponse( + List results) { + FreezePartitionResponse response = new FreezePartitionResponse(); + List bucketResponses = new ArrayList<>(); + for (FreezePartitionResultForBucket result : results) { + PbFreezePartitionRespForBucket bucketResponse = + new PbFreezePartitionRespForBucket() + .setTableBucket(fromTableBucket(result.getTableBucket())); + if (result.failed()) { + bucketResponse.setError(result.getErrorCode(), result.getErrorMessage()); + } else { + bucketResponse + .setHighWatermark(result.getHighWatermark()) + .setLogEndOffset(result.getLogEndOffset()); + } + bucketResponses.add(bucketResponse); + } + return response.addAllBucketsResps(bucketResponses); + } + public static Map getProduceLogData( ProduceLogRequest produceRequest) { long tableId = produceRequest.getTableId(); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java index 8a6e9fa2c9e..d421c029535 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java @@ -466,6 +466,17 @@ private static void checkPartition( boolean hasExplicitTimeFormat = tableConf.contains(ConfigOptions.TABLE_AUTO_PARTITION_TIME_FORMAT); + if (autoPartition.ensureLakeTieredBeforeDrop() + && (!autoPartition.isAutoPartitionEnabled() + || !tableConf.get(ConfigOptions.TABLE_DATALAKE_ENABLED))) { + throw new InvalidConfigException( + String.format( + "'%s' requires both '%s' and '%s' to be enabled.", + ConfigOptions.TABLE_AUTO_PARTITION_RETENTION_ENSURE_TIERED.key(), + ConfigOptions.TABLE_AUTO_PARTITION_ENABLED.key(), + ConfigOptions.TABLE_DATALAKE_ENABLED.key())); + } + if (!isPartitioned && autoPartition.isAutoPartitionEnabled()) { throw new InvalidConfigException( String.format( 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 e9eab00e003..dd1f0bbf6f6 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 @@ -924,6 +924,14 @@ public Map getPartitionRegistrations(TablePath ta return partitions; } + /** Update partition metadata. */ + public void updatePartition( + TablePath tablePath, String partitionName, PartitionRegistration partitionRegistration) + throws Exception { + String path = PartitionZNode.path(tablePath, partitionName); + zkClient.setData().forPath(path, PartitionZNode.encode(partitionRegistration)); + } + /** Get the partition and the id for the partitions of tables in ZK. */ public Map> getPartitionNameAndIdsForTables( List tablePaths) throws Exception { diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/PartitionRegistration.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/PartitionRegistration.java index 2aef5858f82..8fdbac90cc2 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/PartitionRegistration.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/PartitionRegistration.java @@ -22,6 +22,9 @@ import javax.annotation.Nullable; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; import java.util.Objects; /** @@ -45,10 +48,22 @@ public class PartitionRegistration { */ private final @Nullable String remoteDataDir; + /** Bucket offsets captured after all partition leaders have fenced writes for retention. */ + private final Map frozenBuckets; + public PartitionRegistration(long tableId, long partitionId, @Nullable String remoteDataDir) { + this(tableId, partitionId, remoteDataDir, Collections.emptyMap()); + } + + public PartitionRegistration( + long tableId, + long partitionId, + @Nullable String remoteDataDir, + Map frozenBuckets) { this.tableId = tableId; this.partitionId = partitionId; this.remoteDataDir = remoteDataDir; + this.frozenBuckets = Collections.unmodifiableMap(new HashMap<>(frozenBuckets)); } public long getTableId() { @@ -64,6 +79,14 @@ public String getRemoteDataDir() { return remoteDataDir; } + public Map getFrozenBuckets() { + return frozenBuckets; + } + + public boolean isFrozenForRetention() { + return !frozenBuckets.isEmpty(); + } + public TablePartition toTablePartition() { return new TablePartition(tableId, partitionId); } @@ -77,7 +100,11 @@ public TablePartition toTablePartition() { * @return a new registration with the given remote data directory */ public PartitionRegistration newRemoteDataDir(String remoteDataDir) { - return new PartitionRegistration(tableId, partitionId, remoteDataDir); + return new PartitionRegistration(tableId, partitionId, remoteDataDir, frozenBuckets); + } + + public PartitionRegistration withFrozenBuckets(Map frozenBuckets) { + return new PartitionRegistration(tableId, partitionId, remoteDataDir, frozenBuckets); } @Override @@ -88,12 +115,13 @@ public boolean equals(Object o) { PartitionRegistration that = (PartitionRegistration) o; return tableId == that.tableId && partitionId == that.partitionId - && Objects.equals(remoteDataDir, that.remoteDataDir); + && Objects.equals(remoteDataDir, that.remoteDataDir) + && Objects.equals(frozenBuckets, that.frozenBuckets); } @Override public int hashCode() { - return Objects.hash(tableId, partitionId, remoteDataDir); + return Objects.hash(tableId, partitionId, remoteDataDir, frozenBuckets); } @Override @@ -106,6 +134,73 @@ public String toString() { + ", remoteDataDir='" + remoteDataDir + '\'' + + ", frozenBuckets=" + + frozenBuckets + '}'; } + + /** The leader and offsets captured when a bucket is frozen for partition retention. */ + public static final class FrozenBucket { + private final int leaderId; + private final int leaderEpoch; + private final long highWatermark; + private final long logEndOffset; + + public FrozenBucket(int leaderId, int leaderEpoch, long highWatermark, long logEndOffset) { + this.leaderId = leaderId; + this.leaderEpoch = leaderEpoch; + this.highWatermark = highWatermark; + this.logEndOffset = logEndOffset; + } + + public int getLeaderId() { + return leaderId; + } + + public int getLeaderEpoch() { + return leaderEpoch; + } + + public long getHighWatermark() { + return highWatermark; + } + + public long getLogEndOffset() { + return logEndOffset; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + FrozenBucket that = (FrozenBucket) o; + return leaderId == that.leaderId + && leaderEpoch == that.leaderEpoch + && highWatermark == that.highWatermark + && logEndOffset == that.logEndOffset; + } + + @Override + public int hashCode() { + return Objects.hash(leaderId, leaderEpoch, highWatermark, logEndOffset); + } + + @Override + public String toString() { + return "FrozenBucket{" + + "leaderId=" + + leaderId + + ", leaderEpoch=" + + leaderEpoch + + ", highWatermark=" + + highWatermark + + ", logEndOffset=" + + logEndOffset + + '}'; + } + } } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/PartitionRegistrationJsonSerde.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/PartitionRegistrationJsonSerde.java index ad83b18a079..ef47fd0a1c9 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/PartitionRegistrationJsonSerde.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/PartitionRegistrationJsonSerde.java @@ -24,6 +24,9 @@ import org.apache.fluss.utils.json.JsonSerializer; import java.io.IOException; +import java.util.HashMap; +import java.util.Map; +import java.util.TreeMap; /** Json serializer and deserializer for {@link PartitionRegistration}. */ @Internal @@ -37,7 +40,13 @@ public class PartitionRegistrationJsonSerde private static final String TABLE_ID_KEY = "table_id"; private static final String PARTITION_ID_KEY = "partition_id"; private static final String REMOTE_DATA_DIR_KEY = "remote_data_dir"; - private static final int VERSION = 1; + private static final String FROZEN_BUCKETS_KEY = "retention_frozen_buckets"; + private static final String BUCKET_ID_KEY = "bucket_id"; + private static final String LEADER_ID_KEY = "leader_id"; + private static final String LEADER_EPOCH_KEY = "leader_epoch"; + private static final String HIGH_WATERMARK_KEY = "high_watermark"; + private static final String LOG_END_OFFSET_KEY = "log_end_offset"; + private static final int VERSION = 2; @Override public void serialize(PartitionRegistration registration, JsonGenerator generator) @@ -49,6 +58,21 @@ public void serialize(PartitionRegistration registration, JsonGenerator generato if (registration.getRemoteDataDir() != null) { generator.writeStringField(REMOTE_DATA_DIR_KEY, registration.getRemoteDataDir()); } + if (registration.isFrozenForRetention()) { + generator.writeArrayFieldStart(FROZEN_BUCKETS_KEY); + for (Map.Entry entry : + new TreeMap<>(registration.getFrozenBuckets()).entrySet()) { + PartitionRegistration.FrozenBucket frozenBucket = entry.getValue(); + generator.writeStartObject(); + generator.writeNumberField(BUCKET_ID_KEY, entry.getKey()); + generator.writeNumberField(LEADER_ID_KEY, frozenBucket.getLeaderId()); + generator.writeNumberField(LEADER_EPOCH_KEY, frozenBucket.getLeaderEpoch()); + generator.writeNumberField(HIGH_WATERMARK_KEY, frozenBucket.getHighWatermark()); + generator.writeNumberField(LOG_END_OFFSET_KEY, frozenBucket.getLogEndOffset()); + generator.writeEndObject(); + } + generator.writeEndArray(); + } generator.writeEndObject(); } @@ -62,6 +86,18 @@ public PartitionRegistration deserialize(JsonNode node) { if (node.has(REMOTE_DATA_DIR_KEY)) { remoteDataDir = node.get(REMOTE_DATA_DIR_KEY).asText(); } - return new PartitionRegistration(tableId, partitionId, remoteDataDir); + Map frozenBuckets = new HashMap<>(); + if (node.has(FROZEN_BUCKETS_KEY)) { + for (JsonNode frozenBucketNode : node.get(FROZEN_BUCKETS_KEY)) { + frozenBuckets.put( + frozenBucketNode.get(BUCKET_ID_KEY).asInt(), + new PartitionRegistration.FrozenBucket( + frozenBucketNode.get(LEADER_ID_KEY).asInt(), + frozenBucketNode.get(LEADER_EPOCH_KEY).asInt(), + frozenBucketNode.get(HIGH_WATERMARK_KEY).asLong(), + frozenBucketNode.get(LOG_END_OFFSET_KEY).asLong())); + } + } + return new PartitionRegistration(tableId, partitionId, remoteDataDir, frozenBuckets); } } diff --git a/fluss-server/src/test/java/org/apache/fluss/server/tablet/TestTabletServerGateway.java b/fluss-server/src/test/java/org/apache/fluss/server/tablet/TestTabletServerGateway.java index c78270ea5ca..84f03d1112c 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/tablet/TestTabletServerGateway.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/tablet/TestTabletServerGateway.java @@ -30,6 +30,8 @@ import org.apache.fluss.rpc.messages.DescribeClusterConfigsResponse; import org.apache.fluss.rpc.messages.FetchLogRequest; import org.apache.fluss.rpc.messages.FetchLogResponse; +import org.apache.fluss.rpc.messages.FreezePartitionRequest; +import org.apache.fluss.rpc.messages.FreezePartitionResponse; import org.apache.fluss.rpc.messages.GetClusterHealthRequest; import org.apache.fluss.rpc.messages.GetClusterHealthResponse; import org.apache.fluss.rpc.messages.GetDatabaseInfoRequest; @@ -349,6 +351,14 @@ public CompletableFuture notifyRemoteLogOffsets( return response; } + @Override + public CompletableFuture freezePartition( + FreezePartitionRequest request) { + CompletableFuture response = new CompletableFuture<>(); + requests.add(Tuple2.of(request, response)); + return response; + } + @Override public CompletableFuture notifyKvSnapshotOffset( NotifyKvSnapshotOffsetRequest request) { diff --git a/fluss-server/src/test/java/org/apache/fluss/server/zk/data/PartitionRegistrationJsonSerdeTest.java b/fluss-server/src/test/java/org/apache/fluss/server/zk/data/PartitionRegistrationJsonSerdeTest.java index 3676688f5c0..a1c70805165 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/zk/data/PartitionRegistrationJsonSerdeTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/zk/data/PartitionRegistrationJsonSerdeTest.java @@ -47,8 +47,8 @@ protected PartitionRegistration[] createObjects() { @Override protected String[] expectedJsons() { return new String[] { - "{\"version\":1,\"table_id\":1234,\"partition_id\":5678,\"remote_data_dir\":\"file://local/remote\"}", - "{\"version\":1,\"table_id\":246,\"partition_id\":135}" + "{\"version\":2,\"table_id\":1234,\"partition_id\":5678,\"remote_data_dir\":\"file://local/remote\"}", + "{\"version\":2,\"table_id\":246,\"partition_id\":135}" }; }