Copilot commented on code in PR #3174:
URL: https://github.com/apache/fluss/pull/3174#discussion_r3334505984
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/watcher/TableChangeWatcher.java:
##########
@@ -184,13 +201,11 @@ public void event(Type type, ChildData oldData, ChildData
newData) {
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()));
+ replicaCleanupManager.submitTableDrop(
+ table.tableId,
+
tableConfig.getAutoPartitionStrategy().isAutoPartitionEnabled(),
+ tableConfig.isDataLakeEnabled(),
+ table.bucketCount);
Review Comment:
The TableChangeWatcher keeps a `tableId -> bucketCount` cache, but entries
are never evicted when a table is dropped. In long-running coordinators with
table churn this can grow unbounded. Since NODE_DELETED for the table is
handled here, remove the cached entry when processing the table deletion event.
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/watcher/TableChangeWatcher.java:
##########
@@ -277,6 +292,37 @@ private void processTableRegistrationChange(TablePath
tablePath, ChildData newDa
}
}
+ /**
+ * Returns the bucket count owned by a single partition of the given
table. Looks up the local
+ * cache first and falls back to a synchronous {@link
ZooKeeperClient#getTable} read on miss. If
+ * the table znode is already gone (e.g. cascading drop where the table
node was deleted before
+ * this partition NODE_DELETED is delivered), returns {@code 0}; the drop
is still admitted
+ * thanks to the manager's starvation guard.
+ */
Review Comment:
The Javadoc claims bucketCount=0 is still admitted "thanks to the manager's
starvation guard", but in practice the manager normalizes non-positive bucket
counts to a minimum cost (and the starvation guard is about oversized drops).
The comment should describe the actual behavior to avoid misleading readers.
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/watcher/TableChangeWatcher.java:
##########
@@ -277,6 +292,37 @@ private void processTableRegistrationChange(TablePath
tablePath, ChildData newDa
}
}
+ /**
+ * Returns the bucket count owned by a single partition of the given
table. Looks up the local
+ * cache first and falls back to a synchronous {@link
ZooKeeperClient#getTable} read on miss. If
+ * the table znode is already gone (e.g. cascading drop where the table
node was deleted before
+ * this partition NODE_DELETED is delivered), returns {@code 0}; the drop
is still admitted
+ * thanks to the manager's starvation guard.
+ */
+ private int resolveTableBucketCount(long tableId, TablePath tablePath) {
+ Integer cached = tableBucketCount.get(tableId);
+ if (cached != null) {
+ return cached;
+ }
+ try {
+ Optional<TableRegistration> registration =
zooKeeperClient.getTable(tablePath);
+ if (registration.isPresent()) {
+ int buckets = registration.get().bucketCount;
+ tableBucketCount.put(tableId, buckets);
+ return buckets;
+ }
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to read TableRegistration for {} (tableId={})
while resolving "
+ + "bucket count for partition drop. Falling back
to bucketCount=0; "
+ + "the drop will still be admitted via the
starvation guard.",
+ tablePath,
+ tableId,
+ e);
Review Comment:
This WARN message says the drop will be admitted via the starvation guard,
but bucketCount=0 is actually handled by the cleanup manager normalizing it to
a minimum cost. Keeping the log accurate helps when diagnosing throttling
behavior.
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/ReplicaCleanupManager.java:
##########
@@ -0,0 +1,674 @@
+/*
+ * 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.ArrayList;
+import java.util.Collections;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+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. The
synchronous part of a drop
+ * (i.e. the ZooKeeper metadata removal performed by {@link MetadataManager})
is intentionally NOT
+ * handled here: by the time a drop is submitted to this manager the
partition/table is already
+ * invisible to clients.
+ *
+ * <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}
+ * (carrying the drop's bucket count) in an in-memory FIFO queue. The manager
then dispatches drops
+ * to the coordinator event queue as full {@code DropPartitionEvent} / {@code
DropTableEvent},
+ * capped by a per-batch bucket budget. The {@code CoordinatorEventProcessor}
processing logic is
+ * unchanged and continues to drive {@code TableManager} state transitions on
the coordinator's
+ * single event thread.
+ *
+ * <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). The same FIFO +
+ * bucket budget then keeps a startup with thousands of stale tables from
blasting all tablet
+ * servers with stop-replica RPCs at once.
+ *
+ * <p>Submission flow:
+ *
+ * <ol>
+ * <li>A drop source ({@link
+ *
org.apache.fluss.server.coordinator.event.watcher.TableChangeWatcher},
coordinator startup,
+ * etc.) calls {@link #submitPartitionDrop} / {@link #submitTableDrop}
(or their {@code
+ * *ForResume} variants) together with the drop's bucket count.
+ * <li>Pending drops are buffered. {@link #collectNextBatch()} pulls drops
in FIFO order while the
+ * running sum of in-flight bucket counts stays within the per-batch
bucket budget. If the
+ * head-of-queue drop alone already exceeds the budget, it is admitted
on its own to avoid
+ * starvation.
+ * <li>For each admitted drop, the manager either rebuilds the {@code
DropPartitionEvent} / {@code
+ * DropTableEvent} and puts it into the {@link EventManager} (watcher
path) or runs the resume
+ * {@link Runnable} directly (startup reconciliation path).
+ * <li>The coordinator event processor handles the event normally, which
eventually drives the
+ * replica state machine to {@code DeletionSuccessful}. {@code
TableManager} then notifies
+ * this manager via {@link #onPartitionDropCompleted} / {@link
#onTableDropCompleted},
+ * releasing the drop's bucket budget and triggering the next batch.
+ * </ol>
+ *
+ * <p>Timeout-based abandon: each in-flight drop is timestamped at admission
time. A periodic task
+ * scans for drops whose completion callback has not arrived within {@code
+ * coordinator.replica-cleanup.inflight-timeout} and abandons them 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 EventManager eventManager;
+ private final Clock clock;
+ private final ScheduledExecutorService timeoutChecker;
+
+ /**
+ * Per-batch budget measured in number of buckets across the drops
admitted in one batch. A
+ * non-positive value disables the cap. If the head-of-queue drop alone
already exceeds this
+ * budget, it is admitted on its own to avoid starvation.
+ */
+ private final int dropMaxBucketsPerBatch;
+
+ /**
+ * Maximum amount of time, in milliseconds, an in-flight drop may wait for
its completion
+ * callback before being abandoned.
+ */
+ private final long inflightTimeoutMs;
+
+ private final long timeoutCheckIntervalMs;
+
+ 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<>();
+
+ /** Total number of buckets across drops currently admitted but not yet
completed. */
+ @GuardedBy("lock")
+ private int inflightBuckets = 0;
+
+ /** Number of drops currently admitted but not yet completed (for
observability/tests). */
+ @GuardedBy("lock")
+ private int inflightCount = 0;
+
+ /** In-flight partition drops, keyed by {@link TablePartition}. */
+ @GuardedBy("lock")
+ private final Map<TablePartition, InflightDrop> inflightPartitions = new
HashMap<>();
+
+ /** In-flight table drops, keyed by tableId. */
+ @GuardedBy("lock")
+ private final Map<Long, InflightDrop> inflightTables = new HashMap<>();
+
+ public ReplicaCleanupManager(EventManager eventManager, Configuration
conf, Clock clock) {
+ this(
+ eventManager,
+ conf,
+ clock,
+ Executors.newScheduledThreadPool(
+ 1, new
ExecutorThreadFactory("replica-cleanup-timeout")));
+ }
+
+ @VisibleForTesting
+ ReplicaCleanupManager(
+ EventManager eventManager,
+ Configuration conf,
+ Clock clock,
+ ScheduledExecutorService timeoutChecker) {
+ this.eventManager = eventManager;
+ this.clock = clock == null ? SystemClock.getInstance() : clock;
+ this.timeoutChecker = timeoutChecker;
+ this.dropMaxBucketsPerBatch =
+
conf.get(ConfigOptions.COORDINATOR_REPLICA_CLEANUP_MAX_BUCKETS_PER_BATCH);
+ this.inflightTimeoutMs =
+
conf.get(ConfigOptions.COORDINATOR_REPLICA_CLEANUP_INFLIGHT_TIMEOUT).toMillis();
+ this.timeoutCheckIntervalMs =
+
conf.get(ConfigOptions.COORDINATOR_REPLICA_CLEANUP_TIMEOUT_CHECK_INTERVAL)
+ .toMillis();
+ }
+
+ /** Starts the periodic timeout checker. Idempotent; subsequent calls are
no-ops. */
+ 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: maxBucketsPerBatch={},
inflightTimeoutMs={}, "
+ + "timeoutCheckIntervalMs={}",
+ dropMaxBucketsPerBatch,
+ 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.
+ *
+ * @param bucketCount number of buckets owned by this partition; used to
charge the per-batch
+ * bucket budget.
+ */
+ public void submitPartitionDrop(
+ long tableId, long partitionId, String partitionName, int
bucketCount) {
+ admitPartitionDrop(tableId, partitionId, partitionName, bucketCount,
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, where the
+ * regular event-queue path would NPE; the action typically drives {@link
+ * TableManager#onDeletePartition} directly.
+ */
+ public void submitPartitionDropForResume(
+ long tableId,
+ long partitionId,
+ String partitionName,
+ int bucketCount,
+ Runnable resumeAction) {
+ admitPartitionDrop(tableId, partitionId, partitionName, bucketCount,
resumeAction);
+ }
+
+ private void admitPartitionDrop(
+ long tableId,
+ long partitionId,
+ String partitionName,
+ int bucketCount,
+ @Nullable Runnable resumeAction) {
+ int normalized = Math.max(1, bucketCount);
+ List<PendingDrop> batch;
Review Comment:
ReplicaCleanupManager can still accept and admit new drops after `close()`
has been called (only `start()` checks `closed`). During shutdown, watchers may
still be delivering NODE_DELETED callbacks, and accepting new drops after close
can enqueue events onto a coordinator event manager that has already been
stopped, leading to unbounded queue growth and confusing metrics. Guard
submissions when closed (no-op or throw) so `close()` reliably stops further
work.
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/ReplicaCleanupManager.java:
##########
@@ -0,0 +1,674 @@
+/*
+ * 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.ArrayList;
+import java.util.Collections;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+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. The
synchronous part of a drop
+ * (i.e. the ZooKeeper metadata removal performed by {@link MetadataManager})
is intentionally NOT
+ * handled here: by the time a drop is submitted to this manager the
partition/table is already
+ * invisible to clients.
+ *
+ * <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}
+ * (carrying the drop's bucket count) in an in-memory FIFO queue. The manager
then dispatches drops
+ * to the coordinator event queue as full {@code DropPartitionEvent} / {@code
DropTableEvent},
+ * capped by a per-batch bucket budget. The {@code CoordinatorEventProcessor}
processing logic is
+ * unchanged and continues to drive {@code TableManager} state transitions on
the coordinator's
+ * single event thread.
+ *
+ * <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). The same FIFO +
+ * bucket budget then keeps a startup with thousands of stale tables from
blasting all tablet
+ * servers with stop-replica RPCs at once.
+ *
+ * <p>Submission flow:
+ *
+ * <ol>
+ * <li>A drop source ({@link
+ *
org.apache.fluss.server.coordinator.event.watcher.TableChangeWatcher},
coordinator startup,
+ * etc.) calls {@link #submitPartitionDrop} / {@link #submitTableDrop}
(or their {@code
+ * *ForResume} variants) together with the drop's bucket count.
+ * <li>Pending drops are buffered. {@link #collectNextBatch()} pulls drops
in FIFO order while the
+ * running sum of in-flight bucket counts stays within the per-batch
bucket budget. If the
+ * head-of-queue drop alone already exceeds the budget, it is admitted
on its own to avoid
+ * starvation.
+ * <li>For each admitted drop, the manager either rebuilds the {@code
DropPartitionEvent} / {@code
+ * DropTableEvent} and puts it into the {@link EventManager} (watcher
path) or runs the resume
+ * {@link Runnable} directly (startup reconciliation path).
+ * <li>The coordinator event processor handles the event normally, which
eventually drives the
+ * replica state machine to {@code DeletionSuccessful}. {@code
TableManager} then notifies
+ * this manager via {@link #onPartitionDropCompleted} / {@link
#onTableDropCompleted},
+ * releasing the drop's bucket budget and triggering the next batch.
+ * </ol>
+ *
+ * <p>Timeout-based abandon: each in-flight drop is timestamped at admission
time. A periodic task
+ * scans for drops whose completion callback has not arrived within {@code
+ * coordinator.replica-cleanup.inflight-timeout} and abandons them 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 EventManager eventManager;
+ private final Clock clock;
+ private final ScheduledExecutorService timeoutChecker;
+
+ /**
+ * Per-batch budget measured in number of buckets across the drops
admitted in one batch. A
+ * non-positive value disables the cap. If the head-of-queue drop alone
already exceeds this
+ * budget, it is admitted on its own to avoid starvation.
+ */
+ private final int dropMaxBucketsPerBatch;
+
+ /**
+ * Maximum amount of time, in milliseconds, an in-flight drop may wait for
its completion
+ * callback before being abandoned.
+ */
+ private final long inflightTimeoutMs;
+
+ private final long timeoutCheckIntervalMs;
+
+ 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<>();
+
+ /** Total number of buckets across drops currently admitted but not yet
completed. */
+ @GuardedBy("lock")
+ private int inflightBuckets = 0;
+
+ /** Number of drops currently admitted but not yet completed (for
observability/tests). */
+ @GuardedBy("lock")
+ private int inflightCount = 0;
+
+ /** In-flight partition drops, keyed by {@link TablePartition}. */
+ @GuardedBy("lock")
+ private final Map<TablePartition, InflightDrop> inflightPartitions = new
HashMap<>();
+
+ /** In-flight table drops, keyed by tableId. */
+ @GuardedBy("lock")
+ private final Map<Long, InflightDrop> inflightTables = new HashMap<>();
+
+ public ReplicaCleanupManager(EventManager eventManager, Configuration
conf, Clock clock) {
+ this(
+ eventManager,
+ conf,
+ clock,
+ Executors.newScheduledThreadPool(
+ 1, new
ExecutorThreadFactory("replica-cleanup-timeout")));
+ }
+
+ @VisibleForTesting
+ ReplicaCleanupManager(
+ EventManager eventManager,
+ Configuration conf,
+ Clock clock,
+ ScheduledExecutorService timeoutChecker) {
+ this.eventManager = eventManager;
+ this.clock = clock == null ? SystemClock.getInstance() : clock;
+ this.timeoutChecker = timeoutChecker;
+ this.dropMaxBucketsPerBatch =
+
conf.get(ConfigOptions.COORDINATOR_REPLICA_CLEANUP_MAX_BUCKETS_PER_BATCH);
+ this.inflightTimeoutMs =
+
conf.get(ConfigOptions.COORDINATOR_REPLICA_CLEANUP_INFLIGHT_TIMEOUT).toMillis();
+ this.timeoutCheckIntervalMs =
+
conf.get(ConfigOptions.COORDINATOR_REPLICA_CLEANUP_TIMEOUT_CHECK_INTERVAL)
+ .toMillis();
+ }
+
+ /** Starts the periodic timeout checker. Idempotent; subsequent calls are
no-ops. */
+ 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: maxBucketsPerBatch={},
inflightTimeoutMs={}, "
+ + "timeoutCheckIntervalMs={}",
+ dropMaxBucketsPerBatch,
+ 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.
+ *
+ * @param bucketCount number of buckets owned by this partition; used to
charge the per-batch
+ * bucket budget.
+ */
+ public void submitPartitionDrop(
+ long tableId, long partitionId, String partitionName, int
bucketCount) {
+ admitPartitionDrop(tableId, partitionId, partitionName, bucketCount,
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, where the
+ * regular event-queue path would NPE; the action typically drives {@link
+ * TableManager#onDeletePartition} directly.
+ */
+ public void submitPartitionDropForResume(
+ long tableId,
+ long partitionId,
+ String partitionName,
+ int bucketCount,
+ Runnable resumeAction) {
+ admitPartitionDrop(tableId, partitionId, partitionName, bucketCount,
resumeAction);
+ }
+
+ private void admitPartitionDrop(
+ long tableId,
+ long partitionId,
+ String partitionName,
+ int bucketCount,
+ @Nullable Runnable resumeAction) {
+ int normalized = Math.max(1, bucketCount);
+ List<PendingDrop> batch;
+ lock.lock();
+ try {
+ pendingDrops.add(
+ new PendingPartitionDrop(
+ tableId, partitionId, partitionName, normalized,
resumeAction));
+ LOG.debug(
+ "Submitted partition drop: tableId={} partitionId={} ({})
buckets={} "
+ + "resume={}; pending={} inflight={}
inflightBuckets={}",
+ tableId,
+ partitionId,
+ partitionName,
+ normalized,
+ resumeAction != null,
+ pendingDrops.size(),
+ inflightCount,
+ inflightBuckets);
+ batch = collectNextBatch();
+ } finally {
+ lock.unlock();
+ }
+ executeBatch(batch);
+ }
+
+ /**
+ * Submits a table drop request. The table's ZK metadata is expected to
have been removed by the
+ * caller prior to this submission.
+ *
+ * @param bucketCount number of buckets owned by this table at drop time.
For non-partitioned
+ * tables this is the table bucket count; for partitioned tables,
partition znodes are
+ * deleted before the table znode, so by the time the table drop is
submitted the remaining
+ * bucket footprint is typically the per-partition bucket count (one
shard) or smaller.
+ */
+ public void submitTableDrop(
+ long tableId,
+ boolean isAutoPartitionTable,
+ boolean isDataLakeEnabled,
+ int bucketCount) {
+ admitTableDrop(tableId, isAutoPartitionTable, isDataLakeEnabled,
bucketCount, 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, where the regular
+ * event-queue path would NPE; the action typically drives {@link
TableManager#onDeleteTable}
+ * directly.
+ */
+ public void submitTableDropForResume(long tableId, int bucketCount,
Runnable resumeAction) {
+ // For startup resume the auto-partition / data-lake flags would only
have been used to
+ // build a DropTableEvent (which we are bypassing), so they are
intentionally omitted.
+ admitTableDrop(tableId, false, false, bucketCount, resumeAction);
+ }
+
+ private void admitTableDrop(
+ long tableId,
+ boolean isAutoPartitionTable,
+ boolean isDataLakeEnabled,
+ int bucketCount,
+ @Nullable Runnable resumeAction) {
+ int normalized = Math.max(1, bucketCount);
+ List<PendingDrop> batch;
Review Comment:
`close()` should also stop table-drop submissions. Right now, even after
shutdown, `submitTableDrop*` can still enqueue new PendingDrops (and
potentially coordinator events) because `admitTableDrop` doesn't check
`closed`. Add the same closed-guard here as in partition drops to make shutdown
behavior consistent.
--
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]
