This is an automated email from the ASF dual-hosted git repository.
swuferhong pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new fa091e876 [server] Add TableLifecycleThrottler to throttle drops in
coordinator queue (#3174)
fa091e876 is described below
commit fa091e876d2cad97ddf889f1fbbf33d479b2eb91
Author: yunhong <[email protected]>
AuthorDate: Thu Jun 11 09:08:39 2026 +0800
[server] Add TableLifecycleThrottler to throttle drops in coordinator queue
(#3174)
* [server] Add ReplicaCleanupManager to throttle drops in coordinator queue
* address copilot's comments
* address yuxia's comments
* Fix unstable it case
* address jark's comments
* fix failed test case
* improve code
---------
Co-authored-by: Jark Wu <[email protected]>
---
.../org/apache/fluss/config/ConfigOptions.java | 21 +
.../server/coordinator/AutoPartitionManager.java | 6 +-
.../coordinator/CoordinatorEventProcessor.java | 37 +-
.../server/coordinator/CoordinatorServer.java | 3 +-
.../coordinator/TableLifecycleThrottler.java | 553 +++++++++++++++++++++
.../fluss/server/coordinator/TableManager.java | 48 +-
.../server/coordinator/event/ResumeDropEvent.java | 73 +++
.../event/watcher/TableChangeWatcher.java | 37 +-
.../coordinator/AutoPartitionManagerTest.java | 56 +++
.../coordinator/CoordinatorEventProcessorTest.java | 201 +++++++-
.../coordinator/TableLifecycleThrottlerTest.java | 430 ++++++++++++++++
.../fluss/server/coordinator/TableManagerTest.java | 34 +-
.../event/watcher/TableChangeWatcherTest.java | 40 +-
.../rebalance/RebalanceManagerTest.java | 3 +-
.../statemachine/TableBucketStateMachineTest.java | 3 +-
.../server/testutils/FlussClusterExtension.java | 8 +
website/docs/maintenance/configuration.md | 12 +-
17 files changed, 1507 insertions(+), 58 deletions(-)
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 3cb61db27..fafcc2905 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
@@ -200,6 +200,27 @@ public class ConfigOptions {
"The interval of auto partition check. "
+ "The default value is 10 minutes.");
+ public static final ConfigOption<Duration>
COORDINATOR_LIFECYCLE_THROTTLER_INFLIGHT_TIMEOUT =
+ key("coordinator.lifecycle-throttler.inflight-timeout")
+ .durationType()
+ .defaultValue(Duration.ofMinutes(3))
+ .withDescription(
+ "The timeout for an in-flight drop event in the
coordinator's "
+ + "TableLifecycleThrottler. If a drop
event has been admitted "
+ + "but the corresponding completion
callback has not arrived "
+ + "within this timeout, the throttler
abandons tracking of "
+ + "that drop and continues admitting the
next pending drop.");
+
+ public static final ConfigOption<Duration>
+ COORDINATOR_LIFECYCLE_THROTTLER_TIMEOUT_CHECK_INTERVAL =
+
key("coordinator.lifecycle-throttler.timeout-check-interval")
+ .durationType()
+ .defaultValue(Duration.ofMinutes(1))
+ .withDescription(
+ "The periodic interval at which the
coordinator's "
+ + "TableLifecycleThrottler scans
in-flight drops for "
+ + "timeouts.");
+
public static final ConfigOption<Boolean> LOG_TABLE_ALLOW_CREATION =
key("allow.create.log.tables")
.booleanType()
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 ae3ceef33..33f9513a2 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
@@ -76,6 +76,10 @@ import static
org.apache.fluss.utils.concurrent.LockUtils.inLock;
* An auto partition manager which will trigger auto partition for the tables
in cluster
* periodically. It'll use a {@link ScheduledExecutorService} to schedule the
auto partition which
* will trigger auto partition for them.
+ *
+ * <p>TODO: migrate the jittered partition-creation throttling logic into
{@link
+ * TableLifecycleThrottler} so that partition creation and deletion share the
same lifecycle gate.
+ * Tracked by https://github.com/apache/fluss/issues/3457.
*/
public class AutoPartitionManager implements AutoCloseable {
@@ -344,7 +348,7 @@ public class AutoPartitionManager implements AutoCloseable {
dropPartitions(
tablePath,
tableInfo.getPartitionKeys(),
- createPartitionInstant,
+ now,
tableInfo.getTableConfig().getAutoPartitionStrategy(),
currentPartitions);
createPartitions(tableInfo, createPartitionInstant,
currentPartitions);
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
index 13eaba282..6c68a5583 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
@@ -87,6 +87,7 @@ import
org.apache.fluss.server.coordinator.event.NotifyLeaderAndIsrResponseRecei
import org.apache.fluss.server.coordinator.event.RebalanceEvent;
import org.apache.fluss.server.coordinator.event.RebalanceTaskTimeoutEvent;
import org.apache.fluss.server.coordinator.event.RemoveServerTagEvent;
+import org.apache.fluss.server.coordinator.event.ResumeDropEvent;
import org.apache.fluss.server.coordinator.event.SchemaChangeEvent;
import org.apache.fluss.server.coordinator.event.TableRegistrationChangeEvent;
import
org.apache.fluss.server.coordinator.event.watcher.CoordinatorChangeWatcher;
@@ -123,6 +124,7 @@ import org.apache.fluss.server.zk.data.ZkData.TableIdsZNode;
import org.apache.fluss.server.zk.data.lake.LakeTableHelper;
import org.apache.fluss.server.zk.data.lake.LakeTableSnapshot;
import org.apache.fluss.utils.AutoPartitionStrategy;
+import org.apache.fluss.utils.clock.Clock;
import org.apache.fluss.utils.clock.SystemClock;
import org.apache.fluss.utils.types.Tuple2;
@@ -174,6 +176,7 @@ public class CoordinatorEventProcessor implements
EventProcessor {
private final CoordinatorEventManager coordinatorEventManager;
private final MetadataManager metadataManager;
private final TableManager tableManager;
+ private final TableLifecycleThrottler lifecycleThrottler;
private final AutoPartitionManager autoPartitionManager;
private final LakeTableTieringManager lakeTableTieringManager;
private final TableChangeWatcher tableChangeWatcher;
@@ -199,7 +202,8 @@ public class CoordinatorEventProcessor implements
EventProcessor {
Configuration conf,
ExecutorService ioExecutor,
MetadataManager metadataManager,
- KvSnapshotLeaseManager kvSnapshotLeaseManager) {
+ KvSnapshotLeaseManager kvSnapshotLeaseManager,
+ Clock clock) {
this.zooKeeperClient = zooKeeperClient;
this.serverMetadataCache = serverMetadataCache;
this.coordinatorChannelManager = coordinatorChannelManager;
@@ -223,6 +227,7 @@ public class CoordinatorEventProcessor implements
EventProcessor {
zooKeeperClient);
this.metadataManager = metadataManager;
+ this.lifecycleThrottler = new
TableLifecycleThrottler(coordinatorEventManager, clock, conf);
this.tableManager =
new TableManager(
metadataManager,
@@ -230,10 +235,13 @@ public class CoordinatorEventProcessor implements
EventProcessor {
replicaStateMachine,
tableBucketStateMachine,
new RemoteStorageCleaner(conf, ioExecutor),
- ioExecutor);
+ ioExecutor,
+ lifecycleThrottler);
this.coordinatorChangeWatcher =
new CoordinatorChangeWatcher(zooKeeperClient,
coordinatorEventManager);
- this.tableChangeWatcher = new TableChangeWatcher(zooKeeperClient,
coordinatorEventManager);
+ this.tableChangeWatcher =
+ new TableChangeWatcher(
+ zooKeeperClient, coordinatorEventManager,
lifecycleThrottler);
this.tabletServerChangeWatcher =
new TabletServerChangeWatcher(zooKeeperClient,
coordinatorEventManager);
this.coordinatorRequestBatch =
@@ -271,6 +279,11 @@ public class CoordinatorEventProcessor implements
EventProcessor {
return coordinatorContext;
}
+ @VisibleForTesting
+ TableLifecycleThrottler getLifecycleThrottler() {
+ return lifecycleThrottler;
+ }
+
public void startup() {
coordinatorContext.setCoordinatorServerInfo(getCoordinatorServerInfo());
// start watchers first so that we won't miss node in zk;
@@ -298,6 +311,8 @@ public class CoordinatorEventProcessor implements
EventProcessor {
coordinatorContext.getServerTags());
updateTabletServerMetadataCacheWhenStartup(tabletServerInfoList);
+ lifecycleThrottler.start();
+
// start table manager
tableManager.startup();
@@ -563,6 +578,9 @@ public class CoordinatorEventProcessor implements
EventProcessor {
// first shutdown table manager
tableManager.shutdown();
+ // shut down lifecycle throttler timeout checker
+ lifecycleThrottler.close();
+
// then stop watchers
coordinatorChangeWatcher.stop();
tableChangeWatcher.stop();
@@ -657,6 +675,19 @@ public class CoordinatorEventProcessor implements
EventProcessor {
timeoutEvent.getTableBucket());
rebalanceManager.finishRebalanceTask(
timeoutEvent.getTableBucket(), RebalanceStatus.TIMEOUT);
+ } else if (event instanceof ResumeDropEvent) {
+ // Resume-mode reconciliation queued by TableLifecycleThrottler:
dispatch on the event
+ // thread so the @NotThreadSafe CoordinatorContext / state
machines are only mutated
+ // here. The event carries identity-only (kind + ids); the handler
chosen below is
+ // determined by the event-thread code path, not by the
(potentially non-event-thread)
+ // submitter, so callers cannot smuggle in unsafe lambdas.
+ ResumeDropEvent resumeDropEvent = (ResumeDropEvent) event;
+ if (resumeDropEvent.getPartitionId() == null) {
+ tableManager.onDeleteTable(resumeDropEvent.getTableId());
+ } else {
+ tableManager.onDeletePartition(
+ resumeDropEvent.getTableId(),
resumeDropEvent.getPartitionId());
+ }
} else if (event instanceof ListRebalanceProgressEvent) {
ListRebalanceProgressEvent listRebalanceProgressEvent =
(ListRebalanceProgressEvent) event;
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 6914e8207..2c4fcf5e4 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
@@ -336,7 +336,8 @@ public class CoordinatorServer extends ServerBase {
conf,
ioExecutor,
metadataManager,
- kvSnapshotLeaseManager);
+ kvSnapshotLeaseManager,
+ clock);
coordinatorEventProcessor.startup();
createDefaultDatabase();
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/TableLifecycleThrottler.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/TableLifecycleThrottler.java
new file mode 100644
index 000000000..d7b6a84dd
--- /dev/null
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/TableLifecycleThrottler.java
@@ -0,0 +1,553 @@
+/*
+ * 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.server.coordinator.event.ResumeDropEvent;
+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 throttler 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 throttler 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 throttler 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 throttler via {@link
+ * #submitPartitionDropForResume} / {@link #submitTableDropForResume}: stale
tables/partitions
+ * detected by {@code initCoordinatorContext} (whose {@code TableInfo} is
already gone, so a regular
+ * {@code DropTableEvent} is not viable). At admission time the throttler
enqueues a {@link
+ * ResumeDropEvent} carrying only identity (kind + ids); the coordinator
event-thread handler routes
+ * it to {@code TableManager#onDeleteTable} / {@code
TableManager#onDeletePartition}. This API
+ * deliberately accepts no {@link Runnable} from the submitter so the
(potentially non-event-thread)
+ * caller cannot smuggle in logic that would mutate the {@code @NotThreadSafe}
{@code
+ * CoordinatorContext} off the event thread.
+ *
+ * <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
configured timeout and abandons
+ * the drop with a WARN log. Abandonment only releases the throttler's
in-memory tracking; any
+ * residual replica state machine entries are reconciled on the next
coordinator startup via {@link
+ * TableManager#resumeDeletions()}.
+ */
+public class TableLifecycleThrottler implements AutoCloseable {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(TableLifecycleThrottler.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 TableLifecycleThrottler(EventManager eventManager, Clock clock,
Configuration conf) {
+ this(
+ eventManager,
+ clock,
+ Executors.newScheduledThreadPool(
+ 1, new
ExecutorThreadFactory("lifecycle-throttler-timeout")),
+
conf.get(ConfigOptions.COORDINATOR_LIFECYCLE_THROTTLER_INFLIGHT_TIMEOUT).toMillis(),
+
conf.get(ConfigOptions.COORDINATOR_LIFECYCLE_THROTTLER_TIMEOUT_CHECK_INTERVAL)
+ .toMillis());
+ }
+
+ @VisibleForTesting
+ TableLifecycleThrottler(
+ 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. Idempotent. */
+ public void start() {
+ if (closed.get()) {
+ throw new IllegalStateException("TableLifecycleThrottler is
already closed.");
+ }
+ if (started.compareAndSet(false, true)) {
+ timeoutChecker.scheduleWithFixedDelay(
+ this::checkTimeoutsSafely,
+ timeoutCheckIntervalMs,
+ timeoutCheckIntervalMs,
+ TimeUnit.MILLISECONDS);
+ LOG.info(
+ "TableLifecycleThrottler timeout checker started: "
+ + "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 throttler 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,
false));
+ }
+
+ /**
+ * Submits a partition drop request that, when admitted, enqueues a {@link
ResumeDropEvent} (so
+ * the event-thread handler in {@code CoordinatorEventProcessor} can call
{@code
+ * TableManager#onDeletePartition}) instead of a {@link
DropPartitionEvent}. Used by coordinator
+ * startup to reconcile stale partitions whose {@code TableInfo} is
already gone.
+ */
+ public void submitPartitionDropForResume(long tableId, long partitionId,
String partitionName) {
+ admit(new PendingPartitionDrop(tableId, partitionId, partitionName,
true));
+ }
+
+ /**
+ * 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,
+ false));
+ }
+
+ /**
+ * Submits a table drop request that, when admitted, enqueues a {@link
ResumeDropEvent} (so the
+ * event-thread handler in {@code CoordinatorEventProcessor} can call
{@code
+ * TableManager#onDeleteTable}) instead of a {@link DropTableEvent}. Used
by coordinator startup
+ * to reconcile stale tables whose {@code TableInfo} is already gone.
+ */
+ public void submitTableDropForResume(long tableId) {
+ admit(new PendingTableDrop(tableId, false, false, false, true));
+ }
+
+ /**
+ * 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 =
+ inLock(
+ lock,
+ () -> {
+ if (currentInflight == null) {
+ return null;
+ }
+ PendingDrop inflight = currentInflight.drop;
+ if (!(inflight instanceof PendingPartitionDrop)) {
+ return null;
+ }
+ PendingPartitionDrop p = (PendingPartitionDrop)
inflight;
+ if (p.tableId != tablePartition.getTableId()
+ || p.partitionId !=
tablePartition.getPartitionId()) {
+ return null;
+ }
+ LOG.debug(
+ "Partition drop completed: {} after {}ms;
pending={}",
+ tablePartition,
+ clock.milliseconds() -
currentInflight.submittedAtMs,
+ pendingDrops.size());
+ currentInflight = null;
+ return admitNext();
+ });
+ executeUntilBlocked(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 =
+ inLock(
+ lock,
+ () -> {
+ if (currentInflight == null) {
+ return null;
+ }
+ PendingDrop inflight = currentInflight.drop;
+ if (!(inflight instanceof PendingTableDrop)) {
+ return null;
+ }
+ PendingTableDrop t = (PendingTableDrop) inflight;
+ if (t.tableId != tableId) {
+ return null;
+ }
+ LOG.debug(
+ "Table drop completed: tableId={} after
{}ms; pending={}",
+ tableId,
+ clock.milliseconds() -
currentInflight.submittedAtMs,
+ pendingDrops.size());
+ currentInflight = null;
+ return admitNext();
+ });
+ executeUntilBlocked(next);
+ }
+
+ private void admit(PendingDrop drop) {
+ if (closed.get()) {
+ LOG.debug(
+ "Ignoring drop submission ({}) because
TableLifecycleThrottler is closed.",
+ drop);
+ return;
+ }
+ PendingDrop toExecute =
+ inLock(
+ lock,
+ () -> {
+ // 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 null;
+ }
+ pendingDrops.add(drop);
+ LOG.debug(
+ "Submitted drop: {}; pending={}
inflight={}",
+ drop,
+ pendingDrops.size(),
+ currentInflight != null);
+ return admitNext();
+ });
+ executeUntilBlocked(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 admitted drops until one needs an asynchronous completion
callback. */
+ private void executeUntilBlocked(@Nullable PendingDrop drop) {
+ PendingDrop current = drop;
+ while (current != null) {
+ if (!executeDrop(current)) {
+ return;
+ }
+ current = releaseInflightAndAdmitNext();
+ }
+ }
+
+ private boolean executeDrop(PendingDrop drop) {
+ drop.putEventInto(eventManager);
+ return drop.isFireAndForget();
+ }
+
+ private PendingDrop releaseInflightAndAdmitNext() {
+ return inLock(
+ lock,
+ () -> {
+ currentInflight = null;
+ return admitNext();
+ });
+ }
+
+ private void checkTimeoutsSafely() {
+ try {
+ checkTimeouts();
+ } catch (Throwable t) {
+ LOG.error("Unexpected error in TableLifecycleThrottler timeout
check.", t);
+ }
+ }
+
+ @VisibleForTesting
+ void checkTimeouts() {
+ PendingDrop next =
+ inLock(
+ lock,
+ () -> {
+ if (currentInflight == null) {
+ return null;
+ }
+ long elapsed = clock.milliseconds() -
currentInflight.submittedAtMs;
+ if (elapsed <= inflightTimeoutMs) {
+ return null;
+ }
+ 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;
+ return admitNext();
+ });
+ executeUntilBlocked(next);
+ }
+
+ @VisibleForTesting
+ int getInflightCount() {
+ return inLock(lock, () -> currentInflight != null ? 1 : 0);
+ }
+
+ @VisibleForTesting
+ int getPendingDropCount() {
+ return inLock(lock, pendingDrops::size);
+ }
+
+ @Override
+ public void close() {
+ if (closed.compareAndSet(false, true)) {
+ timeoutChecker.shutdownNow();
+ }
+ }
+
+ //
------------------------------------------------------------------------------------------
+ // PendingDrop hierarchy
+ //
------------------------------------------------------------------------------------------
+
+ /** A unit of work queued for admission into the coordinator event queue.
*/
+ interface PendingDrop {
+ void putEventInto(EventManager eventManager);
+
+ /**
+ * Whether this drop completes immediately without waiting for a
completion callback.
+ * Auto-partition table drops return {@code true} because partitioned
tables have no
+ * table-level replicas, so {@link #onTableDropCompleted} will never
be called.
+ */
+ default boolean isFireAndForget() {
+ return false;
+ }
+
+ /**
+ * Returns {@code true} if this drop targets the same table or
partition as {@code other}.
+ * Used for deduplication: if a drop with the same target is already
in-flight or pending, a
+ * duplicate submission is skipped.
+ */
+ boolean hasSameTarget(PendingDrop other);
+ }
+
+ /** Pending admission of a {@link DropPartitionEvent} for a single
partition. */
+ static final class PendingPartitionDrop implements PendingDrop {
+ final long tableId;
+ final long partitionId;
+ final String partitionName;
+ final boolean forResume;
+
+ PendingPartitionDrop(
+ long tableId, long partitionId, String partitionName, boolean
forResume) {
+ this.tableId = tableId;
+ this.partitionId = partitionId;
+ this.partitionName = partitionName;
+ this.forResume = forResume;
+ }
+
+ @Override
+ public void putEventInto(EventManager eventManager) {
+ if (forResume) {
+ // Resume-mode reconciliation: dispatch identity-only event to
the coordinator
+ // event thread, which mutates the @NotThreadSafe
CoordinatorContext. The
+ // admission point can be a watcher thread or the
timeout-checker thread, so
+ // the handler must NOT run inline here.
+ eventManager.put(ResumeDropEvent.forPartition(tableId,
partitionId));
+ } else {
+ eventManager.put(new DropPartitionEvent(tableId, partitionId,
partitionName));
+ }
+ }
+
+ @Override
+ public boolean hasSameTarget(PendingDrop other) {
+ if (!(other instanceof PendingPartitionDrop)) {
+ return false;
+ }
+ PendingPartitionDrop that = (PendingPartitionDrop) other;
+ return this.tableId == that.tableId && this.partitionId ==
that.partitionId;
+ }
+
+ @Override
+ public String toString() {
+ return "PendingPartitionDrop{tableId="
+ + tableId
+ + ", partitionId="
+ + partitionId
+ + ", partitionName='"
+ + partitionName
+ + "', forResume="
+ + forResume
+ + "}";
+ }
+ }
+
+ /** Pending admission of a {@link DropTableEvent} for a whole table. */
+ static final class PendingTableDrop implements PendingDrop {
+ final long tableId;
+ final boolean isPartitionedTable;
+ final boolean isAutoPartitionTable;
+ final boolean isDataLakeEnabled;
+ final boolean forResume;
+
+ PendingTableDrop(
+ long tableId,
+ boolean isPartitionedTable,
+ boolean isAutoPartitionTable,
+ boolean isDataLakeEnabled,
+ boolean forResume) {
+ this.tableId = tableId;
+ this.isPartitionedTable = isPartitionedTable;
+ this.isAutoPartitionTable = isAutoPartitionTable;
+ this.isDataLakeEnabled = isDataLakeEnabled;
+ this.forResume = forResume;
+ }
+
+ @Override
+ public boolean isFireAndForget() {
+ // Partitioned tables have no table-level replicas, so the
completion callback
+ // (onTableDropCompleted) will never arrive. Skip inflight
tracking.
+ // Resume-mode drops are excluded because the event-thread handler
triggers
+ // onDeleteTable() which may send RPCs for non-vacuously-complete
cases.
+ return isPartitionedTable && !forResume;
+ }
+
+ @Override
+ public void putEventInto(EventManager eventManager) {
+ if (forResume) {
+ // See PendingPartitionDrop#execute.
+ eventManager.put(ResumeDropEvent.forTable(tableId));
+ } else {
+ eventManager.put(
+ new DropTableEvent(tableId, isAutoPartitionTable,
isDataLakeEnabled));
+ }
+ }
+
+ @Override
+ public boolean hasSameTarget(PendingDrop other) {
+ if (!(other instanceof PendingTableDrop)) {
+ return false;
+ }
+ PendingTableDrop that = (PendingTableDrop) other;
+ return this.tableId == that.tableId;
+ }
+
+ @Override
+ public String toString() {
+ return "PendingTableDrop{tableId="
+ + tableId
+ + ", partitioned="
+ + isPartitionedTable
+ + ", autoPartition="
+ + isAutoPartitionTable
+ + ", dataLake="
+ + isDataLakeEnabled
+ + ", forResume="
+ + forResume
+ + "}";
+ }
+ }
+
+ /** Tracks the in-flight drop along with its admission timestamp for
timeout detection. */
+ private static final class InflightDrop {
+ final PendingDrop drop;
+ final long submittedAtMs;
+
+ InflightDrop(PendingDrop drop, long submittedAtMs) {
+ this.drop = drop;
+ this.submittedAtMs = submittedAtMs;
+ }
+ }
+}
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/TableManager.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/TableManager.java
index b92a2dad6..978751ec4 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/TableManager.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/TableManager.java
@@ -51,19 +51,33 @@ public class TableManager {
private final TableBucketStateMachine tableBucketStateMachine;
private final ExecutorService ioExecutor;
+ /**
+ * Throttler used to pace the asynchronous replica cleanup that follows
{@link
+ * #onDeleteTable(long)} / {@link #onDeletePartition(long, long)}. {@link
#resumeDeletions()}
+ * routes eligible drops through it (one at a time) instead of invoking
the state-machine
+ * transitions directly, and {@link #completeDeleteTable(long)} / {@link
+ * #completeDeletePartition(TablePartition)} notify it so the next pending
drop can be admitted.
+ *
+ * <p>Wired in via the constructor so that the reference publication is
safe (final field) and
+ * cannot race with the coordinator event thread.
+ */
+ private final TableLifecycleThrottler lifecycleThrottler;
+
public TableManager(
MetadataManager metadataManager,
CoordinatorContext coordinatorContext,
ReplicaStateMachine replicaStateMachine,
TableBucketStateMachine tableBucketStateMachine,
RemoteStorageCleaner remoteStorageCleaner,
- ExecutorService ioExecutor) {
+ ExecutorService ioExecutor,
+ TableLifecycleThrottler lifecycleThrottler) {
this.metadataManager = metadataManager;
this.remoteStorageCleaner = remoteStorageCleaner;
this.coordinatorContext = coordinatorContext;
this.replicaStateMachine = replicaStateMachine;
this.tableBucketStateMachine = tableBucketStateMachine;
this.ioExecutor = ioExecutor;
+ this.lifecycleThrottler = lifecycleThrottler;
}
public void startup() {
@@ -208,22 +222,13 @@ public class TableManager {
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)) {
+ lifecycleThrottler.submitTableDropForResume(tableId);
}
}
}
@@ -231,22 +236,18 @@ public class TableManager {
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();
+ String partitionName =
coordinatorContext.getPartitionName(partitionId);
+ lifecycleThrottler.submitPartitionDropForResume(
+ tableId, partitionId, partitionName);
}
}
}
@@ -257,6 +258,8 @@ public class TableManager {
asyncDeleteRemoteDirectory(tableId);
asyncDeleteTableMetadata(tableId);
coordinatorContext.removeTable(tableId);
+ // Release the manager's in-flight tracking so the next batch can be
submitted.
+ lifecycleThrottler.onTableDropCompleted(tableId);
}
private void completeDeletePartition(TablePartition tablePartition) {
@@ -267,6 +270,7 @@ public class TableManager {
asyncDeleteRemoteDirectory(tablePartition);
asyncDeletePartitionMetadata(tablePartition.getPartitionId());
coordinatorContext.removePartition(tablePartition);
+ lifecycleThrottler.onPartitionDropCompleted(tablePartition);
}
private void asyncDeleteRemoteDirectory(long tableId) {
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/ResumeDropEvent.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/ResumeDropEvent.java
new file mode 100644
index 000000000..056cfc154
--- /dev/null
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/ResumeDropEvent.java
@@ -0,0 +1,73 @@
+/*
+ * 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.event;
+
+import javax.annotation.Nullable;
+
+/**
+ * Marker event emitted by {@code TableLifecycleThrottler} to drive
coordinator-startup
+ * reconciliation of stale tables/partitions whose {@code TableInfo} is
already gone (i.e. a regular
+ * {@code DropTableEvent} / {@code DropPartitionEvent} would NPE on the
missing {@code TableInfo}).
+ * It carries only identity (tableId + optional partitionId), never a {@link
Runnable}, so the
+ * dispatch target is determined by the event-thread handler and cannot be
influenced by the
+ * (potentially non-event-thread) submitter.
+ *
+ * <p>A {@code null} {@link #getPartitionId()} indicates a whole-table resume;
otherwise the resume
+ * targets the specific partition.
+ *
+ * <p>The event handler in {@code CoordinatorEventProcessor} routes this to
{@code
+ * TableManager#onDeleteTable} / {@code TableManager#onDeletePartition},
ensuring the
+ * {@code @NotThreadSafe} {@code CoordinatorContext} is only ever mutated from
the coordinator event
+ * thread.
+ */
+public class ResumeDropEvent implements CoordinatorEvent {
+
+ private final long tableId;
+ @Nullable private final Long partitionId;
+
+ private ResumeDropEvent(long tableId, @Nullable Long partitionId) {
+ this.tableId = tableId;
+ this.partitionId = partitionId;
+ }
+
+ public static ResumeDropEvent forTable(long tableId) {
+ return new ResumeDropEvent(tableId, null);
+ }
+
+ public static ResumeDropEvent forPartition(long tableId, long partitionId)
{
+ return new ResumeDropEvent(tableId, partitionId);
+ }
+
+ public long getTableId() {
+ return tableId;
+ }
+
+ /** Returns {@code null} when this resume targets the whole table. */
+ @Nullable
+ public Long getPartitionId() {
+ return partitionId;
+ }
+
+ @Override
+ public String toString() {
+ if (partitionId == null) {
+ return "ResumeDropEvent{tableId=" + tableId + "}";
+ }
+ return "ResumeDropEvent{tableId=" + tableId + ", partitionId=" +
partitionId + "}";
+ }
+}
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/watcher/TableChangeWatcher.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/watcher/TableChangeWatcher.java
index 862092dfd..5e05877b4 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/watcher/TableChangeWatcher.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/watcher/TableChangeWatcher.java
@@ -23,10 +23,9 @@ import org.apache.fluss.metadata.PhysicalTablePath;
import org.apache.fluss.metadata.SchemaInfo;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.server.coordinator.TableLifecycleThrottler;
import org.apache.fluss.server.coordinator.event.CreatePartitionEvent;
import org.apache.fluss.server.coordinator.event.CreateTableEvent;
-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.server.coordinator.event.SchemaChangeEvent;
import org.apache.fluss.server.coordinator.event.TableRegistrationChangeEvent;
@@ -58,13 +57,18 @@ public class TableChangeWatcher {
private volatile boolean running;
private final EventManager eventManager;
+ private final TableLifecycleThrottler lifecycleThrottler;
private final ZooKeeperClient zooKeeperClient;
- public TableChangeWatcher(ZooKeeperClient zooKeeperClient, EventManager
eventManager) {
+ public TableChangeWatcher(
+ ZooKeeperClient zooKeeperClient,
+ EventManager eventManager,
+ TableLifecycleThrottler lifecycleThrottler) {
this.zooKeeperClient = zooKeeperClient;
this.curatorCache =
CuratorCache.build(zooKeeperClient.getCuratorClient(),
DatabasesZNode.path());
this.eventManager = eventManager;
+ this.lifecycleThrottler = lifecycleThrottler;
this.curatorCache.listenable().addListener(new
TablePathChangeListener());
}
@@ -166,14 +170,17 @@ public class TableChangeWatcher {
PhysicalTablePath physicalTablePath =
PartitionZNode.parsePath(oldData.getPath());
if (physicalTablePath != null) {
- // it's for deletion of a table partition node
+ // it's for deletion of a table partition node.
Submit to the
+ // replica cleanup manager rather than putting a
DropPartitionEvent
+ // directly: the manager throttles admission into
the coordinator
+ // event queue so that a cascading drop (10k+
partitions) cannot
+ // starve unrelated coordinator work.
PartitionRegistration partition =
PartitionZNode.decode(oldData.getData());
- eventManager.put(
- new DropPartitionEvent(
- partition.getTableId(),
- partition.getPartitionId(),
-
physicalTablePath.getPartitionName()));
+ lifecycleThrottler.submitPartitionDrop(
+ partition.getTableId(),
+ partition.getPartitionId(),
+ physicalTablePath.getPartitionName());
} else {
// maybe table node is deleted
// try to parse the path as a table node
@@ -184,13 +191,11 @@ public class TableChangeWatcher {
TableRegistration table =
TableZNode.decode(oldData.getData());
TableConfig tableConfig =
new
TableConfig(Configuration.fromMap(table.properties));
- eventManager.put(
- new DropTableEvent(
- table.tableId,
- tableConfig
- .getAutoPartitionStrategy()
- .isAutoPartitionEnabled(),
- tableConfig.isDataLakeEnabled()));
+ lifecycleThrottler.submitTableDrop(
+ table.tableId,
+ table.isPartitioned(),
+
tableConfig.getAutoPartitionStrategy().isAutoPartitionEnabled(),
+ tableConfig.isDataLakeEnabled());
}
break;
}
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/AutoPartitionManagerTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/AutoPartitionManagerTest.java
index 9b7b33fed..175634129 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/AutoPartitionManagerTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/AutoPartitionManagerTest.java
@@ -531,6 +531,56 @@ class AutoPartitionManagerTest {
"20250419", "20250420", "20250421", "20250422",
"20250423");
}
+ @Test
+ void testDayPartitionDropShouldNotBeDelayedByJitter() throws Exception {
+ ZonedDateTime startTime =
+
LocalDateTime.parse("2025-04-19T00:00:00").atZone(ZoneId.systemDefault());
+ long startMs = startTime.toInstant().toEpochMilli();
+ ManualClock clock = new ManualClock(startMs);
+ ManuallyTriggeredScheduledExecutorService periodicExecutor =
+ new ManuallyTriggeredScheduledExecutorService();
+ AutoPartitionManager autoPartitionManager =
+ new AutoPartitionManager(
+ new TestingServerMetadataCache(3),
+ metadataManager,
+ remoteDirDynamicLoader,
+ new Configuration(),
+ clock,
+ periodicExecutor);
+ autoPartitionManager.start();
+
+ // Create a DAY-partitioned table: numRetention=2, numPreCreate=4
+ TableInfo table = createPartitionedTable(2, 4,
AutoPartitionTimeUnit.DAY);
+ TablePath tablePath = table.getTablePath();
+ autoPartitionManager.addAutoPartitionTable(table, true);
+ periodicExecutor.triggerNonPeriodicScheduledTasks();
+
+ Map<String, PartitionRegistration> partitions =
+ zookeeperClient.getPartitionRegistrations(tablePath);
+ // Initial: 20250419, 20250420, 20250421, 20250422
+ assertThat(partitions.keySet())
+ .containsExactlyInAnyOrder("20250419", "20250420", "20250421",
"20250422");
+
+ Integer delayMinutes =
+
autoPartitionManager.getAutoCreateDayDelayMinutes(table.getTableId());
+ assertThat(delayMinutes).isNotNull();
+
+ // Advance exactly 3 days to 2025-04-22T00:00:00.
+ // From 'now' perspective: current day is 20250422, retain 2 => keep
20250420, 20250421.
+ // So 20250419 should be dropped.
+ //
+ // With the bug (drop used delayed time): if delay > 0, delayed time
would still
+ // be on 20250421, retain 2 => keep 20250419, 20250420 => 20250419 NOT
dropped.
+ clock.advanceTime(Duration.ofDays(3));
+ periodicExecutor.triggerPeriodicScheduledTasks();
+
+ partitions = zookeeperClient.getPartitionRegistrations(tablePath);
+ // 20250419 must be dropped regardless of jitter delay
+ assertThat(partitions.keySet()).doesNotContain("20250419");
+ // Retained partitions should still exist
+ assertThat(partitions.keySet()).contains("20250420", "20250421",
"20250422");
+ }
+
/**
* Test if AutoPartionManager.createPartition adheres to maxBucketLimit
while adding new
* parition automatically, skip if it breaches limit.
@@ -696,6 +746,12 @@ class AutoPartitionManagerTest {
.containsExactlyInAnyOrder("2024091000", "2024091001",
"2024091002", "2024091003");
}
+ //
---------------------------------------------------------------------------------------
+ // Batch / inflight tests previously housed here have moved to
TableLifecycleThrottlerTest.
+ // The AutoPartitionManager now drops expired partitions synchronously and
the asynchronous
+ // replica cleanup throttling is handled by TableLifecycleThrottler.
+ //
---------------------------------------------------------------------------------------
+
private static class TestParams {
final AutoPartitionTimeUnit timeUnit;
final boolean multiplePartitionKeys;
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java
index d6813abb0..8f27cdd10 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java
@@ -759,6 +759,204 @@ class CoordinatorEventProcessorTest {
replicationFactor);
}
+ @Test
+ void testDropPartitionRoutedThroughCleanupManager() throws Exception {
+ TablePath tablePath = TablePath.of(defaultDatabase,
"test_drop_partition_via_cleanup");
+ initCoordinatorChannel();
+ TableDescriptor partitionedTableDescriptor = getPartitionedTable();
+ long tableId =
+ metadataManager.createTable(
+ tablePath, remoteDataDir, partitionedTableDescriptor,
null, false);
+
+ int nBuckets = 3;
+ int replicationFactor = 3;
+ Map<Integer, BucketAssignment> assignments =
+ generateAssignment(
+ nBuckets,
+ replicationFactor,
+ new TabletServerInfo[] {
+ new TabletServerInfo(0, "rack0"),
+ new TabletServerInfo(1, "rack1"),
+ new TabletServerInfo(2, "rack2")
+ })
+ .getBucketAssignments();
+ PartitionAssignment partitionAssignment = new
PartitionAssignment(tableId, assignments);
+ Tuple2<PartitionIdName, PartitionIdName> partitionIdAndNameTuple2 =
+ preparePartitionAssignment(tablePath, tableId,
partitionAssignment);
+
+ long partition1Id = partitionIdAndNameTuple2.f0.partitionId;
+ String partition1Name = partitionIdAndNameTuple2.f0.partitionName;
+ long partition2Id = partitionIdAndNameTuple2.f1.partitionId;
+
+ verifyPartitionCreated(
+ new TablePartition(tableId, partition1Id),
+ partitionAssignment,
+ nBuckets,
+ replicationFactor);
+ verifyPartitionCreated(
+ new TablePartition(tableId, partition2Id),
+ partitionAssignment,
+ nBuckets,
+ replicationFactor);
+
+ // Drop partition1 via ZK (simulates the watcher path).
+ zookeeperClient.deletePartition(tablePath, partition1Name);
+
+ // Verify the drop entered the lifecycle throttler and partition is
fully deleted.
+ TableLifecycleThrottler throttler =
eventProcessor.getLifecycleThrottler();
+ verifyPartitionDropped(tableId, partition1Id);
+
+ // After completion, the lifecycle throttler should have released
tracking.
+ assertThat(throttler.getInflightCount()).isZero();
+ assertThat(throttler.getPendingDropCount()).isZero();
+
+ // Partition2 should remain online.
+ verifyPartitionCreated(
+ new TablePartition(tableId, partition2Id),
+ partitionAssignment,
+ nBuckets,
+ replicationFactor);
+ }
+
+ @Test
+ void testDropPartitionedTableRoutedThroughCleanupManager() throws
Exception {
+ TablePath tablePath = TablePath.of(defaultDatabase,
"test_drop_table_via_cleanup");
+ initCoordinatorChannel();
+ TableDescriptor partitionedTableDescriptor = getPartitionedTable();
+ long tableId =
+ metadataManager.createTable(
+ tablePath, remoteDataDir, partitionedTableDescriptor,
null, false);
+
+ int nBuckets = 3;
+ int replicationFactor = 3;
+ Map<Integer, BucketAssignment> assignments =
+ generateAssignment(
+ nBuckets,
+ replicationFactor,
+ new TabletServerInfo[] {
+ new TabletServerInfo(0, "rack0"),
+ new TabletServerInfo(1, "rack1"),
+ new TabletServerInfo(2, "rack2")
+ })
+ .getBucketAssignments();
+ PartitionAssignment partitionAssignment = new
PartitionAssignment(tableId, assignments);
+ Tuple2<PartitionIdName, PartitionIdName> partitionIdAndNameTuple2 =
+ preparePartitionAssignment(tablePath, tableId,
partitionAssignment);
+
+ long partition1Id = partitionIdAndNameTuple2.f0.partitionId;
+ long partition2Id = partitionIdAndNameTuple2.f1.partitionId;
+
+ verifyPartitionCreated(
+ new TablePartition(tableId, partition1Id),
+ partitionAssignment,
+ nBuckets,
+ replicationFactor);
+ verifyPartitionCreated(
+ new TablePartition(tableId, partition2Id),
+ partitionAssignment,
+ nBuckets,
+ replicationFactor);
+
+ // Drop the entire table (triggers NODE_DELETED for partitions + table
via watcher).
+ metadataManager.dropTable(tablePath, false);
+
+ // Verify the drops entered the lifecycle throttler.
+ TableLifecycleThrottler cleanupManager =
eventProcessor.getLifecycleThrottler();
+ retry(
+ Duration.ofMinutes(1),
+ () ->
+ assertThat(
+ cleanupManager.getInflightCount() > 0
+ || (!zookeeperClient
+
.getPartitionAssignment(
+
partition1Id)
+ .isPresent()
+ && !zookeeperClient
+
.getPartitionAssignment(
+
partition2Id)
+ .isPresent()))
+ .isTrue());
+
+ // Verify both partitions and the table are fully deleted.
+ verifyPartitionDropped(tableId, partition1Id);
+ verifyPartitionDropped(tableId, partition2Id);
+ verifyTableDropped(tableId);
+
+ // Lifecycle throttler returns to idle state after all completions
propagate.
+ retry(
+ Duration.ofMinutes(1),
+ () -> {
+ assertThat(cleanupManager.getInflightCount()).isZero();
+ assertThat(cleanupManager.getPendingDropCount()).isZero();
+ });
+ }
+
+ @Test
+ void testStartupResumesDropPartitionThroughCleanupManager() throws
Exception {
+ TablePath tablePath = TablePath.of(defaultDatabase,
"test_startup_resume_via_cleanup");
+ initCoordinatorChannel();
+ TableDescriptor partitionedTableDescriptor = getPartitionedTable();
+ long tableId =
+ metadataManager.createTable(
+ tablePath, remoteDataDir, partitionedTableDescriptor,
null, false);
+
+ int nBuckets = 3;
+ int replicationFactor = 3;
+ Map<Integer, BucketAssignment> assignments =
+ generateAssignment(
+ nBuckets,
+ replicationFactor,
+ new TabletServerInfo[] {
+ new TabletServerInfo(0, "rack0"),
+ new TabletServerInfo(1, "rack1"),
+ new TabletServerInfo(2, "rack2")
+ })
+ .getBucketAssignments();
+ PartitionAssignment partitionAssignment = new
PartitionAssignment(tableId, assignments);
+ Tuple2<PartitionIdName, PartitionIdName> partitionIdAndNameTuple2 =
+ preparePartitionAssignment(tablePath, tableId,
partitionAssignment);
+
+ long partition1Id = partitionIdAndNameTuple2.f0.partitionId;
+ String partition2Name = partitionIdAndNameTuple2.f1.partitionName;
+ long partition2Id = partitionIdAndNameTuple2.f1.partitionId;
+
+ verifyPartitionCreated(
+ new TablePartition(tableId, partition1Id),
+ partitionAssignment,
+ nBuckets,
+ replicationFactor);
+ verifyPartitionCreated(
+ new TablePartition(tableId, partition2Id),
+ partitionAssignment,
+ nBuckets,
+ replicationFactor);
+
+ // Shutdown the event processor, then delete partition2 in ZK while
it's down.
+ eventProcessor.shutdown();
+ zookeeperClient.deletePartition(tablePath, partition2Name);
+
+ // Restart the event processor. During startup, the stale partition2
should be
+ // detected and routed through the cleanup manager.
+ eventProcessor = buildCoordinatorEventProcessor();
+ initCoordinatorChannel();
+ eventProcessor.startup();
+
+ // Verify partition2 is fully deleted (routed through lifecycle
throttler on startup).
+ TableLifecycleThrottler cleanupManager =
eventProcessor.getLifecycleThrottler();
+ verifyPartitionDropped(tableId, partition2Id);
+
+ // Partition1 should remain online.
+ verifyPartitionCreated(
+ new TablePartition(tableId, partition1Id),
+ partitionAssignment,
+ nBuckets,
+ replicationFactor);
+
+ // Lifecycle throttler returns to idle.
+ assertThat(cleanupManager.getInflightCount()).isZero();
+ assertThat(cleanupManager.getPendingDropCount()).isZero();
+ }
+
@Test
void testNotifyOffsetsWithShrinkISR(@TempDir Path tempDir) throws
Exception {
initCoordinatorChannel(Collections.singleton(ApiKeys.UPDATE_METADATA));
@@ -1392,7 +1590,8 @@ class CoordinatorEventProcessorTest {
conf,
Executors.newFixedThreadPool(1, new
ExecutorThreadFactory("test-coordinator-io")),
metadataManager,
- kvSnapshotLeaseManager);
+ kvSnapshotLeaseManager,
+ SystemClock.getInstance());
}
private void initCoordinatorChannel() throws Exception {
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableLifecycleThrottlerTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableLifecycleThrottlerTest.java
new file mode 100644
index 000000000..fa5dd9069
--- /dev/null
+++
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableLifecycleThrottlerTest.java
@@ -0,0 +1,430 @@
+/*
+ * 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.metadata.TablePartition;
+import org.apache.fluss.server.coordinator.event.CoordinatorEvent;
+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.server.coordinator.event.ResumeDropEvent;
+import org.apache.fluss.utils.clock.ManualClock;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Unit tests for {@link TableLifecycleThrottler}. */
+class TableLifecycleThrottlerTest {
+
+ private RecordingEventManager eventManager;
+ private ManualClock clock;
+ private NoOpScheduledExecutor timeoutExecutor;
+
+ @BeforeEach
+ void setup() {
+ eventManager = new RecordingEventManager();
+ clock = new ManualClock(0L);
+ timeoutExecutor = new NoOpScheduledExecutor();
+ }
+
+ @AfterEach
+ void tearDown() {
+ timeoutExecutor.shutdownNow();
+ }
+
+ @Test
+ void testSubmitPartitionDropAdmitsImmediately() {
+ TableLifecycleThrottler manager = newThrottler();
+
+ manager.submitPartitionDrop(1L, 100L, "p20240101");
+
+
assertThat(eventManager.summaries()).containsExactly("partition:1:100:p20240101");
+ assertThat(manager.getInflightCount()).isOne();
+ assertThat(manager.getPendingDropCount()).isZero();
+ }
+
+ @Test
+ void testSubmitTableDropAdmitsImmediately() {
+ TableLifecycleThrottler manager = newThrottler();
+
+ manager.submitTableDrop(7L, false, false, false);
+
+
assertThat(eventManager.summaries()).containsExactly("table:7:auto=false:lake=false");
+ assertThat(manager.getInflightCount()).isOne();
+ }
+
+ @Test
+ void testOneAtATimeThrottling() {
+ TableLifecycleThrottler manager = newThrottler();
+
+ // First drop is admitted immediately; subsequent drops pend.
+ manager.submitPartitionDrop(1L, 100L, "p1");
+ manager.submitPartitionDrop(1L, 101L, "p2");
+ manager.submitPartitionDrop(1L, 102L, "p3");
+
+
assertThat(eventManager.summaries()).containsExactly("partition:1:100:p1");
+ assertThat(manager.getInflightCount()).isOne();
+ assertThat(manager.getPendingDropCount()).isEqualTo(2);
+ }
+
+ @Test
+ void testCompletionAdmitsNextPending() {
+ TableLifecycleThrottler manager = newThrottler();
+
+ manager.submitPartitionDrop(1L, 100L, "p1");
+ manager.submitPartitionDrop(1L, 101L, "p2");
+ manager.submitPartitionDrop(1L, 102L, "p3");
+
+
assertThat(eventManager.summaries()).containsExactly("partition:1:100:p1");
+
+ // Complete the first; second is admitted.
+ manager.onPartitionDropCompleted(new TablePartition(1L, 100L));
+ assertThat(eventManager.summaries())
+ .containsExactly("partition:1:100:p1", "partition:1:101:p2");
+ assertThat(manager.getInflightCount()).isOne();
+ assertThat(manager.getPendingDropCount()).isOne();
+
+ // Complete the second; third is admitted.
+ manager.onPartitionDropCompleted(new TablePartition(1L, 101L));
+ assertThat(eventManager.summaries())
+ .containsExactly("partition:1:100:p1", "partition:1:101:p2",
"partition:1:102:p3");
+ assertThat(manager.getInflightCount()).isOne();
+ assertThat(manager.getPendingDropCount()).isZero();
+ }
+
+ @Test
+ void testTableDropCompletionAdmitsNext() {
+ TableLifecycleThrottler manager = newThrottler();
+
+ manager.submitTableDrop(1L, false, false, false);
+ manager.submitPartitionDrop(2L, 200L, "p2");
+
+
assertThat(eventManager.summaries()).containsExactly("table:1:auto=false:lake=false");
+ assertThat(manager.getPendingDropCount()).isOne();
+
+ manager.onTableDropCompleted(1L);
+
+ assertThat(eventManager.summaries())
+ .containsExactly("table:1:auto=false:lake=false",
"partition:2:200:p2");
+ assertThat(manager.getPendingDropCount()).isZero();
+ }
+
+ @Test
+ void testFifoOrder() {
+ TableLifecycleThrottler manager = newThrottler();
+
+ manager.submitPartitionDrop(1L, 10L, "p10");
+ manager.submitPartitionDrop(1L, 11L, "p11");
+ manager.submitPartitionDrop(1L, 12L, "p12");
+
+
assertThat(eventManager.summaries()).containsExactly("partition:1:10:p10");
+
+ manager.onPartitionDropCompleted(new TablePartition(1L, 10L));
+ manager.onPartitionDropCompleted(new TablePartition(1L, 11L));
+
+ assertThat(eventManager.summaries())
+ .containsExactly("partition:1:10:p10", "partition:1:11:p11",
"partition:1:12:p12");
+ }
+
+ @Test
+ void testCompletionForUnknownDropIsNoOp() {
+ TableLifecycleThrottler manager = newThrottler();
+
+ manager.onPartitionDropCompleted(new TablePartition(99L, 999L));
+ manager.onTableDropCompleted(123L);
+
+ assertThat(eventManager.summaries()).isEmpty();
+ assertThat(manager.getInflightCount()).isZero();
+ }
+
+ @Test
+ void testCompletionForMismatchedDropIsNoOp() {
+ TableLifecycleThrottler manager = newThrottler();
+
+ manager.submitPartitionDrop(1L, 100L, "p1");
+
+ // Complete with wrong partition id — should be no-op.
+ manager.onPartitionDropCompleted(new TablePartition(1L, 999L));
+ assertThat(manager.getInflightCount()).isOne();
+
+ // Complete with table callback — should be no-op (inflight is a
partition drop).
+ manager.onTableDropCompleted(1L);
+ assertThat(manager.getInflightCount()).isOne();
+ }
+
+ @Test
+ void testTimeoutAbandonsInflightDrop() {
+ TableLifecycleThrottler manager = newThrottler();
+
+ manager.submitTableDrop(1L, false, false, false);
+ manager.submitPartitionDrop(2L, 200L, "px");
+
+ assertThat(manager.getInflightCount()).isOne();
+ assertThat(manager.getPendingDropCount()).isOne();
+
+ // Not yet expired.
+ clock.advanceTime(Duration.ofMillis(100_000));
+ manager.checkTimeouts();
+ assertThat(manager.getInflightCount()).isOne();
+
+ // Crosses the 3-minute timeout boundary.
+ clock.advanceTime(Duration.ofMillis(90_000));
+ manager.checkTimeouts();
+
+ assertThat(eventManager.summaries())
+ .containsExactly("table:1:auto=false:lake=false",
"partition:2:200:px");
+ assertThat(manager.getPendingDropCount()).isZero();
+ assertThat(manager.getInflightCount()).isOne();
+ }
+
+ @Test
+ void testDuplicateCompletionIsNoOp() {
+ TableLifecycleThrottler manager = newThrottler();
+ TablePartition tp = new TablePartition(1L, 100L);
+
+ manager.submitPartitionDrop(tp.getTableId(), tp.getPartitionId(),
"px");
+ assertThat(manager.getInflightCount()).isOne();
+
+ manager.onPartitionDropCompleted(tp);
+ assertThat(manager.getInflightCount()).isZero();
+
+ // Second completion is a no-op.
+ manager.onPartitionDropCompleted(tp);
+ assertThat(manager.getInflightCount()).isZero();
+ }
+
+ @Test
+ void testStartIsIdempotentAndCloseDisablesStart() {
+ TableLifecycleThrottler manager = newThrottler();
+
+ manager.start();
+ manager.start(); // second start is a no-op
+
+ manager.close();
+
+
assertThatThrownBy(manager::start).isInstanceOf(IllegalStateException.class);
+ }
+
+ @Test
+ void testCloseIsIdempotent() {
+ TableLifecycleThrottler manager = newThrottler();
+
+ manager.close();
+ manager.close(); // must not throw
+ }
+
+ @Test
+ void testSubmitTableDropForResumeEnqueuesResumeEvent() {
+ TableLifecycleThrottler manager = newThrottler();
+
+ manager.submitTableDropForResume(7L);
+
+ // Resume-mode submission enqueues an identity-only ResumeDropEvent
for the event thread.
+ assertThat(eventManager.events).hasSize(1);
+
assertThat(eventManager.events.get(0)).isInstanceOf(ResumeDropEvent.class);
+ ResumeDropEvent resumeEvent = (ResumeDropEvent)
eventManager.events.get(0);
+ assertThat(resumeEvent.getPartitionId()).isNull();
+ assertThat(resumeEvent.getTableId()).isEqualTo(7L);
+ assertThat(manager.getInflightCount()).isOne();
+ }
+
+ @Test
+ void testSubmitPartitionDropForResumeEnqueuesResumeEvent() {
+ TableLifecycleThrottler manager = newThrottler();
+
+ manager.submitPartitionDropForResume(1L, 100L, "p20240101");
+
+ assertThat(eventManager.events).hasSize(1);
+
assertThat(eventManager.events.get(0)).isInstanceOf(ResumeDropEvent.class);
+ ResumeDropEvent resumeEvent = (ResumeDropEvent)
eventManager.events.get(0);
+ assertThat(resumeEvent.getTableId()).isEqualTo(1L);
+ assertThat(resumeEvent.getPartitionId()).isEqualTo(100L);
+ assertThat(manager.getInflightCount()).isOne();
+ }
+
+ @Test
+ void testForResumeDropRespectsOneAtATime() {
+ TableLifecycleThrottler manager = newThrottler();
+
+ manager.submitTableDropForResume(1L);
+ manager.submitPartitionDropForResume(2L, 200L, "px");
+
+ // First admitted (resume event enqueued); second pends.
+ assertThat(eventManager.events).hasSize(1);
+ ResumeDropEvent firstEvent = (ResumeDropEvent)
eventManager.events.get(0);
+ assertThat(firstEvent.getPartitionId()).isNull();
+ assertThat(firstEvent.getTableId()).isEqualTo(1L);
+ assertThat(manager.getInflightCount()).isOne();
+ assertThat(manager.getPendingDropCount()).isOne();
+
+ // Completion of the first admits the second.
+ manager.onTableDropCompleted(1L);
+ assertThat(eventManager.events).hasSize(2);
+ ResumeDropEvent secondEvent = (ResumeDropEvent)
eventManager.events.get(1);
+ assertThat(secondEvent.getTableId()).isEqualTo(2L);
+ assertThat(secondEvent.getPartitionId()).isEqualTo(200L);
+ }
+
+ @Test
+ void testPartitionedTableDropIsFireAndForget() {
+ TableLifecycleThrottler manager = newThrottler();
+
+ // Submit a partition drop first (will be admitted as inflight).
+ manager.submitPartitionDrop(1L, 100L, "p1");
+ // Then submit a partitioned table drop (pending).
+ manager.submitTableDrop(1L, true, true, false);
+ // Then submit another partition drop (pending).
+ manager.submitPartitionDrop(2L, 200L, "p2");
+
+
assertThat(eventManager.summaries()).containsExactly("partition:1:100:p1");
+ assertThat(manager.getInflightCount()).isOne();
+ assertThat(manager.getPendingDropCount()).isEqualTo(2);
+
+ // Complete partition1. The auto-partition table drop is admitted next.
+ // Since it's fire-and-forget, it immediately releases and admits the
next drop.
+ manager.onPartitionDropCompleted(new TablePartition(1L, 100L));
+
+ // Both the table drop AND the following partition drop should have
been admitted.
+ assertThat(eventManager.summaries())
+ .containsExactly(
+ "partition:1:100:p1", "table:1:auto=true:lake=false",
"partition:2:200:p2");
+ // The last partition drop is now inflight (waiting for completion).
+ assertThat(manager.getInflightCount()).isOne();
+ assertThat(manager.getPendingDropCount()).isZero();
+ }
+
+ @Test
+ void testPartitionedTableDropAdmittedFirstIsAlsoFireAndForget() {
+ TableLifecycleThrottler manager = newThrottler();
+
+ // Submit a partitioned table drop as the first drop.
+ manager.submitTableDrop(5L, true, true, false);
+
+ // It should be executed immediately AND released (not tracked as
inflight).
+
assertThat(eventManager.summaries()).containsExactly("table:5:auto=true:lake=false");
+ assertThat(manager.getInflightCount()).isZero();
+ assertThat(manager.getPendingDropCount()).isZero();
+ }
+
+ @Test
+ void testPartitionedTableWithoutAutoPartitionIsAlsoFireAndForget() {
+ TableLifecycleThrottler manager = newThrottler();
+
+ // A partitioned table without auto-partition still has no table-level
replicas.
+ manager.submitTableDrop(5L, true, false, false);
+
+ // It should also be fire-and-forget.
+
assertThat(eventManager.summaries()).containsExactly("table:5:auto=false:lake=false");
+ assertThat(manager.getInflightCount()).isZero();
+ assertThat(manager.getPendingDropCount()).isZero();
+ }
+
+ @Test
+ void testNonPartitionedTableDropIsNotFireAndForget() {
+ TableLifecycleThrottler manager = newThrottler();
+
+ // Non-partitioned table drop should be tracked normally.
+ manager.submitTableDrop(5L, false, false, false);
+
+
assertThat(eventManager.summaries()).containsExactly("table:5:auto=false:lake=false");
+ assertThat(manager.getInflightCount()).isOne();
+ assertThat(manager.getPendingDropCount()).isZero();
+ }
+
+ //
------------------------------------------------------------------------------------------
+ // Helpers
+ //
------------------------------------------------------------------------------------------
+
+ private TableLifecycleThrottler newThrottler() {
+ TableLifecycleThrottler throttler =
+ new TableLifecycleThrottler(
+ eventManager, clock, timeoutExecutor, 3 * 60 * 1000L,
60 * 1000L);
+ throttler.start(); // Activate throttling for unit tests.
+ return throttler;
+ }
+
+ /** Records the order of drop events admitted into the coordinator event
queue. */
+ private static final class RecordingEventManager implements EventManager {
+ final List<CoordinatorEvent> events = new ArrayList<>();
+
+ @Override
+ public void put(CoordinatorEvent event) {
+ events.add(event);
+ }
+
+ List<String> summaries() {
+ List<String> out = new ArrayList<>(events.size());
+ for (CoordinatorEvent event : events) {
+ if (event instanceof DropPartitionEvent) {
+ DropPartitionEvent e = (DropPartitionEvent) event;
+ out.add(
+ "partition:"
+ + e.getTableId()
+ + ":"
+ + e.getPartitionId()
+ + ":"
+ + e.getPartitionName());
+ } else if (event instanceof DropTableEvent) {
+ DropTableEvent e = (DropTableEvent) event;
+ out.add(
+ "table:"
+ + e.getTableId()
+ + ":auto="
+ + e.isAutoPartitionTable()
+ + ":lake="
+ + e.isDataLakeEnabled());
+ } else {
+ out.add(event.toString());
+ }
+ }
+ return out;
+ }
+ }
+
+ /**
+ * A scheduled executor that never actually runs scheduled tasks, so tests
retain full control
+ * over when {@link TableLifecycleThrottler#checkTimeouts()} is invoked.
+ */
+ private static final class NoOpScheduledExecutor extends
ScheduledThreadPoolExecutor
+ implements ScheduledExecutorService {
+
+ NoOpScheduledExecutor() {
+ super(0);
+ }
+
+ @Override
+ public java.util.concurrent.ScheduledFuture<?> scheduleWithFixedDelay(
+ Runnable command,
+ long initialDelay,
+ long delay,
+ java.util.concurrent.TimeUnit unit) {
+ // Intentionally drop the schedule request: tests drive
checkTimeouts() directly.
+ return null;
+ }
+ }
+}
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerTest.java
index f5357c549..68b65d80d 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerTest.java
@@ -25,6 +25,7 @@ import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePartition;
import org.apache.fluss.server.coordinator.event.CoordinatorEvent;
import
org.apache.fluss.server.coordinator.event.DeleteReplicaResponseReceivedEvent;
+import org.apache.fluss.server.coordinator.event.ResumeDropEvent;
import org.apache.fluss.server.coordinator.event.TestingEventManager;
import org.apache.fluss.server.coordinator.statemachine.ReplicaStateMachine;
import
org.apache.fluss.server.coordinator.statemachine.TableBucketStateMachine;
@@ -38,6 +39,7 @@ import org.apache.fluss.server.zk.data.BucketAssignment;
import org.apache.fluss.server.zk.data.PartitionAssignment;
import org.apache.fluss.server.zk.data.TableAssignment;
import org.apache.fluss.testutils.common.AllCallbackWrapper;
+import org.apache.fluss.utils.clock.SystemClock;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
@@ -84,6 +86,7 @@ class TableManagerTest {
private CoordinatorContext coordinatorContext;
private TableManager tableManager;
+ private TableLifecycleThrottler lifecycleThrottler;
private TestingEventManager testingEventManager;
private TestCoordinatorChannelManager testCoordinatorChannelManager;
@@ -107,6 +110,9 @@ class TableManagerTest {
if (tableManager != null) {
tableManager.shutdown();
}
+ if (lifecycleThrottler != null) {
+ lifecycleThrottler.close();
+ }
}
@AfterAll
@@ -134,6 +140,9 @@ class TableManagerTest {
zookeeperClient,
new Configuration(),
new LakeCatalogDynamicLoader(new Configuration(),
null, true));
+ lifecycleThrottler =
+ new TableLifecycleThrottler(
+ testingEventManager, SystemClock.getInstance(), new
Configuration());
tableManager =
new TableManager(
metadataManager,
@@ -141,7 +150,8 @@ class TableManagerTest {
replicaStateMachine,
tableBucketStateMachine,
new RemoteStorageCleaner(conf, ioExecutor),
- ioExecutor);
+ ioExecutor,
+ lifecycleThrottler);
tableManager.startup();
coordinatorContext.setLiveTabletServers(
@@ -261,6 +271,9 @@ class TableManagerTest {
// start table manager, should resume table deletion
tableManager.startup();
+ // TableLifecycleThrottler defers resume actions to the coordinator
event thread by
+ // enqueuing ResumeDropEvent; the unit test has no real event loop so
dispatch them inline.
+ dispatchResumeDropEvents();
checkReplicaDelete(tableId, null, assignment);
}
@@ -412,4 +425,23 @@ class TableManagerTest {
}
return deleteReplicaResponseReceivedEvent;
}
+
+ /**
+ * Drives every {@link ResumeDropEvent} currently in the testing event
manager. Mirrors the
+ * dispatch logic of {@code
CoordinatorEventProcessor#process(CoordinatorEvent)} so unit tests
+ * can trigger the deferred resume reconciliation without spinning up the
real event loop.
+ */
+ private void dispatchResumeDropEvents() {
+ for (CoordinatorEvent event : testingEventManager.getEvents()) {
+ if (event instanceof ResumeDropEvent) {
+ ResumeDropEvent resumeEvent = (ResumeDropEvent) event;
+ if (resumeEvent.getPartitionId() == null) {
+ tableManager.onDeleteTable(resumeEvent.getTableId());
+ } else {
+ tableManager.onDeletePartition(
+ resumeEvent.getTableId(),
resumeEvent.getPartitionId());
+ }
+ }
+ }
+ }
}
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/event/watcher/TableChangeWatcherTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/event/watcher/TableChangeWatcherTest.java
index 9baa3ec49..c181e4442 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/event/watcher/TableChangeWatcherTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/event/watcher/TableChangeWatcherTest.java
@@ -26,9 +26,11 @@ import org.apache.fluss.metadata.SchemaInfo;
import org.apache.fluss.metadata.TableChange;
import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.metadata.TablePartition;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.server.coordinator.LakeCatalogDynamicLoader;
import org.apache.fluss.server.coordinator.MetadataManager;
+import org.apache.fluss.server.coordinator.TableLifecycleThrottler;
import org.apache.fluss.server.coordinator.event.CoordinatorEvent;
import org.apache.fluss.server.coordinator.event.CreatePartitionEvent;
import org.apache.fluss.server.coordinator.event.CreateTableEvent;
@@ -46,6 +48,7 @@ import org.apache.fluss.server.zk.data.TableAssignment;
import org.apache.fluss.server.zk.data.TableRegistration;
import org.apache.fluss.testutils.common.AllCallbackWrapper;
import org.apache.fluss.types.DataTypes;
+import org.apache.fluss.utils.clock.SystemClock;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
@@ -81,6 +84,7 @@ class TableChangeWatcherTest {
private static ZooKeeperClient zookeeperClient;
private static String remoteDataDir;
private TestingEventManager eventManager;
+ private TableLifecycleThrottler lifecycleThrottler;
private TableChangeWatcher tableChangeWatcher;
private static MetadataManager metadataManager;
@@ -109,7 +113,11 @@ class TableChangeWatcherTest {
metadataManager.createDatabase(DEFAULT_DB,
DatabaseDescriptor.builder().build(), false);
eventManager = new TestingEventManager();
- tableChangeWatcher = new TableChangeWatcher(zookeeperClient,
eventManager);
+ lifecycleThrottler =
+ new TableLifecycleThrottler(
+ eventManager, SystemClock.getInstance(), new
Configuration());
+ tableChangeWatcher =
+ new TableChangeWatcher(zookeeperClient, eventManager,
lifecycleThrottler);
tableChangeWatcher.start();
}
@@ -118,6 +126,9 @@ class TableChangeWatcherTest {
if (tableChangeWatcher != null) {
tableChangeWatcher.stop();
}
+ if (lifecycleThrottler != null) {
+ lifecycleThrottler.close();
+ }
}
@Test
@@ -172,7 +183,14 @@ class TableChangeWatcherTest {
expectedTableEvents.add(new DropTableEvent(tableInfo.getTableId(),
false, false));
}
- // collect all events and check the all events
+ // The throttler admits one drop at a time; drive it by completing
each drop
+ // as it appears so the next pending drop is admitted.
+ for (CoordinatorEvent event : expectedTableEvents) {
+ DropTableEvent drop = (DropTableEvent) event;
+ lifecycleThrottler.onTableDropCompleted(drop.getTableId());
+ }
+
+ // collect all events and check the all events.
List<CoordinatorEvent> allEvents = new ArrayList<>(expectedEvents);
allEvents.addAll(expectedTableEvents);
retry(
@@ -257,11 +275,16 @@ class TableChangeWatcherTest {
// drop table event
expectedEvents.add(new DropTableEvent(tableId, true, false));
+ // The throttler admits one drop at a time; drive it by completing
partition
+ // drops as they arrive (the table drop is fire-and-forget for
partitioned tables).
retry(
Duration.ofMinutes(1),
- () ->
- assertThat(eventManager.getEvents())
-
.containsExactlyInAnyOrderElementsOf(expectedEvents));
+ () -> {
+ lifecycleThrottler.onPartitionDropCompleted(new
TablePartition(tableId, 1L));
+ lifecycleThrottler.onPartitionDropCompleted(new
TablePartition(tableId, 2L));
+ assertThat(eventManager.getEvents())
+
.containsExactlyInAnyOrderElementsOf(expectedEvents);
+ });
}
@Test
@@ -461,7 +484,11 @@ class TableChangeWatcherTest {
// existing nodes with full data - the same code path as when the async
// getData race causes NODE_CHANGED to be lost.
TestingEventManager newEventManager = new TestingEventManager();
- TableChangeWatcher newWatcher = new
TableChangeWatcher(zookeeperClient, newEventManager);
+ TableLifecycleThrottler newThrottler =
+ new TableLifecycleThrottler(
+ newEventManager, SystemClock.getInstance(), new
Configuration());
+ TableChangeWatcher newWatcher =
+ new TableChangeWatcher(zookeeperClient, newEventManager,
newThrottler);
newWatcher.start();
retry(
@@ -470,5 +497,6 @@ class TableChangeWatcherTest {
assertThat(newEventManager.getEvents())
.containsExactlyInAnyOrderElementsOf(expectedEvents));
newWatcher.stop();
+ newThrottler.close();
}
}
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java
index d550aea1c..0360b6a4e 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java
@@ -311,7 +311,8 @@ public class RebalanceManagerTest {
conf,
Executors.newFixedThreadPool(1, new
ExecutorThreadFactory("test-coordinator-io")),
metadataManager,
- kvSnapshotLeaseManager);
+ kvSnapshotLeaseManager,
+ SystemClock.getInstance());
}
/** Records events put into the coordinator event queue. */
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachineTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachineTest.java
index 4dca20f56..2d3177b2a 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachineTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachineTest.java
@@ -312,7 +312,8 @@ class TableBucketStateMachineTest {
zookeeperClient,
new Configuration(),
new LakeCatalogDynamicLoader(new
Configuration(), null, true)),
- kvSnapshotLeaseManager);
+ kvSnapshotLeaseManager,
+ SystemClock.getInstance());
CoordinatorEventManager eventManager =
new CoordinatorEventManager(
coordinatorEventProcessor,
TestingMetricGroups.COORDINATOR_METRICS);
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java
b/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java
index 3d47a8861..ddf52b8ba 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java
@@ -998,6 +998,14 @@ public final class FlussClusterExtension
// reduce testing resources
clusterConf.set(ConfigOptions.NETTY_SERVER_NUM_NETWORK_THREADS, 1);
clusterConf.set(ConfigOptions.NETTY_SERVER_NUM_WORKER_THREADS, 3);
+ // Use short cleanup timeouts so that a slow in-flight drop does
not block
+ // subsequent drops beyond the test's retry window.
+ clusterConf.set(
+
ConfigOptions.COORDINATOR_LIFECYCLE_THROTTLER_INFLIGHT_TIMEOUT,
+ Duration.ofSeconds(2));
+ clusterConf.set(
+
ConfigOptions.COORDINATOR_LIFECYCLE_THROTTLER_TIMEOUT_CHECK_INTERVAL,
+ Duration.ofSeconds(1));
}
/** Sets the number of tablet servers. */
diff --git a/website/docs/maintenance/configuration.md
b/website/docs/maintenance/configuration.md
index ea3fec7fd..0a4223074 100644
--- a/website/docs/maintenance/configuration.md
+++ b/website/docs/maintenance/configuration.md
@@ -56,11 +56,13 @@ during the Fluss cluster working.
## CoordinatorServer
-| Option | Type | Default |
Description
|
-|----------------------------------------------|----------|---------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| coordinator.io-pool.size | Integer | 10 |
**Deprecated**: This option is deprecated. Please use `server.io-pool.size`
instead. The size of the IO thread pool to run blocking operations for
coordinator server. This includes discard unnecessary snapshot files. Increase
this value if you experience slow unnecessary snapshot files clean. The default
value is 10. |
-| coordinator.producer-offsets.ttl | Duration | 24h | The TTL
(time-to-live) for producer offsets. Producer offsets older than this TTL will
be automatically cleaned up by the coordinator server. Producer offsets are
used for undo recovery when a Flink job fails over before completing its first
checkpoint. The default value is 24 hours. |
-| coordinator.producer-offsets.cleanup-interval | Duration | 1h | The
interval for cleaning up expired producer offsets and orphan files in remote
storage. The cleanup task runs periodically to remove expired offsets and any
orphan files that may have been left behind due to incomplete operations. The
default value is 1 hour. |
+| Option | Type | Default
| Description
[...]
+|----------------------------------------------------|------------|-----------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
+| coordinator.io-pool.size | Integer | 10
| **Deprecated**: This option is deprecated. Please use `server.io-pool.size`
instead. The size of the IO thread pool to run blocking operations for
coordinator server. This includes discard unnecessary snapshot files. Increase
this value if you experience slow unnecessary snapshot files clean. The default
value is 10.
[...]
+| coordinator.producer-offsets.ttl | Duration | 24h
| The TTL (time-to-live) for producer offsets. Producer offsets older than this
TTL will be automatically cleaned up by the coordinator server. Producer
offsets are used for undo recovery when a Flink job fails over before
completing its first checkpoint. The default value is 24 hours.
[...]
+| coordinator.producer-offsets.cleanup-interval | Duration | 1h
| The interval for cleaning up expired producer offsets and orphan files in
remote storage. The cleanup task runs periodically to remove expired offsets
and any orphan files that may have been left behind due to incomplete
operations. The default value is 1 hour.
[...]
+| coordinator.lifecycle-throttler.inflight-timeout | Duration | 3min
| The timeout for an in-flight drop event in the coordinator's
TableLifecycleThrottler. If a drop event has been admitted but the
corresponding completion callback has not arrived within this timeout, the
throttler abandons tracking of that drop and continues admitting the next
pending drop.
[...]
+| coordinator.lifecycle-throttler.timeout-check-interval | Duration | 1min
| The periodic interval at which the coordinator's TableLifecycleThrottler
scans in-flight drops for timeouts.
[...]
## TabletServer
