Copilot commented on code in PR #3174:
URL: https://github.com/apache/fluss/pull/3174#discussion_r3279599069
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java:
##########
@@ -494,13 +607,33 @@ private void dropPartitions(
// only remove when zk success, this reflects to the
partitionsByTable
dropIterator.remove();
+ dropped++;
+ bucketsConsumed += numBucketsPerPartition;
LOG.info(
"Auto partitioning deleted partition {} for table
[{}].",
partitionName,
tablePath);
Review Comment:
dropIterator.remove(), dropped++, bucketsConsumed += numBucketsPerPartition,
and the "deleted partition" log currently execute unconditionally even if
dropPartition threw PartitionNotExistException. This makes
throttling/backpressure accounting inaccurate (non-existent partitions don't
fan out deletion events). Restructure so accounting/logging only happen when a
drop actually results in a deletion event (or explicitly separate the "already
absent" case without charging budget).
##########
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java:
##########
@@ -200,6 +200,36 @@ public class ConfigOptions {
"The interval of auto partition check. "
+ "The default value is 10 minutes.");
+ public static final ConfigOption<Integer>
AUTO_PARTITION_DROP_QUEUE_BACKPRESSURE_THRESHOLD =
+ key("auto-partition.drop.queue-backpressure-threshold")
+ .intType()
+ .defaultValue(1000)
+ .withDescription(
+ "When the coordinator event queue size reaches
this threshold, "
+ + "auto partition will skip the drop step
for the current round and retry "
+ + "in the next
AUTO_PARTITION_CHECK_INTERVAL. This prevents overwhelming "
+ + "the coordinator at midnight day
rotation when one partition drop fans "
+ + "out into thousands of bucket/replica
deletion events. Set to a "
+ + "non-positive value to disable
backpressure (always drop).");
Review Comment:
The description references "AUTO_PARTITION_CHECK_INTERVAL" as a literal
token, which won’t be meaningful to users reading generated config docs (it’s
not the actual config key). Prefer referencing the real key (e.g.,
"auto-partition.check.interval") or interpolating
AUTO_PARTITION_CHECK_INTERVAL.key() in the description text for clarity.
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java:
##########
@@ -297,6 +341,12 @@ private void doAutoPartition(long tableId, boolean
forceDoAutoPartition) {
private void doAutoPartition(Instant now, Set<Long> tableIds, boolean
forceDoAutoPartition) {
LOG.info("Start auto partitioning for {} tables at {}.",
tableIds.size(), now);
+
+ // Per-round drop budget across all tables, in unit of bucket-deletion
events.
+ // Non-positive cap disables the budget (drop without limit).
+ boolean unlimitedBudget = dropMaxBucketsPerRound <= 0;
+ int dropBudgetThisRound = unlimitedBudget ? Integer.MAX_VALUE :
dropMaxBucketsPerRound;
+
for (Long tableId : tableIds) {
Review Comment:
The shared per-round drop budget is consumed in iteration order of tableIds
(backed by HashMap.keySet()), which is not deterministic and can lead to
persistent unfairness: the same subset of tables may consistently consume the
budget each round while later tables rarely get a chance to drop. Consider
iterating tables in a stable order (e.g., sort by tableId or tablePath) when
budgeting is enabled to make throttling behavior predictable and avoid
starvation across tables.
##########
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java:
##########
@@ -200,6 +200,36 @@ public class ConfigOptions {
"The interval of auto partition check. "
+ "The default value is 10 minutes.");
+ public static final ConfigOption<Integer>
AUTO_PARTITION_DROP_QUEUE_BACKPRESSURE_THRESHOLD =
+ key("auto-partition.drop.queue-backpressure-threshold")
+ .intType()
+ .defaultValue(1000)
+ .withDescription(
+ "When the coordinator event queue size reaches
this threshold, "
+ + "auto partition will skip the drop step
for the current round and retry "
+ + "in the next
AUTO_PARTITION_CHECK_INTERVAL. This prevents overwhelming "
+ + "the coordinator at midnight day
rotation when one partition drop fans "
+ + "out into thousands of bucket/replica
deletion events. Set to a "
+ + "non-positive value to disable
backpressure (always drop).");
+
+ public static final ConfigOption<Integer>
AUTO_PARTITION_DROP_MAX_BUCKETS_PER_ROUND =
+ key("auto-partition.drop.max-buckets-per-round")
+ .intType()
+ .defaultValue(1000)
+ .withDescription(
+ "Per-round drop budget across all auto-partition
tables, measured "
+ + "in number of bucket-deletion events.
Because dropping one "
+ + "partition fans out into numBuckets *
replicationFactor "
+ + "deletion events on the coordinator
queue, accounting in "
+ + "buckets gives uniform protection
regardless of each table's "
+ + "bucket count. To avoid starving tables
whose single-partition "
+ + "bucket count exceeds the budget, every
table is allowed at "
+ + "least one partition drop per round even
if it temporarily "
+ + "overshoots the budget. Remaining
over-quota drops will be "
+ + "processed in the next
AUTO_PARTITION_CHECK_INTERVAL. Set to "
+ + "a non-positive value to disable the cap
(drop without "
+ + "limit).");
Review Comment:
Same as above: the description mentions "AUTO_PARTITION_CHECK_INTERVAL"
instead of the actual configuration key, which can confuse operators reading
the docs. Consider using AUTO_PARTITION_CHECK_INTERVAL.key() (or the literal
key string) in the description text.
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java:
##########
@@ -335,12 +385,46 @@ private void doAutoPartition(Instant now, Set<Long>
tableIds, boolean forceDoAut
continue;
}
- dropPartitions(
- tablePath,
- tableInfo.getPartitionKeys(),
- createPartitionInstant,
- tableInfo.getTableConfig().getAutoPartitionStrategy(),
- currentPartitions);
+ // Queue-aware backpressure gate: skip this table's drop for the
current round
+ // when the coordinator event queue is heavily backed up. The
skipped partitions
+ // remain in partitionsByTable and will be retried in the next
check interval.
+ boolean skipDrop = false;
+ if (dropQueueBackpressureThreshold > 0) {
+ int qSize = coordinatorEventQueueSizeProvider.getAsInt();
+ if (qSize >= dropQueueBackpressureThreshold) {
+ LOG.info(
+ "Skipping auto-partition drop for table [{}]
(id={}) this round "
+ + "because coordinator event queue size {}
>= threshold {}.",
+ tablePath,
+ tableId,
+ qSize,
+ dropQueueBackpressureThreshold);
+ skipDrop = true;
+ }
+ }
+
+ if (!skipDrop && (unlimitedBudget || dropBudgetThisRound > 0)) {
+ // Use the actual current time for drop to ensure timely
cleanup.
+ // The random delay is only meant for spreading partition
creation load,
+ // not for delaying retention cleanup.
+ int budgetForCall = unlimitedBudget ? Integer.MAX_VALUE :
dropBudgetThisRound;
+ int bucketsConsumed =
+ dropPartitions(
+ tablePath,
+ tableInfo.getPartitionKeys(),
+ now,
+
tableInfo.getTableConfig().getAutoPartitionStrategy(),
+ currentPartitions,
+ budgetForCall,
+ tableInfo.getNumBuckets());
Review Comment:
The drop budget is described/documented as protecting against a fan-out of
`numBuckets * replicationFactor` deletion events, but the implementation
charges only `numBucketsPerPartition` (tableInfo.getNumBuckets()) per dropped
partition. Since TableManager.onDeletePartition triggers replica deletions for
all replicas in the partition (i.e., ~numBuckets * replicationFactor state
changes), the current accounting likely underestimates queue impact and can
allow bursts larger than intended. Consider charging budget in terms of total
replicas affected (e.g., numBuckets * replicationFactor or actual replica count
from assignments) or adjust the docs/config wording to match the implemented
unit.
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java:
##########
@@ -303,7 +303,18 @@ protected void initCoordinatorLeader() throws Exception {
this.coordinatorChannelManager = new
CoordinatorChannelManager(rpcClient);
this.autoPartitionManager =
- new AutoPartitionManager(metadataCache, metadataManager,
conf);
+ new AutoPartitionManager(
+ metadataCache,
+ metadataManager,
+ conf,
+ // Late-bound queue size probe:
coordinatorEventProcessor is built
+ // after autoPartitionManager, so we read the
field on each call.
+ // Returns 0 before the processor is wired up,
which simply disables
+ // the queue-aware backpressure gate during that
brief window.
+ () -> {
+ CoordinatorEventProcessor p =
this.coordinatorEventProcessor;
+ return p == null ? 0 :
p.getCoordinatorEventManager().queueSize();
Review Comment:
The queue-size supplier reads coordinatorEventProcessor from a different
thread than the one that assigns/clears it under the CoordinatorServer lock.
Since coordinatorEventProcessor is @GuardedBy("lock") and not volatile, this
introduces an unsynchronized cross-thread read and can observe stale values.
Consider making coordinatorEventProcessor (or a dedicated
CoordinatorEventManager reference) safely published (e.g.,
volatile/AtomicReference), or have the supplier acquire the same lock when
reading.
--
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]
