This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new e60a8b2e2f0 correctly update for only partial change on "before
serving query" value and regular value (#18693)
e60a8b2e2f0 is described below
commit e60a8b2e2f0b3f4b9ba85cf40d74ac6a824fcbf1
Author: Jhow <[email protected]>
AuthorDate: Sat Jun 6 01:52:54 2026 -0700
correctly update for only partial change on "before serving query" value
and regular value (#18693)
---
.../local/utils/SegmentOperationsThrottler.java | 17 ++++
.../local/utils/SegmentOperationsThrottlerSet.java | 37 +++++---
.../utils/SegmentOperationsThrottlerSetTest.java | 101 +++++++++++++++++++++
3 files changed, 141 insertions(+), 14 deletions(-)
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/SegmentOperationsThrottler.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/SegmentOperationsThrottler.java
index b09d9c8b064..894b37c017d 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/SegmentOperationsThrottler.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/SegmentOperationsThrottler.java
@@ -197,6 +197,23 @@ public class SegmentOperationsThrottler {
LOGGER.info("Throttler {}: Updated total permits: {}", _throttlerName,
totalPermits());
}
+ /**
+ * Partial variant of {@link #updatePermits(int, int)}: a {@code null}
argument leaves the corresponding value
+ * unchanged. Resolving the current value and writing the new permits happen
atomically under this throttler's
+ * lock, so a partial update cannot interleave with another concurrent
permit change. This allows a single
+ * changed config to be applied without resetting the other (unchanged)
config.
+ *
+ * @param maxConcurrency new max concurrency value, or {@code null} to keep
the current value
+ * @param maxConcurrencyBeforeServingQueries new max concurrency before
serving queries value, or {@code null} to
+ * keep the current value
+ */
+ public synchronized void updatePermits(@Nullable Integer maxConcurrency,
+ @Nullable Integer maxConcurrencyBeforeServingQueries) {
+ updatePermits(maxConcurrency == null ? _maxConcurrency : maxConcurrency,
+ maxConcurrencyBeforeServingQueries == null ?
_maxConcurrencyBeforeServingQueries
+ : maxConcurrencyBeforeServingQueries);
+ }
+
/**
* Block trying to acquire the semaphore to perform the segment operation
steps unless interrupted.
* <p>
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/SegmentOperationsThrottlerSet.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/SegmentOperationsThrottlerSet.java
index 2416e8a1a84..9aa29b89449 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/SegmentOperationsThrottlerSet.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/SegmentOperationsThrottlerSet.java
@@ -155,24 +155,33 @@ public class SegmentOperationsThrottlerSet implements
PinotClusterConfigChangeLi
return;
}
- // Parse maxConcurrency. 0 is a valid kill-switch value; negative values
are treated as misconfiguration.
- int maxConcurrency = parseConfigValue(clusterConfigs,
maxConcurrencyConfigKey, maxConcurrencyDefault);
- if (maxConcurrency < 0) {
- return;
+ // Only parse the config that actually changed; pass null for the
unchanged one so the throttler keeps its
+ // current value. This prevents a single changed config from inadvertently
resetting the other. 0 is a valid
+ // kill-switch value; negative values are treated as misconfiguration.
+ Integer maxConcurrency = null;
+ if (maxConcurrencyChanged) {
+ int parsed = parseConfigValue(clusterConfigs, maxConcurrencyConfigKey,
maxConcurrencyDefault);
+ if (parsed < 0) {
+ return;
+ }
+ maxConcurrency = parsed;
}
- // Parse maxConcurrencyBeforeServingQueries. Same handling as
maxConcurrency.
- int maxConcurrencyBeforeServingQueries =
- parseConfigValue(clusterConfigs,
maxConcurrencyBeforeServingQueriesConfigKey,
- maxConcurrencyBeforeServingQueriesDefault);
- if (maxConcurrencyBeforeServingQueries < 0) {
- return;
+ Integer maxConcurrencyBeforeServingQueries = null;
+ if (maxConcurrencyBeforeServingQueriesChanged) {
+ int parsed = parseConfigValue(clusterConfigs,
maxConcurrencyBeforeServingQueriesConfigKey,
+ maxConcurrencyBeforeServingQueriesDefault);
+ if (parsed < 0) {
+ return;
+ }
+ maxConcurrencyBeforeServingQueries = parsed;
}
- // Update throttler
- LOGGER.info("Updating {} throttler with maxConcurrency: {},
maxConcurrencyBeforeServingQueries: {}",
- throttlerName, maxConcurrency, maxConcurrencyBeforeServingQueries);
- if (maxConcurrency == 0 || maxConcurrencyBeforeServingQueries == 0) {
+ // Update throttler (null = unchanged)
+ LOGGER.info("Updating {} throttler with maxConcurrency: {},
maxConcurrencyBeforeServingQueries: {} "
+ + "(null = unchanged)", throttlerName, maxConcurrency,
maxConcurrencyBeforeServingQueries);
+ if ((maxConcurrency != null && maxConcurrency == 0)
+ || (maxConcurrencyBeforeServingQueries != null &&
maxConcurrencyBeforeServingQueries == 0)) {
LOGGER.warn("Throttler {} configured with 0 permits (maxConcurrency: {},
maxConcurrencyBeforeServingQueries: "
+ "{}) — all operations will block until value is raised",
throttlerName, maxConcurrency, maxConcurrencyBeforeServingQueries);
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/SegmentOperationsThrottlerSetTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/SegmentOperationsThrottlerSetTest.java
index 98610462c4a..00c8098df64 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/SegmentOperationsThrottlerSetTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/SegmentOperationsThrottlerSetTest.java
@@ -387,6 +387,107 @@ public class SegmentOperationsThrottlerSetTest {
}
}
+ @Test
+ public void testChangeOnlyMaxConcurrencyPreservesBeforeServingQueries() {
+ // Regression: changing only maxConcurrency must not reset
maxConcurrencyBeforeServingQueries (which defaults to
+ // Integer.MAX_VALUE and would have been clobbered before the fix).
+ int initialMaxConcurrency = 4;
+ int initialBeforeServing = 8;
+ SegmentOperationsThrottler[] throttlers =
createThrottlers(initialMaxConcurrency, initialBeforeServing, false);
+ SegmentOperationsThrottlerSet sot =
wrapInSegmentOperationsThrottler(throttlers);
+ for (int i = 0; i < NUM_THROTTLERS; i++) {
+ SegmentOperationsThrottler t = throttlers[i];
+ // Not serving queries yet, so permits reflect the before-serving value.
+ Assert.assertEquals(t.totalPermits(), initialBeforeServing);
+
+ // Change only maxConcurrency.
+ Map<String, String> updatedClusterConfigs = new HashMap<>();
+ updatedClusterConfigs.put(PARALLELISM_KEYS[i],
String.valueOf(initialMaxConcurrency * 2));
+ sot.onChange(updatedClusterConfigs.keySet(), updatedClusterConfigs);
+
+ // Before-serving value must be preserved (permits unchanged while not
serving).
+ Assert.assertEquals(t.totalPermits(), initialBeforeServing);
+
+ // After serving begins, permits reflect the newly-applied
maxConcurrency.
+ t.startServingQueries();
+ Assert.assertEquals(t.totalPermits(), initialMaxConcurrency * 2);
+ }
+ }
+
+ @Test
+ public void testChangeOnlyBeforeServingQueriesPreservesMaxConcurrency() {
+ // Regression: changing only maxConcurrencyBeforeServingQueries must not
reset maxConcurrency (which defaults to
+ // Integer.MAX_VALUE and would have been clobbered before the fix).
+ int initialMaxConcurrency = 4;
+ int initialBeforeServing = 8;
+ SegmentOperationsThrottler[] throttlers =
createThrottlers(initialMaxConcurrency, initialBeforeServing, false);
+ SegmentOperationsThrottlerSet sot =
wrapInSegmentOperationsThrottler(throttlers);
+ for (int i = 0; i < NUM_THROTTLERS; i++) {
+ SegmentOperationsThrottler t = throttlers[i];
+ Assert.assertEquals(t.totalPermits(), initialBeforeServing);
+
+ // Change only maxConcurrencyBeforeServingQueries.
+ Map<String, String> updatedClusterConfigs = new HashMap<>();
+ updatedClusterConfigs.put(PARALLELISM_BEFORE_SERVING_KEYS[i],
String.valueOf(initialBeforeServing * 2));
+ sot.onChange(updatedClusterConfigs.keySet(), updatedClusterConfigs);
+
+ // Before-serving value updated.
+ Assert.assertEquals(t.totalPermits(), initialBeforeServing * 2);
+
+ // After serving begins, the preserved maxConcurrency is applied.
+ t.startServingQueries();
+ Assert.assertEquals(t.totalPermits(), initialMaxConcurrency);
+ }
+ }
+
+ @Test
+ public void testInvalidChangedConfigLeavesBothValuesUntouched() {
+ // An unparseable changed config triggers the early return before
updatePermits; neither the changed nor the
+ // unchanged value should be modified.
+ int initialMaxConcurrency = 4;
+ int initialBeforeServing = 8;
+ SegmentOperationsThrottler[] throttlers =
createThrottlers(initialMaxConcurrency, initialBeforeServing, false);
+ SegmentOperationsThrottlerSet sot =
wrapInSegmentOperationsThrottler(throttlers);
+ for (int i = 0; i < NUM_THROTTLERS; i++) {
+ SegmentOperationsThrottler t = throttlers[i];
+ Assert.assertEquals(t.totalPermits(), initialBeforeServing);
+
+ // Change only maxConcurrency to a non-numeric value.
+ Map<String, String> updatedClusterConfigs = new HashMap<>();
+ updatedClusterConfigs.put(PARALLELISM_KEYS[i], "not-a-number");
+ sot.onChange(updatedClusterConfigs.keySet(), updatedClusterConfigs);
+
+ // Before-serving value untouched (still not serving).
+ Assert.assertEquals(t.totalPermits(), initialBeforeServing);
+ // maxConcurrency untouched: after serving begins, the original value
applies.
+ t.startServingQueries();
+ Assert.assertEquals(t.totalPermits(), initialMaxConcurrency);
+ }
+ }
+
+ @Test
+ public void
testChangeOnlyMaxConcurrencyToZeroAppliesKillSwitchAndPreservesBeforeServing() {
+ // 0 is a valid kill-switch for maxConcurrency; changing only it must not
disturb the before-serving value.
+ int initialMaxConcurrency = 4;
+ int initialBeforeServing = 8;
+ SegmentOperationsThrottler[] throttlers =
createThrottlers(initialMaxConcurrency, initialBeforeServing, false);
+ SegmentOperationsThrottlerSet sot =
wrapInSegmentOperationsThrottler(throttlers);
+ for (int i = 0; i < NUM_THROTTLERS; i++) {
+ SegmentOperationsThrottler t = throttlers[i];
+ Assert.assertEquals(t.totalPermits(), initialBeforeServing);
+
+ Map<String, String> updatedClusterConfigs = new HashMap<>();
+ updatedClusterConfigs.put(PARALLELISM_KEYS[i], "0");
+ sot.onChange(updatedClusterConfigs.keySet(), updatedClusterConfigs);
+
+ // Before-serving value preserved while not serving.
+ Assert.assertEquals(t.totalPermits(), initialBeforeServing);
+ // Kill switch takes effect once serving begins.
+ t.startServingQueries();
+ Assert.assertEquals(t.totalPermits(), 0);
+ }
+ }
+
@Test
public void testDecreaseSegmentPreprocessParallelism()
throws Exception {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
