This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang 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 7341df07c8d Preserve live instance fields during controller updates
(#20064)
7341df07c8d is described below
commit 7341df07c8df7f202073eccd4c5089b54ba76281
Author: 2748109647 <[email protected]>
AuthorDate: Fri Oct 30 02:37:11 2026 +0800
Preserve live instance fields during controller updates (#20064)
---
.../helix/core/PinotHelixResourceManager.java | 42 ++++----
...otHelixResourceManagerConfigValidationTest.java | 107 +++++++++++++++++++++
2 files changed, 131 insertions(+), 18 deletions(-)
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
index cef5d5460f0..6c200112508 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
@@ -51,6 +51,7 @@ import java.util.TreeMap;
import java.util.TreeSet;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;
@@ -630,30 +631,35 @@ public class PinotHelixResourceManager {
InstanceConfigValidatorRegistry.validate(newInstance);
- // Asymmetric preservation of the operational `queriesDisabled` flag
managed by
- // PUT /instances/{name}/state?state=QUERIES_DISABLE|QUERIES_ENABLE (see
#setQueriesDisabled):
- // - body=true -> set true (legacy path; explicit operator intent in
this request).
- // - body=false -> if live==true, preserve true (the clobber case: a
generic update to
- // host/tags/ports must NOT silently re-route queries to
a server an
- // operator has explicitly removed from routing).
- // - body=false + live!=true -> leave as is (no-op / cleared).
- boolean liveQueriesDisabled =
-
Boolean.parseBoolean(instanceConfig.getRecord().getSimpleField(CommonConstants.Helix.QUERIES_DISABLED));
-
List<String> newTags = newInstance.getTags();
List<String> oldTags = instanceConfig.getTags();
- InstanceUtils.updateHelixInstanceConfig(instanceConfig, newInstance);
- // If the live record had queriesDisabled=true but the request body
defaulted it to false,
- // restore the operational flag so it survives a request-body-driven
update.
- if (liveQueriesDisabled && !newInstance.isQueriesDisabled()) {
-
instanceConfig.getRecord().setSimpleField(CommonConstants.Helix.QUERIES_DISABLED,
Boolean.TRUE.toString());
+ // Apply request-owned fields to the latest Helix record. A participant
can update its
+ // own fields (for example pinotVersion and capabilities) after the read
above but before
+ // this write. A full-record setProperty would overwrite those newer
values; updateProperty
+ // runs this updater against the current record and preserves fields not
changed here.
+ AtomicBoolean preservedQueriesDisabled = new AtomicBoolean();
+ boolean updated =
_helixDataAccessor.updateProperty(_keyBuilder.instanceConfig(instanceId),
record -> {
+ InstanceConfig currentConfig = new InstanceConfig(record);
+ boolean liveQueriesDisabled =
+
Boolean.parseBoolean(record.getSimpleField(CommonConstants.Helix.QUERIES_DISABLED));
+ InstanceUtils.updateHelixInstanceConfig(currentConfig, newInstance);
+ // Preserve the operator-controlled flag when the request body supplies
its default false.
+ // An explicit true in the request remains supported for the legacy
update path.
+ boolean shouldPreserveQueriesDisabled = liveQueriesDisabled &&
!newInstance.isQueriesDisabled();
+ if (shouldPreserveQueriesDisabled) {
+
currentConfig.getRecord().setSimpleField(CommonConstants.Helix.QUERIES_DISABLED,
Boolean.TRUE.toString());
+ }
+ preservedQueriesDisabled.set(shouldPreserveQueriesDisabled);
+ return currentConfig.getRecord();
+ }, instanceConfig);
+ if (!updated) {
+ throw new RuntimeException("Failed to update instance config for
instance: " + instanceId);
+ }
+ if (preservedQueriesDisabled.get()) {
LOGGER.warn("Ignoring request body's queriesDisabled=false for instance:
{} because the live config has "
+ "queriesDisabled=true; use PUT
/instances/{}/state?state=QUERIES_ENABLE to clear the flag.", instanceId,
instanceId);
}
- if
(!_helixDataAccessor.setProperty(_keyBuilder.instanceConfig(instanceId),
instanceConfig)) {
- throw new RuntimeException("Failed to set instance config for instance:
" + instanceId);
- }
// Update broker resource if necessary
boolean shouldUpdateBrokerResource = false;
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerConfigValidationTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerConfigValidationTest.java
index 72143f7704a..59421512f72 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerConfigValidationTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerConfigValidationTest.java
@@ -20,17 +20,22 @@ package org.apache.pinot.controller.helix.core;
import java.lang.reflect.Field;
import java.util.List;
+import java.util.Map;
import java.util.TreeMap;
import org.apache.helix.HelixAdmin;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.PropertyKey;
import org.apache.helix.model.InstanceConfig;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.helix.zookeeper.zkclient.DataUpdater;
import org.apache.pinot.common.utils.config.InstanceUtils;
import org.apache.pinot.controller.helix.core.lineage.LineageManager;
import org.apache.pinot.spi.config.instance.Instance;
import org.apache.pinot.spi.config.instance.InstanceConfigValidatorRegistry;
import org.apache.pinot.spi.config.instance.InstanceType;
import org.apache.pinot.spi.exception.ConfigValidationException;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.mockito.ArgumentMatchers;
import org.mockito.Mockito;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
@@ -118,6 +123,108 @@ public class
PinotHelixResourceManagerConfigValidationTest {
verify(_helixDataAccessor, never()).setProperty(any(PropertyKey.class),
any(InstanceConfig.class));
}
+ @Test
+ public void testUpdateInstancePreservesFieldsWrittenAfterInitialRead() {
+ Instance original =
+ new Instance("localhost", 1234, InstanceType.SERVER,
List.of("DefaultTenant_OFFLINE"), null,
+ 0, 0, 0, 0, false);
+ InstanceConfig initiallyReadConfig =
InstanceUtils.toHelixInstanceConfig(original);
+ initiallyReadConfig.getRecord().setSimpleField("pinotVersion",
"old-version");
+ initiallyReadConfig.getRecord().setMapField("capabilities",
Map.of("pinotVersion", "old-version"));
+
when(_helixDataAccessor.getProperty(any(PropertyKey.class))).thenReturn(initiallyReadConfig);
+
+ // Model a participant restart after updateInstance's read but before its
write.
+ InstanceConfig latestConfig =
InstanceUtils.toHelixInstanceConfig(original);
+ latestConfig.getRecord().setSimpleField("pinotVersion", "new-version");
+ latestConfig.getRecord().setMapField("capabilities",
Map.of("pinotVersion", "new-version", "feature", "1"));
+
latestConfig.getRecord().setSimpleField(CommonConstants.Helix.QUERIES_DISABLED,
Boolean.TRUE.toString());
+ ZNRecord[] persistedRecord = new ZNRecord[1];
+ when(_helixDataAccessor.updateProperty(any(PropertyKey.class),
+ ArgumentMatchers.<DataUpdater<ZNRecord>>any(),
any(InstanceConfig.class))).thenAnswer(invocation -> {
+ DataUpdater<ZNRecord> updater = invocation.getArgument(1);
+ persistedRecord[0] = updater.update(latestConfig.getRecord());
+ return true;
+ });
+
+ Instance updated =
+ new Instance("localhost", 1234, InstanceType.SERVER,
List.of("UpdatedTenant_OFFLINE"), null,
+ 0, 0, 0, 0, false);
+ _resourceManager.updateInstance("Server_localhost_1234", updated, false);
+
+ assertNotNull(persistedRecord[0]);
+ assertEquals(persistedRecord[0].getSimpleField("pinotVersion"),
"new-version");
+ assertEquals(persistedRecord[0].getMapField("capabilities"),
+ Map.of("pinotVersion", "new-version", "feature", "1"));
+
assertEquals(persistedRecord[0].getListField(InstanceConfig.InstanceConfigProperty.TAG_LIST.name()),
+ List.of("UpdatedTenant_OFFLINE"));
+
assertEquals(persistedRecord[0].getSimpleField(CommonConstants.Helix.QUERIES_DISABLED),
Boolean.TRUE.toString());
+ verify(_helixDataAccessor, never()).setProperty(any(PropertyKey.class),
any(InstanceConfig.class));
+ }
+
+ @Test
+ public void testUpdateInstanceRetriesAgainstLatestRecord() {
+ Instance original =
+ new Instance("localhost", 1234, InstanceType.SERVER,
List.of("DefaultTenant_OFFLINE"), null,
+ 0, 0, 0, 0, false);
+ when(_helixDataAccessor.getProperty(any(PropertyKey.class)))
+ .thenReturn(InstanceUtils.toHelixInstanceConfig(original));
+
+ InstanceConfig firstAttemptConfig =
InstanceUtils.toHelixInstanceConfig(original);
+ firstAttemptConfig.getRecord().setSimpleField("pinotVersion", "v2");
+ firstAttemptConfig.getRecord().setMapField("capabilities",
Map.of("pinotVersion", "v2", "feature", "1"));
+
firstAttemptConfig.getRecord().setSimpleField(CommonConstants.Helix.QUERIES_DISABLED,
Boolean.TRUE.toString());
+
+ // A concurrent writer wins the first CAS and re-enables queries before
the updater is retried.
+ InstanceConfig retryConfig = InstanceUtils.toHelixInstanceConfig(original);
+ retryConfig.getRecord().setSimpleField("pinotVersion", "v3");
+ retryConfig.getRecord().setMapField("capabilities", Map.of("pinotVersion",
"v3", "feature", "2"));
+
retryConfig.getRecord().setSimpleField(CommonConstants.Helix.QUERIES_DISABLED,
Boolean.FALSE.toString());
+ ZNRecord[] attemptedRecords = new ZNRecord[2];
+ when(_helixDataAccessor.updateProperty(any(PropertyKey.class),
+ ArgumentMatchers.<DataUpdater<ZNRecord>>any(),
any(InstanceConfig.class))).thenAnswer(invocation -> {
+ DataUpdater<ZNRecord> updater = invocation.getArgument(1);
+ attemptedRecords[0] = updater.update(firstAttemptConfig.getRecord());
+ attemptedRecords[1] = updater.update(retryConfig.getRecord());
+ return true;
+ });
+
+ Instance updated =
+ new Instance("localhost", 1234, InstanceType.SERVER,
List.of("UpdatedTenant_OFFLINE"), null,
+ 0, 0, 0, 0, false);
+ _resourceManager.updateInstance("Server_localhost_1234", updated, false);
+
+ assertNotNull(attemptedRecords[0]);
+ assertEquals(attemptedRecords[0].getSimpleField("pinotVersion"), "v2");
+ assertEquals(attemptedRecords[0].getMapField("capabilities"),
+ Map.of("pinotVersion", "v2", "feature", "1"));
+
assertEquals(attemptedRecords[0].getListField(InstanceConfig.InstanceConfigProperty.TAG_LIST.name()),
+ List.of("UpdatedTenant_OFFLINE"));
+
assertEquals(attemptedRecords[0].getSimpleField(CommonConstants.Helix.QUERIES_DISABLED),
Boolean.TRUE.toString());
+
+ assertNotNull(attemptedRecords[1]);
+ assertEquals(attemptedRecords[1].getSimpleField("pinotVersion"), "v3");
+ assertEquals(attemptedRecords[1].getMapField("capabilities"),
+ Map.of("pinotVersion", "v3", "feature", "2"));
+
assertEquals(attemptedRecords[1].getListField(InstanceConfig.InstanceConfigProperty.TAG_LIST.name()),
+ List.of("UpdatedTenant_OFFLINE"));
+
assertNull(attemptedRecords[1].getSimpleField(CommonConstants.Helix.QUERIES_DISABLED));
+ }
+
+ @Test
+ public void testUpdateInstanceThrowsWhenWriteFails() {
+ Instance instance =
+ new Instance("localhost", 1234, InstanceType.SERVER,
List.of("DefaultTenant_OFFLINE"), null,
+ 0, 0, 0, 0, false);
+ when(_helixDataAccessor.getProperty(any(PropertyKey.class)))
+ .thenReturn(InstanceUtils.toHelixInstanceConfig(instance));
+ when(_helixDataAccessor.updateProperty(any(PropertyKey.class),
+ ArgumentMatchers.<DataUpdater<ZNRecord>>any(),
any(InstanceConfig.class))).thenReturn(false);
+
+ RuntimeException exception = expectThrows(RuntimeException.class,
+ () -> _resourceManager.updateInstance("Server_localhost_1234",
instance, false));
+ assertEquals(exception.getMessage(), "Failed to update instance config for
instance: Server_localhost_1234");
+ }
+
@Test
public void testUpdateInstanceTagsRejectsOnValidation() {
// getHelixInstanceConfig returns an existing server config with pool tags
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
