wuchong commented on code in PR #3174:
URL: https://github.com/apache/fluss/pull/3174#discussion_r3374761626
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java:
##########
@@ -229,9 +232,14 @@ public CoordinatorEventProcessor(
tableBucketStateMachine,
new RemoteStorageCleaner(conf, ioExecutor),
ioExecutor);
+ this.replicaCleanupManager =
+ new ReplicaCleanupManager(coordinatorEventManager, clock,
conf);
+ this.tableManager.setReplicaCleanupManager(replicaCleanupManager);
Review Comment:
We should move the `replicaCleanupManager` into the constructor of
`TableManager` to make it not null and final. This also simplify the logic of
`TableManager` without dealing with null `replicaCleanupManager`.
##########
fluss-server/src/test/java/org/apache/fluss/server/coordinator/event/watcher/TableChangeWatcherTest.java:
##########
@@ -109,7 +111,16 @@ void before() {
metadataManager.createDatabase(DEFAULT_DB,
DatabaseDescriptor.builder().build(), false);
eventManager = new TestingEventManager();
- tableChangeWatcher = new TableChangeWatcher(zookeeperClient,
eventManager);
+ // Use a real ReplicaCleanupManager so that drop submissions are
forwarded into the
+ // event manager as DropPartitionEvent / DropTableEvent (the existing
assertions in this
+ // test verify exactly that).
+ replicaCleanupManager =
+ new ReplicaCleanupManager(
+ eventManager,
+ org.apache.fluss.utils.clock.SystemClock.getInstance(),
+ new org.apache.fluss.config.Configuration());
Review Comment:
static import them and remove unnecessary qualifications.
##########
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;
+ }
+ PendingDrop toExecute;
+ lock.lock();
+ try {
+ // Deduplicate: skip if the same target is already in-flight or
pending.
+ // This prevents stale resume drops from accumulating when
resumeDeletions()
+ // is called while the original watcher-triggered drop is still in
progress.
+ if (isDuplicate(drop)) {
+ LOG.debug("Skipping duplicate drop submission: {}", drop);
+ return;
+ }
+ pendingDrops.add(drop);
+ LOG.debug(
+ "Submitted drop: {}; pending={} inflight={}",
+ drop,
+ pendingDrops.size(),
+ currentInflight != null);
+ toExecute = admitNext();
+ } finally {
+ lock.unlock();
+ }
+ executeOutsideLock(toExecute);
+ }
+
+ /**
+ * Returns {@code true} if a drop targeting the same table/partition is
already tracked. Must be
+ * called under lock.
+ */
+ @GuardedBy("lock")
+ private boolean isDuplicate(PendingDrop drop) {
+ if (currentInflight != null &&
currentInflight.drop.hasSameTarget(drop)) {
+ return true;
+ }
+ for (PendingDrop pending : pendingDrops) {
+ if (pending.hasSameTarget(drop)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * If nothing is in-flight, pulls the next pending drop and marks it as
in-flight. Must be
+ * called under lock.
+ *
+ * @return the drop to execute outside the lock, or {@code null} if
nothing was admitted.
+ */
+ @GuardedBy("lock")
+ @Nullable
+ private PendingDrop admitNext() {
+ if (currentInflight != null || pendingDrops.isEmpty()) {
+ return null;
+ }
+ PendingDrop next = pendingDrops.poll();
+ currentInflight = new InflightDrop(next, clock.milliseconds());
+ return next;
+ }
+
+ /** Executes a drop outside the lock. No-op if {@code drop} is null. */
+ private void executeOutsideLock(@Nullable PendingDrop drop) {
+ if (drop == null) {
+ return;
+ }
+ try {
+ drop.execute(eventManager);
+ // For fire-and-forget drops (e.g., auto-partition table drops),
no completion
+ // callback will ever arrive because there are no table-level
replicas to delete.
+ // Release the inflight slot immediately and admit the next
pending drop.
+ if (drop.isFireAndForget()) {
+ PendingDrop next;
+ lock.lock();
+ try {
+ currentInflight = null;
+ next = admitNext();
+ } finally {
+ lock.unlock();
+ }
+ executeOutsideLock(next);
+ }
+ } catch (Throwable t) {
+ LOG.error("Failed to execute drop {}; abandoning in-memory
tracking.", drop, t);
+ lock.lock();
+ try {
+ currentInflight = null;
+ // Try to admit the next pending drop despite the failure.
+ PendingDrop next = admitNext();
+ if (next != null) {
+ // Recursive execution outside lock; safe because
admitNext() won't
+ // return non-null again until this next drop completes.
+ executeOutsideLock(next);
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+ }
+
+ private void checkTimeoutsSafely() {
+ try {
+ checkTimeouts();
+ } catch (Throwable t) {
+ LOG.error("Unexpected error in ReplicaCleanupManager timeout
check.", t);
+ }
+ }
+
+ @VisibleForTesting
+ void checkTimeouts() {
+ PendingDrop next;
+ lock.lock();
+ try {
+ if (currentInflight == null) {
+ return;
+ }
+ long elapsed = clock.milliseconds() -
currentInflight.submittedAtMs;
+ if (elapsed <= inflightTimeoutMs) {
+ return;
+ }
+ LOG.warn(
+ "In-flight drop {} timed out after {}ms with no completion
callback. "
+ + "Abandoning in-memory tracking; ZK metadata is
already gone, "
+ + "any residual replica state will be reconciled
on next "
+ + "coordinator startup.",
+ currentInflight.drop,
+ elapsed);
+ currentInflight = null;
+ next = admitNext();
+ } finally {
+ lock.unlock();
+ }
+ executeOutsideLock(next);
Review Comment:
This timeout path can run a resumeAction from the timeout checker thread.
For resume drops, execute() calls TableManager.onDeleteTable/onDeletePartition
directly, and those paths mutate CoordinatorContext and the state machines,
while CoordinatorContext is @NotThreadSafe and normally only touched from the
coordinator event thread. Can we route resume actions through the coordinator
event thread as well, or otherwise make timeout admission only enqueue work?
##########
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;
+ }
+ PendingDrop toExecute;
+ lock.lock();
+ try {
+ // Deduplicate: skip if the same target is already in-flight or
pending.
+ // This prevents stale resume drops from accumulating when
resumeDeletions()
+ // is called while the original watcher-triggered drop is still in
progress.
+ if (isDuplicate(drop)) {
+ LOG.debug("Skipping duplicate drop submission: {}", drop);
+ return;
+ }
+ pendingDrops.add(drop);
+ LOG.debug(
+ "Submitted drop: {}; pending={} inflight={}",
+ drop,
+ pendingDrops.size(),
+ currentInflight != null);
+ toExecute = admitNext();
+ } finally {
+ lock.unlock();
+ }
+ executeOutsideLock(toExecute);
+ }
+
+ /**
+ * Returns {@code true} if a drop targeting the same table/partition is
already tracked. Must be
+ * called under lock.
+ */
+ @GuardedBy("lock")
+ private boolean isDuplicate(PendingDrop drop) {
+ if (currentInflight != null &&
currentInflight.drop.hasSameTarget(drop)) {
+ return true;
+ }
+ for (PendingDrop pending : pendingDrops) {
+ if (pending.hasSameTarget(drop)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * If nothing is in-flight, pulls the next pending drop and marks it as
in-flight. Must be
+ * called under lock.
+ *
+ * @return the drop to execute outside the lock, or {@code null} if
nothing was admitted.
+ */
+ @GuardedBy("lock")
+ @Nullable
+ private PendingDrop admitNext() {
+ if (currentInflight != null || pendingDrops.isEmpty()) {
+ return null;
+ }
+ PendingDrop next = pendingDrops.poll();
+ currentInflight = new InflightDrop(next, clock.milliseconds());
+ return next;
+ }
+
+ /** Executes a drop outside the lock. No-op if {@code drop} is null. */
+ private void executeOutsideLock(@Nullable PendingDrop drop) {
+ if (drop == null) {
+ return;
+ }
+ try {
+ drop.execute(eventManager);
+ // For fire-and-forget drops (e.g., auto-partition table drops),
no completion
+ // callback will ever arrive because there are no table-level
replicas to delete.
+ // Release the inflight slot immediately and admit the next
pending drop.
+ if (drop.isFireAndForget()) {
+ PendingDrop next;
+ lock.lock();
+ try {
+ currentInflight = null;
+ next = admitNext();
+ } finally {
+ lock.unlock();
+ }
+ executeOutsideLock(next);
+ }
+ } catch (Throwable t) {
+ LOG.error("Failed to execute drop {}; abandoning in-memory
tracking.", drop, t);
+ lock.lock();
+ try {
+ currentInflight = null;
+ // Try to admit the next pending drop despite the failure.
+ PendingDrop next = admitNext();
+ if (next != null) {
+ // Recursive execution outside lock; safe because
admitNext() won't
+ // return non-null again until this next drop completes.
+ executeOutsideLock(next);
+ }
Review Comment:
Why not execute `next` outside the lock? It's possible the `resumeAction` is
time consuming.
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/TableManager.java:
##########
@@ -66,6 +78,10 @@ public TableManager(
this.ioExecutor = ioExecutor;
}
+ public void setReplicaCleanupManager(@Nullable ReplicaCleanupManager
replicaCleanupManager) {
Review Comment:
Could we wire this through the constructor and make it final instead of
mutable/nullable? In production it is set before the event thread starts, but
TableManager itself does not enforce that. A later call to
setReplicaCleanupManager() would race with resumeDeletions()/completeDelete*()
because this field is neither final nor volatile. Constructor injection would
keep the existing threading assumptions explicit and avoid an unnecessary data
race surface.
##########
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:
I suggest refactoring the jittered partition creation logic in
`AutoPartitionManager` into this manager in the future. Since partition
creation and deletion involve similar workloads, they should share the same
underlying logic.
To reflect this broader scope, we could rename the class to
`TableLifecycleThrottler` or `PartitionLifecycleThrottler`. This naming better
captures the idea that partition creation and deletion are lifecycle events,
with table drop naturally included. Such a change would also allow us to
incorporate partition creation management more seamlessly in the future.
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/TableManager.java:
##########
@@ -208,45 +224,44 @@ public void resumeDeletions() {
private void resumeTableDeletions() {
Set<Long> tablesToBeDeleted = new
HashSet<>(coordinatorContext.getTablesToBeDeleted());
- Set<Long> eligibleTableDeletion = new HashSet<>();
-
for (long tableId : tablesToBeDeleted) {
- // if all replicas are marked as deleted successfully, then table
deletion is done
if (coordinatorContext.areAllReplicasInState(
tableId, ReplicaState.ReplicaDeletionSuccessful)) {
completeDeleteTable(tableId);
LOG.info("Deletion of table with id {} successfully
completed.", tableId);
- }
- if (isEligibleForDeletion(tableId)) {
- eligibleTableDeletion.add(tableId);
- }
- }
- if (!eligibleTableDeletion.isEmpty()) {
- for (long tableId : eligibleTableDeletion) {
- onDeleteTable(tableId);
+ } else if (isEligibleForDeletion(tableId)) {
+ if (replicaCleanupManager != null) {
+ replicaCleanupManager.submitTableDropForResume(
+ tableId, () -> onDeleteTable(tableId));
Review Comment:
I don't like the design of `resumeAction` in `replicaCleanupManager`,
because it can easily break the concurrency safety. As `replicaCleanupManager`
is invoked by multiple threads (timeout scheduler, zk watcher,
AutoPartitionManager), but the `onDeleteTable` is not thread-safe.
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/TableManager.java:
##########
@@ -208,45 +224,44 @@ public void resumeDeletions() {
private void resumeTableDeletions() {
Set<Long> tablesToBeDeleted = new
HashSet<>(coordinatorContext.getTablesToBeDeleted());
- Set<Long> eligibleTableDeletion = new HashSet<>();
-
for (long tableId : tablesToBeDeleted) {
- // if all replicas are marked as deleted successfully, then table
deletion is done
if (coordinatorContext.areAllReplicasInState(
tableId, ReplicaState.ReplicaDeletionSuccessful)) {
completeDeleteTable(tableId);
LOG.info("Deletion of table with id {} successfully
completed.", tableId);
- }
- if (isEligibleForDeletion(tableId)) {
- eligibleTableDeletion.add(tableId);
- }
- }
- if (!eligibleTableDeletion.isEmpty()) {
- for (long tableId : eligibleTableDeletion) {
- onDeleteTable(tableId);
+ } else if (isEligibleForDeletion(tableId)) {
+ if (replicaCleanupManager != null) {
+ replicaCleanupManager.submitTableDropForResume(
+ tableId, () -> onDeleteTable(tableId));
+ } else {
+ onDeleteTable(tableId);
+ }
}
}
}
private void resumePartitionDeletions() {
Set<TablePartition> partitionsToDelete =
new HashSet<>(coordinatorContext.getPartitionsToBeDeleted());
- Set<TablePartition> eligiblePartitionDeletion = new HashSet<>();
-
for (TablePartition partition : partitionsToDelete) {
// if all replicas are marked as deleted successfully, then
partition deletion is done
if (coordinatorContext.areAllReplicasInState(
partition, ReplicaState.ReplicaDeletionSuccessful)) {
completeDeletePartition(partition);
LOG.info("Deletion of partition {} successfully completed.",
partition);
- }
- if (isEligibleForDeletion(partition)) {
- eligiblePartitionDeletion.add(partition);
- }
- }
- if (!eligiblePartitionDeletion.isEmpty()) {
- for (TablePartition partition : eligiblePartitionDeletion) {
- onDeletePartition(partition.getTableId(),
partition.getPartitionId());
+ } else if (isEligibleForDeletion(partition)) {
+ long tableId = partition.getTableId();
+ long partitionId = partition.getPartitionId();
+ if (replicaCleanupManager != null) {
+ String partitionName =
coordinatorContext.getPartitionName(partitionId);
+ replicaCleanupManager.submitPartitionDropForResume(
+ tableId,
+ partitionId,
+ partitionName,
+ () -> onDeletePartition(tableId, partitionId));
Review Comment:
Why is `onDeletePartition` invoked after the partition has been fully
deleted? As I understand it, when a drop partition event reaches the
`CoordinatorEventProcessor`, it already triggers
`TableManager.onDeletePartition` in
`org.apache.fluss.server.coordinator.CoordinatorEventProcessor.processDropPartition`.
This leads to redundant state machine transitions.
If we can avoid calling `onDeletePartition` once the partition deletion is
complete, we would no longer need the `resumeAction` in
`PendingPartition/TableDrop`. This change would significantly simplify the
logic in `ReplicaCleanupManager`.
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/TableManager.java:
##########
@@ -51,6 +53,16 @@ public class TableManager {
private final TableBucketStateMachine tableBucketStateMachine;
private final ExecutorService ioExecutor;
+ /**
+ * Optional manager used to throttle the asynchronous replica cleanup that
follows {@link
+ * #onDeleteTable(long)} / {@link #onDeletePartition(long, long)}. When
set, {@link
+ * #resumeDeletions()} routes eligible drops through the manager (one at a
time) instead of
+ * invoking the state-machine transitions directly, and {@link
#completeDeleteTable(long)} /
+ * {@link #completeDeletePartition(TablePartition)} notify the manager so
the next pending drop
+ * can be admitted.
+ */
+ @Nullable private ReplicaCleanupManager replicaCleanupManager;
Review Comment:
should be not-null and final.
##########
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:
I don't fully understand this. Why we need to execute immediately rather
than relying on the throttle approach? Even if the manager has not started yet,
I think the functionality still works (i.e., put event to EventManager).
--
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]
