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 df9eebc6cd6 add api for server query route enable/disable (#18190)
df9eebc6cd6 is described below
commit df9eebc6cd6d34d21639d01bb296674f7e8c649b
Author: Hongkun Xu <[email protected]>
AuthorDate: Sat Jun 6 13:23:56 2026 +0800
add api for server query route enable/disable (#18190)
---
.../pinot/client/admin/InstanceAdminClient.java | 20 ++--
.../resources/PinotInstanceRestletResource.java | 34 ++++++-
.../pinot/controller/api/resources/StateType.java | 2 +-
.../helix/core/PinotHelixResourceManager.java | 39 +++++++
pinot-controller/src/main/resources/app/Models.ts | 2 +
.../main/resources/app/pages/InstanceDetails.tsx | 48 +++++++++
.../api/PinotInstanceRestletResourceTest.java | 113 +++++++++++++++++++++
.../tests/BaseClusterIntegrationTestSet.java | 110 ++++++++++++++++++++
.../tests/BaseRealtimeClusterIntegrationTest.java | 7 ++
.../tests/HybridClusterIntegrationTest.java | 7 ++
...nsumptionRateLimiterClusterIntegrationTest.java | 7 ++
...aIncreaseDecreasePartitionsIntegrationTest.java | 7 ++
12 files changed, 382 insertions(+), 14 deletions(-)
diff --git
a/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/InstanceAdminClient.java
b/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/InstanceAdminClient.java
index 9f15dfe3fa3..05131a22c8e 100644
---
a/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/InstanceAdminClient.java
+++
b/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/InstanceAdminClient.java
@@ -132,15 +132,17 @@ public class InstanceAdminClient extends
BaseServiceAdminClient {
return response.toString();
}
- /**
- * Enables, disables, or drains an instance via PUT with a state query
parameter.
- * DRAIN state is only applicable to minion instances.
- *
- * @param instanceName Name of the instance
- * @param state State to set (enable, disable, or drop, drain)
- * @return Success response
- * @throws PinotAdminException If the request fails
- */
+ /// Changes the operational state of an instance via PUT with a state query
parameter.
+ ///
+ /// Supported states: `enable`, `disable`, `drain`, `queries_disable`,
`queries_enable`.
+ /// - `DRAIN` is only applicable to minion instances.
+ /// - `QUERIES_DISABLE` / `QUERIES_ENABLE` are only applicable to server
instances and control
+ /// whether brokers route queries to the server.
+ ///
+ /// @param instanceName Name of the instance
+ /// @param state State to set (`enable`, `disable`, `drain`,
`queries_disable`, `queries_enable`)
+ /// @return Success response
+ /// @throws PinotAdminException If the request fails
public String updateInstanceState(String instanceName, String state)
throws PinotAdminException {
Map<String, String> queryParams = Map.of("state", state);
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotInstanceRestletResource.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotInstanceRestletResource.java
index de16e8ed20d..9457fba5ef4 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotInstanceRestletResource.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotInstanceRestletResource.java
@@ -260,9 +260,12 @@ public class PinotInstanceRestletResource {
@Authenticate(AccessType.UPDATE)
@Produces(MediaType.APPLICATION_JSON)
@Consumes(MediaType.TEXT_PLAIN)
- @ApiOperation(value = "Enable/disable/drain an instance", notes =
"Enable/disable/drain an instance. "
- + "DRAIN state is only applicable to minion instances and retags them
from minion_untagged to minion_drained, "
- + "preventing new task assignments while allowing existing tasks to
complete.")
+ @ApiOperation(value = "Enable/disable/drain an instance or toggle query
routing", notes =
+ "Enable/disable/drain an instance. "
+ + "DRAIN state is only applicable to minion instances and retags
them from minion_untagged to "
+ + "minion_drained, preventing new task assignments while allowing
existing tasks to complete. "
+ + "QUERIES_DISABLE / QUERIES_ENABLE are only applicable to server
instances and control whether "
+ + "brokers route queries to the server.")
@ApiResponses(value = {
@ApiResponse(code = 200, message = "Success"),
@ApiResponse(code = 400, message = "Bad Request"),
@@ -273,7 +276,8 @@ public class PinotInstanceRestletResource {
@ApiParam(value = "Instance name", required = true, example =
"Server_a.b.com_20000 | Broker_my.broker.com_30000 "
+ "| Minion_hostname_9514")
@PathParam("instanceName") String instanceName,
- @ApiParam(value = "enable|disable|drain", required = true)
@QueryParam("state") String state) {
+ @ApiParam(value = "enable|disable|drain|queries_disable|queries_enable",
required = true)
+ @QueryParam("state") String state) {
if (!_pinotHelixResourceManager.instanceExists(instanceName)) {
throw new ControllerApplicationException(LOGGER, "Instance '" +
instanceName + "' does not exist",
Response.Status.NOT_FOUND);
@@ -305,6 +309,28 @@ public class PinotInstanceRestletResource {
"Failed to drain minion instance '" + instanceName + "': " +
response.getMessage(),
Response.Status.INTERNAL_SERVER_ERROR);
}
+ } else if (StateType.QUERIES_DISABLE.name().equalsIgnoreCase(state)
+ || StateType.QUERIES_ENABLE.name().equalsIgnoreCase(state)) {
+ // QUERIES_DISABLE / QUERIES_ENABLE only toggle the broker routing flag
for a server instance.
+ // They do NOT change the participant's lifecycle (HELIX_ENABLED) -- the
instance stays online
+ // and continues to serve queries that are already routed to it; the
broker simply stops
+ // sending new queries.
+ if (!InstanceTypeUtils.isServer(instanceName)) {
+ throw new ControllerApplicationException(LOGGER,
+ "QUERIES_DISABLE/QUERIES_ENABLE only applies to server instances.
Instance '" + instanceName
+ + "' is not a server.", Response.Status.BAD_REQUEST);
+ }
+ boolean disable =
StateType.QUERIES_DISABLE.name().equalsIgnoreCase(state);
+ try {
+ _pinotHelixResourceManager.setQueriesDisabled(instanceName, disable);
+ } catch (IllegalArgumentException e) {
+ // Defensive: setQueriesDisabled's own Preconditions check throws
IllegalArgumentException.
+ // The InstanceTypeUtils.isServer check above already filters this
case, but if the underlying
+ // contract changes, surface a 400 rather than a 500.
+ throw new ControllerApplicationException(LOGGER, e.getMessage(),
Response.Status.BAD_REQUEST, e);
+ }
+ return new SuccessResponse(
+ (disable ? "Disabled" : "Enabled") + " query routing for server
instance '" + instanceName + "'");
} else {
throw new ControllerApplicationException(LOGGER, "Unknown state '" +
state + "'", Response.Status.BAD_REQUEST);
}
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/StateType.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/StateType.java
index 879b49e368c..7f4a85ba5ec 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/StateType.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/StateType.java
@@ -19,5 +19,5 @@
package org.apache.pinot.controller.api.resources;
public enum StateType {
- ENABLE, DISABLE, DROP, DRAIN
+ ENABLE, DISABLE, DROP, DRAIN, QUERIES_DISABLE, QUERIES_ENABLE
}
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 e0175a146b7..24d779e2507 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
@@ -626,9 +626,27 @@ 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());
+ 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);
}
@@ -659,6 +677,27 @@ public class PinotHelixResourceManager {
}
}
+ /// Sets the `queriesDisabled` flag on the instance config using a
field-scoped Helix write.
+ ///
+ /// This avoids clobbering other fields (e.g. `shutdownInProgress`) that are
concurrently set
+ /// by other field-scoped writes during rolling upgrades. Full-record writes
([#updateInstance])
+ /// preserve this flag explicitly; see that method for details.
+ ///
+ /// @param serverInstanceName the Helix participant id (must be a server
instance)
+ /// @param disabled `true` to disable query routing, `false` to re-enable it
+ /// @throws IllegalArgumentException if `serverInstanceName` is not a server
instance
+ public synchronized void setQueriesDisabled(String serverInstanceName,
boolean disabled) {
+ Preconditions.checkArgument(InstanceTypeUtils.isServer(serverInstanceName),
+ "setQueriesDisabled only applies to server instances, got: %s",
serverInstanceName);
+ Map<String, String> propToUpdate =
+ Collections.singletonMap(CommonConstants.Helix.QUERIES_DISABLED,
Boolean.toString(disabled));
+ HelixConfigScope scope =
+ new
HelixConfigScopeBuilder(HelixConfigScope.ConfigScopeProperty.PARTICIPANT,
_helixClusterName)
+ .forParticipant(serverInstanceName).build();
+ LOGGER.info("Setting queriesDisabled={} for server instance: {}",
disabled, serverInstanceName);
+ _helixAdmin.setConfig(scope, propToUpdate);
+ }
+
/**
* Updates the tags of the specified instance ID
*/
diff --git a/pinot-controller/src/main/resources/app/Models.ts
b/pinot-controller/src/main/resources/app/Models.ts
index 1d9f6de5252..ca2c2926ca0 100644
--- a/pinot-controller/src/main/resources/app/Models.ts
+++ b/pinot-controller/src/main/resources/app/Models.ts
@@ -60,6 +60,8 @@ export enum DISPLAY_SEGMENT_STATUS {
export enum InstanceState {
ENABLE = 'enable',
DISABLE = 'disable',
+ QUERIES_DISABLE = 'queries_disable',
+ QUERIES_ENABLE = 'queries_enable',
}
export enum InstanceType {
diff --git a/pinot-controller/src/main/resources/app/pages/InstanceDetails.tsx
b/pinot-controller/src/main/resources/app/pages/InstanceDetails.tsx
index 8aaabd4a64e..a4e5e17052e 100644
--- a/pinot-controller/src/main/resources/app/pages/InstanceDetails.tsx
+++ b/pinot-controller/src/main/resources/app/pages/InstanceDetails.tsx
@@ -97,6 +97,8 @@ const InstanceDetails = ({ match }:
RouteComponentProps<Props>) => {
enabled: true,
});
+ const [queriesDisabled, setQueriesDisabled] = useState(false);
+
const [showEditTag, setShowEditTag] = useState(false);
const [showEditConfig, setShowEditConfig] = useState(false);
const { dispatch } = React.useContext(NotificationContext);
@@ -135,6 +137,7 @@ const InstanceDetails = ({ match }:
RouteComponentProps<Props>) => {
queriesDisabled: instanceDetails.queriesDisabled,
};
setState({ enabled: instanceDetails.enabled });
+ setQueriesDisabled(String(instanceDetails.queriesDisabled) === 'true');
setInstanceDetails(JSON.stringify(instancePutObj, null, 2));
setLiveConfig(JSON.stringify(liveConfigResponse, null, 2));
}
@@ -245,6 +248,32 @@ const InstanceDetails = ({ match }:
RouteComponentProps<Props>) => {
closeDialog();
};
+ const handleQueryRouteChange = () => {
+ setDialogDetails({
+ title: queriesDisabled ? 'Enable Query Route' : 'Disable Query Route',
+ content: `Are you sure you want to ${
+ queriesDisabled ? 'enable' : 'disable'
+ } query routing for this server?`,
+ successCb: () => toggleQueryRoute(),
+ });
+ setConfirmDialog(true);
+ };
+
+ const toggleQueryRoute = async () => {
+ const result = await PinotMethodUtils.toggleInstanceState(
+ instanceName,
+ queriesDisabled ? InstanceState.QUERIES_ENABLE :
InstanceState.QUERIES_DISABLE
+ );
+ if (result.status) {
+ dispatch({ type: 'success', message: result.status, show: true });
+ setQueriesDisabled(!queriesDisabled);
+ fetchData();
+ } else {
+ dispatch({ type: 'error', message: result.error, show: true });
+ }
+ closeDialog();
+ };
+
const handleConfigChange = (value: string) => {
setConfig(value);
};
@@ -339,6 +368,25 @@ const InstanceDetails = ({ match }:
RouteComponentProps<Props>) => {
label="Enable"
/>
</Tooltip>
+ {instanceType === InstanceType.SERVER && (
+ <Tooltip
+ title="Controls whether brokers route queries to this
server. Disabling removes the server from the broker routing table."
+ arrow
+ placement="top-start"
+ >
+ <FormControlLabel
+ control={
+ <Switch
+ checked={!queriesDisabled}
+ onChange={handleQueryRouteChange}
+ name="queryRoute"
+ color="primary"
+ />
+ }
+ label="Query Route"
+ />
+ </Tooltip>
+ )}
</div>
</SimpleAccordion>
</div>
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotInstanceRestletResourceTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotInstanceRestletResourceTest.java
index 82fa672d84b..8ea287452ac 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotInstanceRestletResourceTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotInstanceRestletResourceTest.java
@@ -30,6 +30,7 @@ import java.util.Map;
import java.util.TreeMap;
import javax.annotation.Nullable;
import org.apache.pinot.client.admin.PinotAdminClient;
+import org.apache.pinot.client.admin.PinotAdminValidationException;
import org.apache.pinot.controller.api.resources.InstanceTagUpdateRequest;
import org.apache.pinot.controller.api.resources.OperationValidationResponse;
import org.apache.pinot.controller.helix.ControllerTest;
@@ -49,6 +50,7 @@ import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertThrows;
import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
/**
@@ -437,6 +439,117 @@ public class PinotInstanceRestletResourceTest extends
ControllerTest {
adminClient.getInstanceClient().dropInstance(minionInstanceId);
}
+ @Test
+ public void testQueriesDisableOnServerInstance()
+ throws Exception {
+ PinotAdminClient adminClient = DEFAULT_INSTANCE.getOrCreateAdminClient();
+
+ Instance serverInstance =
+ new Instance("queryroute.test.com", 20000, InstanceType.SERVER, null,
null, 0, 0, 0, 0, false);
+
adminClient.getInstanceClient().createInstance(serverInstance.toJsonString());
+ String instanceId = "Server_queryroute.test.com_20000";
+ try {
+ // Queries routed by default
+ checkInstanceInfo(adminClient, instanceId, "queryroute.test.com", 20000,
new String[0], null,
+ -1, -1, -1, -1, false);
+
+ // Disable query routing via toggleInstanceState
+ adminClient.getInstanceClient().updateInstanceState(instanceId,
"QUERIES_DISABLE");
+ checkInstanceInfo(adminClient, instanceId, "queryroute.test.com", 20000,
new String[0], null,
+ -1, -1, -1, -1, true);
+
+ // Re-enable query routing via toggleInstanceState
+ adminClient.getInstanceClient().updateInstanceState(instanceId,
"QUERIES_ENABLE");
+ checkInstanceInfo(adminClient, instanceId, "queryroute.test.com", 20000,
new String[0], null,
+ -1, -1, -1, -1, false);
+ } finally {
+ adminClient.getInstanceClient().dropInstance(instanceId);
+ }
+ }
+
+ @Test
+ public void testUpdateInstancePreservesQueriesDisabled()
+ throws Exception {
+ // Regression test: a generic PUT /instances/{name} with the request
body's queriesDisabled defaulted
+ // to false must NOT clobber an operationally-set queriesDisabled=true
(operator workflow:
+ // 1. operator calls QUERIES_DISABLE to remove server from broker
routing,
+ // 2. some automation later updates host/tags/ports via PUT
/instances/{name},
+ // 3. server must remain out of routing).
+ PinotAdminClient adminClient = DEFAULT_INSTANCE.getOrCreateAdminClient();
+
+ Instance serverInstance =
+ new Instance("preserve.test.com", 20000, InstanceType.SERVER, null,
null, 0, 0, 0, 0, false);
+
adminClient.getInstanceClient().createInstance(serverInstance.toJsonString());
+ String instanceId = "Server_preserve.test.com_20000";
+ try {
+ // Set queriesDisabled=true via the dedicated state API.
+ adminClient.getInstanceClient().updateInstanceState(instanceId,
"QUERIES_DISABLE");
+ checkInstanceInfo(adminClient, instanceId, "preserve.test.com", 20000,
new String[0], null,
+ -1, -1, -1, -1, true);
+
+ // PUT /instances/{name} with a request body where queriesDisabled
defaults to false (legacy operator path).
+ Instance updateBody =
+ new Instance("preserve.test.com", 20000, InstanceType.SERVER, null,
null, 0, 0, 0, 0, false);
+ adminClient.getInstanceClient().updateInstance(instanceId,
updateBody.toJsonString());
+
+ // queriesDisabled must still be true (operational flag preserved).
+ checkInstanceInfo(adminClient, instanceId, "preserve.test.com", 20000,
new String[0], null,
+ -1, -1, -1, -1, true);
+
+ // Operator can still clear the flag via the dedicated state API.
+ adminClient.getInstanceClient().updateInstanceState(instanceId,
"QUERIES_ENABLE");
+ checkInstanceInfo(adminClient, instanceId, "preserve.test.com", 20000,
new String[0], null,
+ -1, -1, -1, -1, false);
+ } finally {
+ adminClient.getInstanceClient().dropInstance(instanceId);
+ }
+ }
+
+ @Test
+ public void testQueriesDisableOnNonServerInstanceFails()
+ throws Exception {
+ PinotAdminClient adminClient = DEFAULT_INSTANCE.getOrCreateAdminClient();
+
+ // Broker instance
+ Instance brokerInstance =
+ new Instance("queryroute-broker.test.com", 30000, InstanceType.BROKER,
null, null, 0, 0, 0, 0, false);
+
adminClient.getInstanceClient().createInstance(brokerInstance.toJsonString());
+ String brokerId = "Broker_queryroute-broker.test.com_30000";
+ try {
+ // Expect HTTP 400 BAD_REQUEST surfaced as
PinotAdminValidationException; verify the message
+ // identifies the failing precondition so a future regression that
throws a generic 500 fails this test.
+ PinotAdminValidationException ex =
expectThrows(PinotAdminValidationException.class,
+ () -> adminClient.getInstanceClient().updateInstanceState(brokerId,
"QUERIES_DISABLE"));
+ assertTrue(ex.getMessage().contains("status: 400"),
+ "Expected HTTP 400 BAD_REQUEST, got: " + ex.getMessage());
+ assertTrue(ex.getMessage().contains("only applies to server instances"),
+ "Expected message to mention server-only precondition, got: " +
ex.getMessage());
+ } finally {
+ adminClient.getInstanceClient().dropInstance(brokerId);
+ }
+ }
+
+ @Test
+ public void testQueriesDisableOnMinionInstanceFails()
+ throws Exception {
+ PinotAdminClient adminClient = DEFAULT_INSTANCE.getOrCreateAdminClient();
+
+ Instance minionInstance =
+ new Instance("queryroute-minion.test.com", 9520, InstanceType.MINION,
null, null, 0, 0, 0, 0, false);
+
adminClient.getInstanceClient().createInstance(minionInstance.toJsonString());
+ String minionId = "Minion_queryroute-minion.test.com_9520";
+ try {
+ PinotAdminValidationException ex =
expectThrows(PinotAdminValidationException.class,
+ () -> adminClient.getInstanceClient().updateInstanceState(minionId,
"QUERIES_DISABLE"));
+ assertTrue(ex.getMessage().contains("status: 400"),
+ "Expected HTTP 400 BAD_REQUEST, got: " + ex.getMessage());
+ assertTrue(ex.getMessage().contains("only applies to server instances"),
+ "Expected message to mention server-only precondition, got: " +
ex.getMessage());
+ } finally {
+ adminClient.getInstanceClient().dropInstance(minionId);
+ }
+ }
+
@AfterClass
public void tearDown()
throws Exception {
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTestSet.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTestSet.java
index 244b22a15d4..0e0f1310f48 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTestSet.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTestSet.java
@@ -22,6 +22,7 @@ import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -31,8 +32,10 @@ import java.util.Set;
import org.apache.commons.lang3.StringUtils;
import org.apache.helix.PropertyKey;
import org.apache.helix.model.ExternalView;
+import org.apache.helix.model.HelixConfigScope;
import org.apache.helix.model.IdealState;
import org.apache.helix.model.InstanceConfig;
+import org.apache.helix.model.builder.HelixConfigScopeBuilder;
import org.apache.pinot.client.ResultSet;
import org.apache.pinot.client.ResultSetGroup;
import org.apache.pinot.client.admin.PinotAdminException;
@@ -44,9 +47,11 @@ import
org.apache.pinot.common.restlet.resources.ServerRebalanceJobStatusRespons
import
org.apache.pinot.common.restlet.resources.TableSegmentsReloadCheckResponse;
import org.apache.pinot.common.restlet.resources.TableView;
import org.apache.pinot.common.utils.LLCSegmentName;
+import org.apache.pinot.common.utils.config.InstanceUtils;
import org.apache.pinot.controller.util.ConsumingSegmentInfoReader;
import org.apache.pinot.core.query.utils.idset.IdSet;
import org.apache.pinot.core.query.utils.idset.IdSets;
+import org.apache.pinot.spi.config.instance.Instance;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.TenantConfig;
@@ -605,6 +610,111 @@ public abstract class BaseClusterIntegrationTestSet
extends BaseClusterIntegrati
}
}
+ /// Test that:
+ ///
+ /// 1. setting `queriesDisabled` via the controller API removes and re-adds
a server from broker routing;
+ /// 2. concurrent field-scoped updates to `shutdownInProgress` do not
clobber `queriesDisabled`;
+ /// 3. a full-record `PUT /instances/{name}` (updateInstance) does NOT
clobber the operational
+ /// `queriesDisabled` flag, even when the request body's default value is
`false`.
+ public void testQueriesDisabled()
+ throws Exception {
+ List<String> instances =
_helixAdmin.getInstancesInCluster(getHelixClusterName());
+ // Iterate over all servers to find one that actually appears in the
routing table for {@code getTableName()}.
+ // Picking the first server unconditionally can be flaky when multiple
servers exist with different tags
+ // (e.g. hybrid clusters with offline/realtime tenants).
+ String serverInstance = null;
+ for (String candidate : instances) {
+ if (!InstanceTypeUtils.isServer(candidate)) {
+ continue;
+ }
+ if (isInstanceInRoutingTable(candidate)) {
+ serverInstance = candidate;
+ break;
+ }
+ }
+ assertNotNull(serverInstance,
+ "No server instance found in routing table for table: " +
getTableName() + "; instances: " + instances);
+ HelixConfigScope scope =
+ new
HelixConfigScopeBuilder(HelixConfigScope.ConfigScopeProperty.PARTICIPANT,
getHelixClusterName())
+ .forParticipant(serverInstance).build();
+ try {
+ // Server should be in routing initially
+ checkForInstanceInRoutingTable(serverInstance, true);
+
+ // Disable query routing via the controller API
+
getOrCreateAdminClient().getInstanceClient().updateInstanceState(serverInstance,
"QUERIES_DISABLE");
+
+ // Server should leave routing
+ checkForInstanceInRoutingTable(serverInstance, false);
+
+ // Re-enable query routing
+
getOrCreateAdminClient().getInstanceClient().updateInstanceState(serverInstance,
"QUERIES_ENABLE");
+
+ // Server should re-enter routing
+ checkForInstanceInRoutingTable(serverInstance, true);
+
+ // Interleaving test: set queriesDisabled, then set shutdownInProgress,
then clear shutdownInProgress.
+ // queriesDisabled must still be set (i.e. the server must remain out of
routing) because we use
+ // field-scoped writes that do not clobber each other.
+
getOrCreateAdminClient().getInstanceClient().updateInstanceState(serverInstance,
"QUERIES_DISABLE");
+ checkForInstanceInRoutingTable(serverInstance, false);
+
+ _helixAdmin.setConfig(scope,
+
Collections.singletonMap(CommonConstants.Helix.IS_SHUTDOWN_IN_PROGRESS,
Boolean.TRUE.toString()));
+ checkForInstanceInRoutingTable(serverInstance, false);
+
+ _helixAdmin.setConfig(scope,
+
Collections.singletonMap(CommonConstants.Helix.IS_SHUTDOWN_IN_PROGRESS,
Boolean.FALSE.toString()));
+
+ // queriesDisabled was set via a separate field-scoped write, so the
server must still be out of routing
+ checkForInstanceInRoutingTable(serverInstance, false);
+
+ // Full-record updateInstance test: round-trip the live config through
PUT /instances/{name} with the
+ // request body's {@code queriesDisabled} defaulted to false. The
operational flag must NOT be cleared,
+ // since it is owned by /state?state=QUERIES_DISABLE|QUERIES_ENABLE.
+ InstanceConfig liveInstanceConfig =
_helixAdmin.getInstanceConfig(getHelixClusterName(), serverInstance);
+ Instance instanceBody = InstanceUtils.toInstance(liveInstanceConfig);
+ // Force the request-body flag to false to simulate an operator updating
host/tags/ports without realizing
+ // that queriesDisabled is operationally owned by a different API.
+ Instance updateBody = new Instance(instanceBody.getHost(),
instanceBody.getPort(), instanceBody.getType(),
+ instanceBody.getTags(), instanceBody.getPools(),
instanceBody.getGrpcPort(), instanceBody.getAdminPort(),
+ instanceBody.getQueryServicePort(),
instanceBody.getQueryMailboxPort(), false);
+
getOrCreateAdminClient().getInstanceClient().updateInstance(serverInstance,
updateBody.toJsonString());
+ // Operational flag must survive a request-body-driven update.
+ checkForInstanceInRoutingTable(serverInstance, false);
+ } finally {
+ // Defensive cleanup: restore the server to the routing-eligible state
regardless of which assertion failed.
+ // Each cleanup step is wrapped so a primary assertion failure isn't
masked by a cleanup throw.
+ try {
+
getOrCreateAdminClient().getInstanceClient().updateInstanceState(serverInstance,
"QUERIES_ENABLE");
+ } catch (Exception ignored) {
+ }
+ try {
+ _helixAdmin.setConfig(scope,
+
Collections.singletonMap(CommonConstants.Helix.IS_SHUTDOWN_IN_PROGRESS,
Boolean.FALSE.toString()));
+ } catch (Exception ignored) {
+ }
+ }
+ // Verify routing is restored after cleanup.
+ checkForInstanceInRoutingTable(serverInstance, true);
+ }
+
+ /// Non-waiting check for whether `instance` appears in any routing table
for the current table.
+ /// Used by [#testQueriesDisabled()] to pick a server that is actually
serving queries.
+ private boolean isInstanceInRoutingTable(String instance) {
+ try {
+ JsonNode routingTables = getDebugInfo("debug/routingTable/" +
getTableName());
+ for (JsonNode routingTable : routingTables) {
+ if (routingTable.has(instance)) {
+ return true;
+ }
+ }
+ } catch (Exception ignored) {
+ // Routing table not yet populated; treat as not present.
+ }
+ return false;
+ }
+
private void checkForInstanceInRoutingTable(String instance, boolean
shouldExist) {
String errorMessage;
if (shouldExist) {
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseRealtimeClusterIntegrationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseRealtimeClusterIntegrationTest.java
index aa6ff7e59c3..7c9abd13d19 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseRealtimeClusterIntegrationTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseRealtimeClusterIntegrationTest.java
@@ -202,6 +202,13 @@ public abstract class BaseRealtimeClusterIntegrationTest
extends BaseClusterInte
super.testInstanceShutdown();
}
+ @Test
+ @Override
+ public void testQueriesDisabled()
+ throws Exception {
+ super.testQueriesDisabled();
+ }
+
@AfterClass
public void tearDown()
throws Exception {
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/HybridClusterIntegrationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/HybridClusterIntegrationTest.java
index 22afb55d7f1..518326e73e1 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/HybridClusterIntegrationTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/HybridClusterIntegrationTest.java
@@ -377,6 +377,13 @@ public class HybridClusterIntegrationTest extends
BaseHybridClusterIntegrationTe
}
}
+ @Test
+ @Override
+ public void testQueriesDisabled()
+ throws Exception {
+ super.testQueriesDisabled();
+ }
+
@Test
void testControllerJoinQuerySubmit()
throws Exception {
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RealtimeConsumptionRateLimiterClusterIntegrationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RealtimeConsumptionRateLimiterClusterIntegrationTest.java
index 7473bcc75b1..8a46dcf1954 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RealtimeConsumptionRateLimiterClusterIntegrationTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RealtimeConsumptionRateLimiterClusterIntegrationTest.java
@@ -281,6 +281,13 @@ public class
RealtimeConsumptionRateLimiterClusterIntegrationTest extends BaseRe
// Do nothing
}
+ @Test(enabled = false)
+ @Override
+ public void testQueriesDisabled() {
+ // Routing-table-dependent test; the rate-limiter cluster setup keeps the
table out of broker routing,
+ // making this test inapplicable here. Mirrors the testInstanceShutdown
override above.
+ }
+
@Test(enabled = false)
public void testQueriesFromQueryFile(boolean useMultiStageQueryEngine) {
// Do nothing
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/realtime/ingestion/KafkaIncreaseDecreasePartitionsIntegrationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/realtime/ingestion/KafkaIncreaseDecreasePartitionsIntegrationTest.java
index 021eaca9718..a54ae80bdb4 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/realtime/ingestion/KafkaIncreaseDecreasePartitionsIntegrationTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/realtime/ingestion/KafkaIncreaseDecreasePartitionsIntegrationTest.java
@@ -87,6 +87,13 @@ public class KafkaIncreaseDecreasePartitionsIntegrationTest
extends BaseRealtime
// Do nothing
}
+ @Test(enabled = false)
+ @Override
+ public void testQueriesDisabled() {
+ // Routing-table-dependent test that does not apply to this
partition-change test setup.
+ // Mirrors the testInstanceShutdown override above.
+ }
+
@Test(enabled = false)
public void testQueriesFromQueryFile(boolean useMultiStageQueryEngine) {
// Do nothing
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
