platinumhamburg commented on code in PR #3174:
URL: https://github.com/apache/fluss/pull/3174#discussion_r3279660644
##########
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;
Review Comment:
dropBudgetThisRound appears to be a local variable. Why is it necessary to
recompute a new variable based on it?
##########
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)) {
Review Comment:
dropBudgetThisRound is always greater than 0.
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java:
##########
@@ -467,6 +569,9 @@ private void dropPartitions(
// (a=?,dt=20250506,b=?) (a=?,dt=20250507,b=?) will be retained.
Iterator<Map.Entry<String, Set<String>>> iterator =
currentPartitions.headMap(lastRetainPartitionTime).entrySet().iterator();
+ int dropped = 0;
Review Comment:
Why is this variable needed?
--
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]
