swuferhong commented on code in PR #3174: URL: https://github.com/apache/fluss/pull/3174#discussion_r3377430494
########## fluss-server/src/main/java/org/apache/fluss/server/coordinator/ReplicaCleanupManager.java: ########## @@ -0,0 +1,566 @@ +/* + * 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.annotation.VisibleForTesting; +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.metadata.TablePartition; +import org.apache.fluss.server.coordinator.event.DropPartitionEvent; +import org.apache.fluss.server.coordinator.event.DropTableEvent; +import org.apache.fluss.server.coordinator.event.EventManager; +import org.apache.fluss.utils.clock.Clock; +import org.apache.fluss.utils.clock.SystemClock; +import org.apache.fluss.utils.concurrent.ExecutorThreadFactory; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; +import javax.annotation.concurrent.GuardedBy; + +import java.util.ArrayDeque; +import java.util.Deque; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; + +import static org.apache.fluss.utils.concurrent.LockUtils.inLock; + +/** + * Coordinator-wide manager that gates the rate at which {@link DropPartitionEvent} / {@link + * DropTableEvent} are admitted into the coordinator event queue. + * + * <p>Why pre-queue throttling: a cascading {@code DROP TABLE} or auto-partition expiration can + * delete N partition znodes from ZooKeeper synchronously, which fans out into N {@code + * NODE_DELETED} watcher callbacks. If each callback directly enqueued a {@code DropPartitionEvent}, + * the coordinator event queue would be flooded with N drop events ahead of unrelated work like + * leader election, heartbeat handling and metadata changes. + * + * <p>This manager intercepts that step. Watchers (and other drop sources) call {@link + * #submitPartitionDrop} / {@link #submitTableDrop} which buffer a lightweight {@code PendingDrop} + * in an in-memory FIFO queue. The manager admits <b>one</b> drop at a time into the coordinator + * event queue. The next drop is admitted only after the current in-flight drop completes (all + * replicas reach {@code DeletionSuccessful}) or times out. + * + * <p>Coordinator startup also routes through this manager via {@link #submitPartitionDropForResume} + * / {@link #submitTableDropForResume}: stale tables/partitions detected by {@code + * initCoordinatorContext} are submitted with a {@link Runnable} that drives {@code + * TableManager#onDeleteTable} / {@code TableManager#onDeletePartition} directly (their {@code + * TableInfo} is already gone, so a regular {@code DropTableEvent} is not viable). + * + * <p>Timeout-based abandon: the in-flight drop is timestamped at admission time. A periodic task + * checks whether the completion callback has not arrived within the hardcoded timeout (3 minutes) + * and abandons the drop with a WARN log. Abandonment only releases the manager's in-memory + * tracking; any residual replica state machine entries are reconciled on the next coordinator + * startup via {@link TableManager#resumeDeletions()}. + */ +public class ReplicaCleanupManager implements AutoCloseable { Review Comment: Ok. I will rename the class to `TableLifecycleThrottler`, and the work of refactoring the jittered partition creation logic in AutoPartitionManager into this manager will be traced by: https://github.com/apache/fluss/issues/3457 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
