swuferhong commented on code in PR #3174: URL: https://github.com/apache/fluss/pull/3174#discussion_r3377499856
########## 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 { + + private static final Logger LOG = LoggerFactory.getLogger(ReplicaCleanupManager.class); + + private final long inflightTimeoutMs; + private final long timeoutCheckIntervalMs; + + private final EventManager eventManager; + private final Clock clock; + private final ScheduledExecutorService timeoutChecker; + + private final AtomicBoolean started = new AtomicBoolean(false); + private final AtomicBoolean closed = new AtomicBoolean(false); + + private final Lock lock = new ReentrantLock(); + + /** Pending drops waiting to be admitted into the coordinator event queue. */ + @GuardedBy("lock") + private final Deque<PendingDrop> pendingDrops = new ArrayDeque<>(); + + /** The currently in-flight drop, or {@code null} if nothing is in-flight. */ + @GuardedBy("lock") + @Nullable + private InflightDrop currentInflight; + + public ReplicaCleanupManager(EventManager eventManager, Clock clock, Configuration conf) { + this( + eventManager, + clock, + Executors.newScheduledThreadPool( + 1, new ExecutorThreadFactory("replica-cleanup-timeout")), + conf.get(ConfigOptions.COORDINATOR_REPLICA_CLEANUP_INFLIGHT_TIMEOUT).toMillis(), + conf.get(ConfigOptions.COORDINATOR_REPLICA_CLEANUP_TIMEOUT_CHECK_INTERVAL) + .toMillis()); + } + + @VisibleForTesting + ReplicaCleanupManager( + EventManager eventManager, + Clock clock, + ScheduledExecutorService timeoutChecker, + long inflightTimeoutMs, + long timeoutCheckIntervalMs) { + this.eventManager = eventManager; + this.clock = clock == null ? SystemClock.getInstance() : clock; + this.timeoutChecker = timeoutChecker; + this.inflightTimeoutMs = inflightTimeoutMs; + this.timeoutCheckIntervalMs = timeoutCheckIntervalMs; + } + + /** Starts the periodic timeout checker and activates one-at-a-time throttling. Idempotent. */ + public void start() { + if (closed.get()) { + throw new IllegalStateException("ReplicaCleanupManager is already closed."); + } + if (started.compareAndSet(false, true)) { + timeoutChecker.scheduleWithFixedDelay( + this::checkTimeoutsSafely, + timeoutCheckIntervalMs, + timeoutCheckIntervalMs, + TimeUnit.MILLISECONDS); + LOG.info( + "ReplicaCleanupManager started: one-at-a-time throttling, " + + "inflightTimeoutMs={}, timeoutCheckIntervalMs={}", + inflightTimeoutMs, + timeoutCheckIntervalMs); + } + } + + /** + * Submits a partition drop request. The partition's ZK metadata is expected to have been + * removed by the caller prior to this submission; the manager only governs how fast the + * resulting {@link DropPartitionEvent} is admitted into the coordinator event queue. + */ + public void submitPartitionDrop(long tableId, long partitionId, String partitionName) { + admit(new PendingPartitionDrop(tableId, partitionId, partitionName, null)); + } + + /** + * Submits a partition drop request that, when admitted, runs {@code resumeAction} instead of + * putting a {@link DropPartitionEvent} into the coordinator event queue. Used by coordinator + * startup to reconcile stale partitions whose {@code TableInfo} is already gone. + */ + public void submitPartitionDropForResume( + long tableId, long partitionId, String partitionName, Runnable resumeAction) { + admit(new PendingPartitionDrop(tableId, partitionId, partitionName, resumeAction)); + } + + /** + * Submits a table drop request. The table's ZK metadata is expected to have been removed by the + * caller prior to this submission. + */ + public void submitTableDrop( + long tableId, + boolean isPartitionedTable, + boolean isAutoPartitionTable, + boolean isDataLakeEnabled) { + admit( + new PendingTableDrop( + tableId, + isPartitionedTable, + isAutoPartitionTable, + isDataLakeEnabled, + null)); + } + + /** + * Submits a table drop request that, when admitted, runs {@code resumeAction} instead of + * putting a {@link DropTableEvent} into the coordinator event queue. Used by coordinator + * startup to reconcile stale tables whose {@code TableInfo} is already gone. + */ + public void submitTableDropForResume(long tableId, Runnable resumeAction) { + admit(new PendingTableDrop(tableId, false, false, false, resumeAction)); + } + + /** + * Called by {@link TableManager} when all replicas of a partition have reached the {@code + * DeletionSuccessful} state. Releases the in-flight slot and admits the next pending drop. + */ + public void onPartitionDropCompleted(TablePartition tablePartition) { + PendingDrop next; + lock.lock(); + try { + if (currentInflight == null) { + return; + } + PendingDrop inflight = currentInflight.drop; + if (!(inflight instanceof PendingPartitionDrop)) { + return; + } + PendingPartitionDrop p = (PendingPartitionDrop) inflight; + if (p.tableId != tablePartition.getTableId() + || p.partitionId != tablePartition.getPartitionId()) { + return; + } + LOG.debug( + "Partition drop completed: {} after {}ms; pending={}", + tablePartition, + clock.milliseconds() - currentInflight.submittedAtMs, + pendingDrops.size()); + currentInflight = null; + next = admitNext(); + } finally { + lock.unlock(); + } + executeOutsideLock(next); + } + + /** + * Called by {@link TableManager} when all replicas of a table have reached the {@code + * DeletionSuccessful} state. Releases the in-flight slot and admits the next pending drop. + */ + public void onTableDropCompleted(long tableId) { + PendingDrop next; + lock.lock(); + try { + if (currentInflight == null) { + return; + } + PendingDrop inflight = currentInflight.drop; + if (!(inflight instanceof PendingTableDrop)) { + return; + } + PendingTableDrop t = (PendingTableDrop) inflight; + if (t.tableId != tableId) { + return; + } + LOG.debug( + "Table drop completed: tableId={} after {}ms; pending={}", + tableId, + clock.milliseconds() - currentInflight.submittedAtMs, + pendingDrops.size()); + currentInflight = null; + next = admitNext(); + } finally { + lock.unlock(); + } + executeOutsideLock(next); + } + + private void admit(PendingDrop drop) { + if (closed.get()) { + LOG.debug( + "Ignoring drop submission ({}) because ReplicaCleanupManager is closed.", drop); + return; + } + // Before start() is called, throttling is inactive: execute immediately. + // This covers early watcher callbacks that fire before the manager is activated. + if (!started.get()) { + drop.execute(eventManager); + return; + } Review Comment: You're right — the `admitNext()` logic works correctly regardless of whether `start()` has been called: if nothing is in-flight, the first drop is admitted immediately. The timeout checker (activated by start()) only provides liveness as a safety net, not correctness. The early-return was actually counter-productive: if multiple watcher callbacks arrived before start(), they would all bypass throttling and flood the event queue — exactly the scenario the throttler is designed to prevent. I have Removed the block. All submissions now uniformly go through the lock → deduplicate → admitNext → executeOutsideLock path. -- 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]
