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 5617ee70f4e Fix PinotAdminClient controller-contract bugs and retire
ControllerRequestClient (#18755)
5617ee70f4e is described below
commit 5617ee70f4e90a6e9daa6617a264172c37387651
Author: Xiang Fu <[email protected]>
AuthorDate: Sun Jun 14 13:52:31 2026 -0700
Fix PinotAdminClient controller-contract bugs and retire
ControllerRequestClient (#18755)
PinotAdminClient's admin sub-clients had diverged from the controller REST
contract, and the mock-based unit tests encoded the wrong response shapes,
so
the bugs stayed hidden. Align every call with the controller, and complete
the
ControllerRequestClient -> PinotAdminClient refactor (apache/pinot#15537).
Contract fixes (pinot-java-client):
- Parse the bare JSON arrays/scalars/maps the controller actually returns,
instead of reading non-existent wrapper fields: schema list, live
instances,
cancelRebalance job IDs, async tenant list, segment zk metadata, and every
task endpoint (types/state/tasks/count/states/subtasks).
- Correct query-param names: tenant rebalance include/excludeTables, task
counts "table", task debug "tableName", segment select start/endTimestamp,
and table rebalance downtime/minAvailableReplicas.
- Send the required request bodies: tenant rebalance (TenantRebalanceConfig)
and instance updateTags validation (List<InstanceTagUpdateRequest>).
- Consolidate string-array parsing into
PinotAdminTransport.parseStringArrayNode
(strict for synchronous callers, lenient wrapper for async).
Tests: correct the mocks that asserted the wrong shapes and add regression
tests covering each fixed endpoint plus the new parser edge cases.
Refactor: migrate the last ControllerRequestClient usage
(BaseMultiClusterIntegrationTest) to the existing addSchemaToCluster helper,
delete the ControllerRequestClient class, and rename
getControllerRequestClientHeaders -> getAdminClientHeaders.
---
.../pinot/client/admin/BaseServiceAdminClient.java | 19 +
.../pinot/client/admin/InstanceAdminClient.java | 45 +-
.../pinot/client/admin/PinotAdminTransport.java | 39 +-
.../pinot/client/admin/SchemaAdminClient.java | 5 +-
.../pinot/client/admin/SegmentAdminClient.java | 8 +-
.../pinot/client/admin/TableAdminClient.java | 34 +-
.../apache/pinot/client/admin/TaskAdminClient.java | 34 +-
.../pinot/client/admin/TenantAdminClient.java | 55 +-
.../pinot/client/admin/PinotAdminClientTest.java | 204 +++++++-
.../controller/helix/ControllerRequestClient.java | 551 ---------------------
.../PinotAdminUserLogicalTableResourceTest.java | 2 +-
...inotUserWithAccessLogicalTableResourceTest.java | 2 +-
.../pinot/controller/helix/ControllerTest.java | 18 +-
.../tests/CursorWithAuthIntegrationTest.java | 2 +-
.../tests/RowLevelSecurityIntegrationTest.java | 2 +-
.../tests/TimeSeriesAuthIntegrationTest.java | 2 +-
.../integration/tests/TlsIntegrationTest.java | 2 +-
.../tests/UrlAuthRealtimeIntegrationTest.java | 2 +-
.../BaseMultiClusterIntegrationTest.java | 5 +-
19 files changed, 376 insertions(+), 655 deletions(-)
diff --git
a/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/BaseServiceAdminClient.java
b/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/BaseServiceAdminClient.java
index f81b773275b..d2c1140062f 100644
---
a/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/BaseServiceAdminClient.java
+++
b/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/BaseServiceAdminClient.java
@@ -18,7 +18,10 @@
*/
package org.apache.pinot.client.admin;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import javax.annotation.Nullable;
@@ -51,4 +54,20 @@ abstract class BaseServiceAdminClient {
}
return merged;
}
+
+ /// Lenient variant of [PinotAdminTransport#parseStringArrayNode(JsonNode)]
for async / best-effort paths:
+ /// returns an empty list instead of throwing when the node is not a string
array.
+ ///
+ /// Synchronous callers should use
[PinotAdminTransport#parseStringArrayNode(JsonNode)] directly so that an
+ /// unexpected response shape surfaces as a [PinotAdminException] rather
than an empty result.
+ ///
+ /// @param node the response node (expected to be a bare JSON array, or a
comma-separated string)
+ /// @return the parsed list, or an empty list when `node` is not a string
array
+ protected static List<String> toStringListSafe(@Nullable JsonNode node) {
+ try {
+ return PinotAdminTransport.parseStringArrayNode(node);
+ } catch (PinotAdminException e) {
+ return Collections.emptyList();
+ }
+ }
}
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 05131a22c8e..1f8c255181f 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
@@ -19,6 +19,7 @@
package org.apache.pinot.client.admin;
import com.fasterxml.jackson.databind.JsonNode;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -60,7 +61,8 @@ public class InstanceAdminClient extends
BaseServiceAdminClient {
public List<String> listLiveInstances()
throws PinotAdminException {
JsonNode response = _transport.executeGet(_controllerAddress,
"/liveinstances", null, _headers);
- return _transport.parseStringArray(response, "liveInstances");
+ /// GET /liveinstances returns an Instances wrapper serialized as
{"instances": [...]}.
+ return _transport.parseStringArray(response, "instances");
}
/**
@@ -245,22 +247,39 @@ public class InstanceAdminClient extends
BaseServiceAdminClient {
return response.toString();
}
- /**
- * Validates whether it's safe to update the tags of the given instances.
- *
- * @param instanceNames Comma-separated list of instance names to validate
- * @param newTags New tags to assign
- * @return Validation response as JSON string
- * @throws PinotAdminException If the request fails
- */
+ /// Validates whether it's safe to update the tags of the given instances.
+ ///
+ /// The controller endpoint reads a `List<InstanceTagUpdateRequest>` from
the request body (not query
+ /// params). This method builds one request per instance name, all sharing
the given tags. Use
+ /// [#validateInstanceTagUpdates(String)] when you need to supply
per-instance tags.
+ ///
+ /// @param instanceNames Comma-separated list of instance names to validate
+ /// @param newTags Comma-separated list of tags to assign to every listed
instance
+ /// @return Validation response as JSON string
+ /// @throws PinotAdminException If the request fails
public String validateUpdateInstanceTags(String instanceNames, String
newTags)
throws PinotAdminException {
- Map<String, String> queryParams = new HashMap<>();
- queryParams.put("instanceNames", instanceNames);
- queryParams.put("newTags", newTags);
+ List<String> tags = new ArrayList<>();
+ for (String tag : newTags.split(",")) {
+ String trimmedTag = tag.trim();
+ if (!trimmedTag.isEmpty()) {
+ tags.add(trimmedTag);
+ }
+ }
+ List<Map<String, Object>> requestBody = new ArrayList<>();
+ for (String instanceName : instanceNames.split(",")) {
+ String trimmedName = instanceName.trim();
+ if (trimmedName.isEmpty()) {
+ continue;
+ }
+ Map<String, Object> request = new HashMap<>();
+ request.put("instanceName", trimmedName);
+ request.put("newTags", tags);
+ requestBody.add(request);
+ }
JsonNode response = _transport.executePost(_controllerAddress,
"/instances/updateTags/validate",
- null, queryParams, _headers);
+ requestBody, null, _headers);
return response.toString();
}
diff --git
a/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/PinotAdminTransport.java
b/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/PinotAdminTransport.java
index 6571a70e37c..184f42b0e3c 100644
---
a/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/PinotAdminTransport.java
+++
b/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/PinotAdminTransport.java
@@ -26,7 +26,6 @@ import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -443,24 +442,48 @@ public class PinotAdminTransport implements AutoCloseable
{
if (arrayNode == null) {
throw new PinotAdminException("Response missing '" + fieldName + "'
field");
}
+ return parseStringArrayNode(arrayNode);
+ }
+ /// Parses a JSON node that is itself a string array (or a comma-separated
textual value) into a list of strings.
+ ///
+ /// This is the single implementation shared by [#parseStringArray(JsonNode,
String)] (which first extracts a
+ /// named field) and the admin clients that consume controller endpoints
returning a bare JSON array (for example
+ /// `GET /schemas` and `DELETE /tables/{tableName}/rebalance`).
+ ///
+ /// @param arrayNode the node expected to be a JSON array (or a
comma-separated string)
+ /// @return the parsed list of strings
+ /// @throws PinotAdminException if `arrayNode` is `null` or is neither an
array nor a string
+ static List<String> parseStringArrayNode(JsonNode arrayNode)
+ throws PinotAdminException {
+ if (arrayNode == null || arrayNode.isNull()) {
+ throw new PinotAdminException("Expected a JSON array but got a null
node");
+ }
if (arrayNode.isArray()) {
- // Handle JSON array format
- List<String> result = new ArrayList<>();
+ /// Handle JSON array format
+ List<String> result = new ArrayList<>(arrayNode.size());
for (JsonNode element : arrayNode) {
result.add(element.asText());
}
return result;
- } else if (arrayNode.isTextual()) {
- // Handle comma-separated string format for backward compatibility
+ }
+ if (arrayNode.isTextual()) {
+ /// Handle comma-separated string format for backward compatibility:
trim tokens and drop empty entries
+ /// so the result matches the JSON-array path.
String text = arrayNode.asText().trim();
if (text.isEmpty()) {
return Collections.emptyList();
}
- return Arrays.asList(text.split(","));
- } else {
- throw new PinotAdminException("Field '" + fieldName + "' is not an array
or string: " + arrayNode.getNodeType());
+ List<String> result = new ArrayList<>();
+ for (String token : text.split(",")) {
+ String trimmed = token.trim();
+ if (!trimmed.isEmpty()) {
+ result.add(trimmed);
+ }
+ }
+ return result;
}
+ throw new PinotAdminException("Expected a JSON array or string but got: "
+ arrayNode.getNodeType());
}
/**
diff --git
a/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/SchemaAdminClient.java
b/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/SchemaAdminClient.java
index f5d06442d93..1c77cb2bfcf 100644
---
a/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/SchemaAdminClient.java
+++
b/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/SchemaAdminClient.java
@@ -53,8 +53,9 @@ public class SchemaAdminClient extends BaseServiceAdminClient
{
*/
public List<String> listSchemaNames()
throws PinotAdminException {
+ /// GET /schemas returns a bare JSON array of schema names, not a wrapper
object.
JsonNode response = _transport.executeGet(_controllerAddress, "/schemas",
null, _headers);
- return _transport.parseStringArray(response, "schemas");
+ return PinotAdminTransport.parseStringArrayNode(response);
}
/**
@@ -243,7 +244,7 @@ public class SchemaAdminClient extends
BaseServiceAdminClient {
*/
public CompletableFuture<List<String>> listSchemaNamesAsync() {
return _transport.executeGetAsync(_controllerAddress, "/schemas", null,
_headers)
- .thenApply(response -> _transport.parseStringArraySafe(response,
"schemas"));
+ .thenApply(BaseServiceAdminClient::toStringListSafe);
}
/**
diff --git
a/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/SegmentAdminClient.java
b/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/SegmentAdminClient.java
index ab841c487b7..ab87df8ff6b 100644
---
a/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/SegmentAdminClient.java
+++
b/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/SegmentAdminClient.java
@@ -479,8 +479,9 @@ public class SegmentAdminClient extends
BaseServiceAdminClient {
long endTimestampMs, boolean excludeReplacedSegments)
throws PinotAdminException {
Map<String, String> queryParams = new HashMap<>();
- queryParams.put("startTimestampMs", String.valueOf(startTimestampMs));
- queryParams.put("endTimestampMs", String.valueOf(endTimestampMs));
+ /// Controller reads startTimestamp/endTimestamp (ms); see
PinotSegmentRestletResource#getSelectedSegments.
+ queryParams.put("startTimestamp", String.valueOf(startTimestampMs));
+ queryParams.put("endTimestamp", String.valueOf(endTimestampMs));
queryParams.put("excludeReplacedSegments",
String.valueOf(excludeReplacedSegments));
if (tableType != null) {
queryParams.put("type", tableType);
@@ -628,7 +629,8 @@ public class SegmentAdminClient extends
BaseServiceAdminClient {
throws PinotAdminException {
JsonNode response = _transport.executeGet(_controllerAddress, "/segments/"
+ tableName + "/zkmetadata",
null, _headers);
- return
PinotAdminTransport.getObjectMapper().convertValue(response.get("zkMetadata"),
+ /// GET /segments/{tableName}/zkmetadata returns a bare map of segment
name -> metadata, not a wrapper object.
+ return PinotAdminTransport.getObjectMapper().convertValue(response,
new TypeReference<Map<String, Map<String, String>>>() {
});
}
diff --git
a/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/TableAdminClient.java
b/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/TableAdminClient.java
index e34cbe3d7f7..502282458f3 100644
---
a/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/TableAdminClient.java
+++
b/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/TableAdminClient.java
@@ -378,26 +378,25 @@ public class TableAdminClient extends
BaseServiceAdminClient {
return response.toString();
}
- /**
- * Rebalances a table (reassigns instances and segments).
- *
- * @param tableName Name of the table to rebalance
- * @param noDowntime Whether to allow rebalance without downtime
- * @param rebalanceMode Rebalance mode (default or specific)
- * @param minReplicasToKeepAfterRebalance Minimum replicas to keep after
rebalance
- * @return Rebalance result
- * @throws PinotAdminException If the request fails
- */
- public String rebalanceTable(String tableName, boolean noDowntime, @Nullable
String rebalanceMode,
+ /// Rebalances a table (reassigns instances and segments).
+ ///
+ /// Maps to the controller's `downtime` and `minAvailableReplicas` query
params (see
+ /// [#rebalanceTable(String, String, boolean, boolean, boolean, boolean,
int)] for the full parameter set, or
+ /// [#rebalanceTable(String, Map)] to pass arbitrary params).
+ ///
+ /// @param tableName Name of the table to rebalance
+ /// @param noDowntime Whether the rebalance should avoid downtime (maps to
`downtime=false`)
+ /// @param minReplicasToKeepAfterRebalance Minimum number of replicas to
keep available during a no-downtime
+ /// rebalance (maps to
`minAvailableReplicas`)
+ /// @return Rebalance result
+ /// @throws PinotAdminException If the request fails
+ public String rebalanceTable(String tableName, boolean noDowntime,
@Nullable Integer minReplicasToKeepAfterRebalance)
throws PinotAdminException {
Map<String, String> queryParams = new HashMap<>();
- queryParams.put("noDowntime", String.valueOf(noDowntime));
- if (rebalanceMode != null) {
- queryParams.put("rebalanceMode", rebalanceMode);
- }
+ queryParams.put("downtime", String.valueOf(!noDowntime));
if (minReplicasToKeepAfterRebalance != null) {
- queryParams.put("minReplicasToKeepAfterRebalance",
String.valueOf(minReplicasToKeepAfterRebalance));
+ queryParams.put("minAvailableReplicas",
String.valueOf(minReplicasToKeepAfterRebalance));
}
JsonNode response = _transport.executePost(_controllerAddress, "/tables/"
+ tableName + "/rebalance", null,
@@ -432,7 +431,8 @@ public class TableAdminClient extends
BaseServiceAdminClient {
throws PinotAdminException {
JsonNode response =
_transport.executeDelete(_controllerAddress, "/tables/" + tableName +
"/rebalance", null, _headers);
- return _transport.parseStringArray(response, "jobIds");
+ /// DELETE /tables/{tableName}/rebalance returns a bare JSON array of
cancelled job IDs.
+ return PinotAdminTransport.parseStringArrayNode(response);
}
/**
diff --git
a/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/TaskAdminClient.java
b/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/TaskAdminClient.java
index ee6e53d2f98..e2ac609dc1b 100644
---
a/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/TaskAdminClient.java
+++
b/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/TaskAdminClient.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.client.admin;
+import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonNode;
import java.util.HashMap;
import java.util.Map;
@@ -50,7 +51,7 @@ public class TaskAdminClient extends BaseServiceAdminClient {
public Set<String> listTaskTypes()
throws PinotAdminException {
JsonNode response = _transport.executeGet(_controllerAddress,
"/tasks/tasktypes", null, _headers);
- return
PinotAdminTransport.getObjectMapper().convertValue(response.get("taskTypes"),
Set.class);
+ return PinotAdminTransport.getObjectMapper().convertValue(response,
Set.class);
}
/**
@@ -63,7 +64,7 @@ public class TaskAdminClient extends BaseServiceAdminClient {
public TaskState getTaskQueueState(String taskType)
throws PinotAdminException {
JsonNode response = _transport.executeGet(_controllerAddress, "/tasks/" +
taskType + "/state", null, _headers);
- return TaskState.valueOf(response.get("state").asText());
+ return TaskState.valueOf(response.asText());
}
/**
@@ -76,7 +77,7 @@ public class TaskAdminClient extends BaseServiceAdminClient {
public Set<String> getTasks(String taskType)
throws PinotAdminException {
JsonNode response = _transport.executeGet(_controllerAddress, "/tasks/" +
taskType + "/tasks", null, _headers);
- return
PinotAdminTransport.getObjectMapper().convertValue(response.get("tasks"),
Set.class);
+ return PinotAdminTransport.getObjectMapper().convertValue(response,
Set.class);
}
/**
@@ -90,7 +91,7 @@ public class TaskAdminClient extends BaseServiceAdminClient {
throws PinotAdminException {
JsonNode response =
_transport.executeGet(_controllerAddress, "/tasks/" + taskType +
"/tasks/count", null, _headers);
- return response.get("count").asInt();
+ return response.asInt();
}
/**
@@ -106,7 +107,8 @@ public class TaskAdminClient extends BaseServiceAdminClient
{
JsonNode response =
_transport.executeGet(_controllerAddress, "/tasks/" + taskType + "/" +
tableNameWithType + "/state",
null, _headers);
- return
PinotAdminTransport.getObjectMapper().convertValue(response.get("taskStates"),
Map.class);
+ return PinotAdminTransport.getObjectMapper().convertValue(response, new
TypeReference<Map<String, TaskState>>() {
+ });
}
/**
@@ -159,7 +161,8 @@ public class TaskAdminClient extends BaseServiceAdminClient
{
queryParams.put("state", state);
}
if (tableNameWithType != null) {
- queryParams.put("tableNameWithType", tableNameWithType);
+ /// Controller reads the table filter from the "table" query param (see
PinotTaskRestletResource#getTaskCounts).
+ queryParams.put("table", tableNameWithType);
}
if (minNumSubtasks != null) {
queryParams.put("minNumSubtasks", String.valueOf(minNumSubtasks));
@@ -239,7 +242,9 @@ public class TaskAdminClient extends BaseServiceAdminClient
{
Map<String, String> queryParams = new HashMap<>();
queryParams.put("verbosity", String.valueOf(verbosity));
if (tableNameWithType != null) {
- queryParams.put("tableNameWithType", tableNameWithType);
+ /// Controller reads the table filter from the "tableName" query param
(see
+ /// PinotTaskRestletResource#getTaskDebugInfo).
+ queryParams.put("tableName", tableNameWithType);
}
JsonNode response = _transport.executeGet(_controllerAddress,
"/tasks/task/" + taskName + "/debug",
@@ -257,7 +262,8 @@ public class TaskAdminClient extends BaseServiceAdminClient
{
public Map<String, TaskState> getTaskStates(String taskType)
throws PinotAdminException {
JsonNode response = _transport.executeGet(_controllerAddress, "/tasks/" +
taskType + "/taskstates", null, _headers);
- return
PinotAdminTransport.getObjectMapper().convertValue(response.get("taskStates"),
Map.class);
+ return PinotAdminTransport.getObjectMapper().convertValue(response, new
TypeReference<Map<String, TaskState>>() {
+ });
}
/**
@@ -270,7 +276,7 @@ public class TaskAdminClient extends BaseServiceAdminClient
{
public TaskState getTaskState(String taskName)
throws PinotAdminException {
JsonNode response = _transport.executeGet(_controllerAddress,
"/tasks/task/" + taskName + "/state", null, _headers);
- return TaskState.valueOf(response.get("state").asText());
+ return TaskState.valueOf(response.asText());
}
/**
@@ -284,7 +290,7 @@ public class TaskAdminClient extends BaseServiceAdminClient
{
throws PinotAdminException {
JsonNode response =
_transport.executeGet(_controllerAddress, "/tasks/subtask/" + taskName
+ "/state", null, _headers);
- return
PinotAdminTransport.getObjectMapper().convertValue(response.get("subtaskStates"),
Map.class);
+ return PinotAdminTransport.getObjectMapper().convertValue(response,
Map.class);
}
/**
@@ -329,7 +335,7 @@ public class TaskAdminClient extends BaseServiceAdminClient
{
public CompletableFuture<Set<String>> listTaskTypesAsync() {
return _transport.executeGetAsync(_controllerAddress, "/tasks/tasktypes",
null, _headers)
.thenApply(response -> PinotAdminTransport.getObjectMapper()
- .convertValue(response.get("taskTypes"), Set.class));
+ .convertValue(response, Set.class));
}
/**
@@ -337,7 +343,7 @@ public class TaskAdminClient extends BaseServiceAdminClient
{
*/
public CompletableFuture<TaskState> getTaskQueueStateAsync(String taskType) {
return _transport.executeGetAsync(_controllerAddress, "/tasks/" + taskType
+ "/state", null, _headers)
- .thenApply(response ->
TaskState.valueOf(response.get("state").asText()));
+ .thenApply(response -> TaskState.valueOf(response.asText()));
}
/**
@@ -345,7 +351,7 @@ public class TaskAdminClient extends BaseServiceAdminClient
{
*/
public CompletableFuture<Set<String>> getTasksAsync(String taskType) {
return _transport.executeGetAsync(_controllerAddress, "/tasks/" + taskType
+ "/tasks", null, _headers)
- .thenApply(response ->
PinotAdminTransport.getObjectMapper().convertValue(response.get("tasks"),
Set.class));
+ .thenApply(response ->
PinotAdminTransport.getObjectMapper().convertValue(response, Set.class));
}
/**
@@ -353,6 +359,6 @@ public class TaskAdminClient extends BaseServiceAdminClient
{
*/
public CompletableFuture<TaskState> getTaskStateAsync(String taskName) {
return _transport.executeGetAsync(_controllerAddress, "/tasks/task/" +
taskName + "/state", null, _headers)
- .thenApply(response ->
TaskState.valueOf(response.get("state").asText()));
+ .thenApply(response -> TaskState.valueOf(response.asText()));
}
}
diff --git
a/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/TenantAdminClient.java
b/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/TenantAdminClient.java
index 8e6b67763b4..a455cea435d 100644
---
a/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/TenantAdminClient.java
+++
b/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/TenantAdminClient.java
@@ -22,7 +22,7 @@ import com.fasterxml.jackson.databind.JsonNode;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
-import java.util.HashSet;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -71,7 +71,15 @@ public class TenantAdminClient extends
BaseServiceAdminClient {
JsonNode response =
_transport.executeGet(_controllerAddress, "/tenants",
queryParams.isEmpty() ? null : queryParams, _headers);
- Set<String> tenants = new HashSet<>();
+ return parseTenantNames(response);
+ }
+
+ /// Extracts the broker and server tenant names from a `GET /tenants`
response.
+ ///
+ /// The controller returns a `TenantsList` serialized as `{"SERVER_TENANTS":
[...], "BROKER_TENANTS": [...]}`;
+ /// there is no flat `tenants` field.
+ private static List<String> parseTenantNames(JsonNode response) {
+ Set<String> tenants = new LinkedHashSet<>();
JsonNode brokerTenants = response.get("BROKER_TENANTS");
if (brokerTenants != null && brokerTenants.isArray()) {
brokerTenants.forEach(node -> tenants.add(node.asText()));
@@ -317,34 +325,33 @@ public class TenantAdminClient extends
BaseServiceAdminClient {
return response.toString();
}
- /**
- * Rebalances all tables that are part of the tenant.
- *
- * @param tenantName Name of the tenant
- * @param degreeOfParallelism Number of table rebalance jobs allowed to run
at the same time
- * @param includeTableTypes Comma-separated list of table types to include
(optional)
- * @param excludeTableTypes Comma-separated list of table types to exclude
(optional)
- * @param rebalanceMode Rebalance mode (optional)
- * @return Rebalance result
- * @throws PinotAdminException If the request fails
- */
- public String rebalanceTenant(String tenantName, int degreeOfParallelism,
@Nullable String includeTableTypes,
- @Nullable String excludeTableTypes, @Nullable String rebalanceMode)
+ /// Rebalances all tables that are part of the tenant.
+ ///
+ /// The controller endpoint reads the include/exclude filters from the
`includeTables`/`excludeTables`
+ /// query params and requires a `TenantRebalanceConfig` request body; this
method sends an empty config object
+ /// so the query params can populate it. Use
[#rebalanceTenantWithConfig(String, String, Map)] when you need to
+ /// supply a full rebalance config.
+ ///
+ /// @param tenantName Name of the tenant
+ /// @param degreeOfParallelism Number of table rebalance jobs allowed to run
at the same time
+ /// @param includeTables Comma-separated list of tables (with type) to
include (optional)
+ /// @param excludeTables Comma-separated list of tables (with type) to
exclude (optional)
+ /// @return Rebalance result
+ /// @throws PinotAdminException If the request fails
+ public String rebalanceTenant(String tenantName, int degreeOfParallelism,
@Nullable String includeTables,
+ @Nullable String excludeTables)
throws PinotAdminException {
Map<String, String> queryParams = new HashMap<>();
queryParams.put("degreeOfParallelism",
String.valueOf(degreeOfParallelism));
- if (includeTableTypes != null) {
- queryParams.put("includeTableTypes", includeTableTypes);
+ if (includeTables != null) {
+ queryParams.put("includeTables", includeTables);
}
- if (excludeTableTypes != null) {
- queryParams.put("excludeTableTypes", excludeTableTypes);
- }
- if (rebalanceMode != null) {
- queryParams.put("rebalanceMode", rebalanceMode);
+ if (excludeTables != null) {
+ queryParams.put("excludeTables", excludeTables);
}
JsonNode response = _transport.executePost(_controllerAddress, "/tenants/"
+ tenantName + "/rebalance",
- null, queryParams, _headers);
+ "{}", queryParams, _headers);
return response.toString();
}
@@ -431,7 +438,7 @@ public class TenantAdminClient extends
BaseServiceAdminClient {
*/
public CompletableFuture<List<String>> listTenantsAsync() {
return _transport.executeGetAsync(_controllerAddress, "/tenants", null,
_headers)
- .thenApply(response ->
PinotAdminTransport.getObjectMapper().convertValue(response.get("tenants"),
List.class));
+ .thenApply(TenantAdminClient::parseTenantNames);
}
/**
diff --git
a/pinot-clients/pinot-java-client/src/test/java/org/apache/pinot/client/admin/PinotAdminClientTest.java
b/pinot-clients/pinot-java-client/src/test/java/org/apache/pinot/client/admin/PinotAdminClientTest.java
index c6a9a820a01..69a9487587a 100644
---
a/pinot-clients/pinot-java-client/src/test/java/org/apache/pinot/client/admin/PinotAdminClientTest.java
+++
b/pinot-clients/pinot-java-client/src/test/java/org/apache/pinot/client/admin/PinotAdminClientTest.java
@@ -23,7 +23,9 @@ import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import org.apache.helix.task.TaskState;
import org.apache.pinot.common.restlet.resources.PauseStatusDetails;
import
org.apache.pinot.common.restlet.resources.ServerRebalanceJobStatusResponse;
import org.apache.pinot.common.restlet.resources.TableView;
@@ -159,7 +161,8 @@ public class PinotAdminClientTest {
@Test
public void testListSchemas()
throws Exception {
- String jsonResponse = "{\"schemas\": [\"sch1\", \"sch2\"]}";
+ /// GET /schemas returns a bare JSON array of schema names (not a
{"schemas": [...]} wrapper).
+ String jsonResponse = "[\"sch1\", \"sch2\"]";
JsonNode mockResponse = new ObjectMapper().readTree(jsonResponse);
lenient().when(_mockTransport.executeGet(anyString(), anyString(), any(),
any()))
.thenReturn(mockResponse);
@@ -188,7 +191,7 @@ public class PinotAdminClientTest {
@Test
public void testAsyncListSchemas()
throws Exception {
- String jsonResponse = "{\"schemas\": [\"sch1\"]}";
+ String jsonResponse = "[\"sch1\"]";
JsonNode mockResponse = new ObjectMapper().readTree(jsonResponse);
CompletableFuture<JsonNode> jsonNodeCompletableFuture =
CompletableFuture.completedFuture(mockResponse);
lenient().when(_mockTransport.executeGetAsync(anyString(), anyString(),
any(), any()))
@@ -432,4 +435,201 @@ public class PinotAdminClientTest {
verify(_mockTransport).executeDelete(eq(CONTROLLER_ADDRESS),
eq("/clientQuery/client-query-1"), isNull(),
eq(HEADERS));
}
+
+ @Test
+ public void testListLiveInstancesUsesInstancesField()
+ throws Exception {
+ /// GET /liveinstances returns an Instances wrapper serialized as
{"instances": [...]}.
+ JsonNode mockResponse = new ObjectMapper().readTree("{\"instances\":
[\"Server_1\", \"Broker_1\"]}");
+ lenient().when(_mockTransport.executeGet(anyString(), anyString(), any(),
any())).thenReturn(mockResponse);
+
+ List<String> instances =
_adminClient.getInstanceClient().listLiveInstances();
+
+ assertEquals(instances, List.of("Server_1", "Broker_1"));
+ verify(_mockTransport).executeGet(eq(CONTROLLER_ADDRESS),
eq("/liveinstances"), isNull(), eq(HEADERS));
+ }
+
+ @Test
+ public void testValidateUpdateInstanceTagsSendsRequestBody()
+ throws Exception {
+ JsonNode mockResponse = new ObjectMapper().readTree("[]");
+ lenient().when(_mockTransport.executePost(anyString(), anyString(), any(),
any(), any()))
+ .thenReturn(mockResponse);
+
+
_adminClient.getInstanceClient().validateUpdateInstanceTags("Server_1,Server_2",
"tag1,tag2");
+
+ /// Controller reads a List<InstanceTagUpdateRequest> from the body; no
query params are used.
+ List<Map<String, Object>> expectedBody = List.of(
+ Map.of("instanceName", "Server_1", "newTags", List.of("tag1", "tag2")),
+ Map.of("instanceName", "Server_2", "newTags", List.of("tag1",
"tag2")));
+ verify(_mockTransport).executePost(eq(CONTROLLER_ADDRESS),
eq("/instances/updateTags/validate"), eq(expectedBody),
+ isNull(), eq(HEADERS));
+ }
+
+ @Test
+ public void testCancelRebalanceParsesBareArray()
+ throws Exception {
+ /// DELETE /tables/{tableName}/rebalance returns a bare JSON array of
cancelled job IDs.
+ JsonNode mockResponse = new ObjectMapper().readTree("[\"job-1\",
\"job-2\"]");
+ lenient().when(_mockTransport.executeDelete(anyString(), anyString(),
any(), any())).thenReturn(mockResponse);
+
+ List<String> jobIds =
_adminClient.getTableClient().cancelRebalance("tbl1_OFFLINE");
+
+ assertEquals(jobIds, List.of("job-1", "job-2"));
+ verify(_mockTransport).executeDelete(eq(CONTROLLER_ADDRESS),
eq("/tables/tbl1_OFFLINE/rebalance"), isNull(),
+ eq(HEADERS));
+ }
+
+ @Test
+ public void testRebalanceTableMapsDowntimeAndMinAvailableReplicas()
+ throws Exception {
+ JsonNode mockResponse = new ObjectMapper().readTree("{\"status\":\"OK\"}");
+ lenient().when(_mockTransport.executePost(anyString(), anyString(), any(),
any(), any()))
+ .thenReturn(mockResponse);
+
+ _adminClient.getTableClient().rebalanceTable("tbl1_OFFLINE", true, 2);
+
+ verify(_mockTransport).executePost(eq(CONTROLLER_ADDRESS),
eq("/tables/tbl1_OFFLINE/rebalance"), isNull(),
+ eq(Map.of("downtime", "false", "minAvailableReplicas", "2")),
eq(HEADERS));
+ }
+
+ @Test
+ public void testSelectSegmentsUsesStartEndTimestampParams()
+ throws Exception {
+ JsonNode mockResponse = new ObjectMapper().readTree("[]");
+ lenient().when(_mockTransport.executeGet(anyString(), anyString(), any(),
any())).thenReturn(mockResponse);
+
+ _adminClient.getSegmentClient().selectSegments("tbl1", "OFFLINE", 100L,
200L, true);
+
+ verify(_mockTransport).executeGet(eq(CONTROLLER_ADDRESS),
eq("/segments/tbl1/select"),
+ eq(Map.of("startTimestamp", "100", "endTimestamp", "200",
"excludeReplacedSegments", "true", "type",
+ "OFFLINE")), eq(HEADERS));
+ }
+
+ @Test
+ public void testGetZookeeperMetadataParsesBareMap()
+ throws Exception {
+ JsonNode mockResponse = new ObjectMapper()
+
.readTree("{\"segment_1\":{\"segment.crc\":\"123\"},\"segment_2\":{\"segment.crc\":\"456\"}}");
+ lenient().when(_mockTransport.executeGet(anyString(), anyString(), any(),
any())).thenReturn(mockResponse);
+
+ Map<String, Map<String, String>> zkMetadata =
_adminClient.getSegmentClient().getZookeeperMetadata("tbl1_OFFLINE");
+
+ assertEquals(zkMetadata.size(), 2);
+ assertEquals(zkMetadata.get("segment_1").get("segment.crc"), "123");
+ verify(_mockTransport).executeGet(eq(CONTROLLER_ADDRESS),
eq("/segments/tbl1_OFFLINE/zkmetadata"), isNull(),
+ eq(HEADERS));
+ }
+
+ @Test
+ public void testTaskEndpointsParseBareValues()
+ throws Exception {
+ ObjectMapper mapper = new ObjectMapper();
+ /// Every task endpoint resumes a bare value (array/string/int/map), not a
wrapper object.
+ lenient().when(_mockTransport.executeGet(eq(CONTROLLER_ADDRESS),
eq("/tasks/tasktypes"), any(), any()))
+ .thenReturn(mapper.readTree("[\"TaskA\", \"TaskB\"]"));
+ lenient().when(_mockTransport.executeGet(eq(CONTROLLER_ADDRESS),
eq("/tasks/TaskA/state"), any(), any()))
+ .thenReturn(mapper.readTree("\"IN_PROGRESS\""));
+ lenient().when(_mockTransport.executeGet(eq(CONTROLLER_ADDRESS),
eq("/tasks/TaskA/tasks"), any(), any()))
+ .thenReturn(mapper.readTree("[\"Task_TaskA_1\"]"));
+ lenient().when(_mockTransport.executeGet(eq(CONTROLLER_ADDRESS),
eq("/tasks/TaskA/tasks/count"), any(), any()))
+ .thenReturn(mapper.readTree("5"));
+ lenient().when(_mockTransport.executeGet(eq(CONTROLLER_ADDRESS),
eq("/tasks/TaskA/taskstates"), any(), any()))
+ .thenReturn(mapper.readTree("{\"Task_TaskA_1\":\"COMPLETED\"}"));
+ lenient().when(_mockTransport.executeGet(eq(CONTROLLER_ADDRESS),
eq("/tasks/TaskA/tbl1_OFFLINE/state"), any(),
+
any())).thenReturn(mapper.readTree("{\"Task_TaskA_1\":\"COMPLETED\"}"));
+ lenient().when(_mockTransport.executeGet(eq(CONTROLLER_ADDRESS),
eq("/tasks/subtask/Task_TaskA_1/state"), any(),
+
any())).thenReturn(mapper.readTree("{\"Task_TaskA_1_0\":\"RUNNING\"}"));
+
+ TaskAdminClient taskClient = _adminClient.getTaskClient();
+ assertEquals(taskClient.listTaskTypes(), Set.of("TaskA", "TaskB"));
+ assertEquals(taskClient.getTaskQueueState("TaskA"), TaskState.IN_PROGRESS);
+ assertEquals(taskClient.getTasks("TaskA"), Set.of("Task_TaskA_1"));
+ assertEquals(taskClient.getTasksCount("TaskA"), 5);
+ assertEquals(taskClient.getTaskStates("TaskA"), Map.of("Task_TaskA_1",
TaskState.COMPLETED));
+ assertEquals(taskClient.getTaskStatesByTable("TaskA", "tbl1_OFFLINE"),
Map.of("Task_TaskA_1", TaskState.COMPLETED));
+ assertEquals(taskClient.getSubtaskStates("Task_TaskA_1"),
Map.of("Task_TaskA_1_0", "RUNNING"));
+ }
+
+ @Test
+ public void testGetTaskCountsUsesTableParam()
+ throws Exception {
+ JsonNode mockResponse = new ObjectMapper().readTree("{}");
+ lenient().when(_mockTransport.executeGet(anyString(), anyString(), any(),
any())).thenReturn(mockResponse);
+
+ _adminClient.getTaskClient().getTaskCounts("TaskA", "IN_PROGRESS",
"tbl1_OFFLINE", null);
+
+ verify(_mockTransport).executeGet(eq(CONTROLLER_ADDRESS),
eq("/tasks/TaskA/taskcounts"),
+ eq(Map.of("state", "IN_PROGRESS", "table", "tbl1_OFFLINE")),
eq(HEADERS));
+ }
+
+ @Test
+ public void testGetTaskDebugInfoUsesTableNameParam()
+ throws Exception {
+ JsonNode mockResponse = new ObjectMapper().readTree("{}");
+ lenient().when(_mockTransport.executeGet(anyString(), anyString(), any(),
any())).thenReturn(mockResponse);
+
+ _adminClient.getTaskClient().getTaskDebugInfo("Task_TaskA_1", 1,
"tbl1_OFFLINE");
+
+ verify(_mockTransport).executeGet(eq(CONTROLLER_ADDRESS),
eq("/tasks/task/Task_TaskA_1/debug"),
+ eq(Map.of("verbosity", "1", "tableName", "tbl1_OFFLINE")),
eq(HEADERS));
+ }
+
+ @Test
+ public void testListTenantsAsyncParsesTenantArrays()
+ throws Exception {
+ /// GET /tenants returns {"SERVER_TENANTS": [...], "BROKER_TENANTS":
[...]} -- there is no flat "tenants" field.
+ JsonNode mockResponse = new ObjectMapper().readTree(
+
"{\"SERVER_TENANTS\":[\"DefaultTenant\"],\"BROKER_TENANTS\":[\"DefaultTenant\",\"brokerTenant\"]}");
+ lenient().when(_mockTransport.executeGetAsync(anyString(), anyString(),
any(), any()))
+ .thenReturn(CompletableFuture.completedFuture(mockResponse));
+
+ List<String> tenants =
_adminClient.getTenantClient().listTenantsAsync().get();
+
+ assertEquals(Set.copyOf(tenants), Set.of("DefaultTenant", "brokerTenant"));
+ }
+
+ @Test
+ public void testRebalanceTenantUsesIncludeExcludeTablesAndBody()
+ throws Exception {
+ JsonNode mockResponse = new ObjectMapper().readTree("{\"status\":\"OK\"}");
+ lenient().when(_mockTransport.executePost(anyString(), anyString(), any(),
any(), any()))
+ .thenReturn(mockResponse);
+
+ _adminClient.getTenantClient().rebalanceTenant("DefaultTenant", 2,
"tbl1_OFFLINE", "tbl2_OFFLINE");
+
+ /// The controller requires a TenantRebalanceConfig body; an empty object
lets the query params populate it.
+ verify(_mockTransport).executePost(eq(CONTROLLER_ADDRESS),
eq("/tenants/DefaultTenant/rebalance"), eq("{}"),
+ eq(Map.of("degreeOfParallelism", "2", "includeTables", "tbl1_OFFLINE",
"excludeTables", "tbl2_OFFLINE")),
+ eq(HEADERS));
+ }
+
+ @Test
+ public void testListSchemaNamesEmptyArrayReturnsEmptyList()
+ throws Exception {
+ JsonNode mockResponse = new ObjectMapper().readTree("[]");
+ lenient().when(_mockTransport.executeGet(anyString(), anyString(), any(),
any())).thenReturn(mockResponse);
+
+ assertEquals(_adminClient.getSchemaClient().listSchemaNames(), List.of());
+ }
+
+ @Test(expectedExceptions = PinotAdminException.class)
+ public void testListSchemaNamesThrowsOnUnexpectedResponse()
+ throws Exception {
+ /// A non-array response (e.g. an error object) must surface as an
exception, not a silently empty list.
+ JsonNode mockResponse = new ObjectMapper().readTree("{\"code\":500}");
+ lenient().when(_mockTransport.executeGet(anyString(), anyString(), any(),
any())).thenReturn(mockResponse);
+
+ _adminClient.getSchemaClient().listSchemaNames();
+ }
+
+ @Test
+ public void testListTenantsHandlesMissingTenantArrays()
+ throws Exception {
+ /// Defensive: a response without SERVER_TENANTS/BROKER_TENANTS must yield
an empty list, not an NPE.
+ JsonNode mockResponse = new ObjectMapper().readTree("{}");
+ lenient().when(_mockTransport.executeGet(anyString(), anyString(), any(),
any())).thenReturn(mockResponse);
+
+ assertEquals(_adminClient.getTenantClient().listTenants(), List.of());
+ }
}
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/ControllerRequestClient.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/ControllerRequestClient.java
deleted file mode 100644
index e3c8fb952ab..00000000000
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/ControllerRequestClient.java
+++ /dev/null
@@ -1,551 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.pinot.controller.helix;
-
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.node.ArrayNode;
-import java.io.IOException;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import javax.annotation.Nullable;
-import org.apache.pinot.common.exception.HttpErrorStatusException;
-import org.apache.pinot.common.restlet.resources.PauseStatusDetails;
-import org.apache.pinot.common.utils.SimpleHttpResponse;
-import org.apache.pinot.common.utils.http.HttpClient;
-import org.apache.pinot.spi.config.table.TableConfig;
-import org.apache.pinot.spi.config.table.TableType;
-import org.apache.pinot.spi.config.tenant.Tenant;
-import org.apache.pinot.spi.config.tenant.TenantRole;
-import org.apache.pinot.spi.config.workload.QueryWorkloadConfig;
-import org.apache.pinot.spi.data.LogicalTableConfig;
-import org.apache.pinot.spi.data.Schema;
-import org.apache.pinot.spi.utils.JsonUtils;
-import org.apache.pinot.spi.utils.builder.ControllerRequestURLBuilder;
-
-
-/**
- * The {@code ControllerRequestClient} provides handy utilities to make
request to controller.
- *
- * <p>It should be provided with a specified {@link
ControllerRequestURLBuilder} for constructing the URL requests
- * as well as a reusable {@link HttpClient} during construction.
- */
-public class ControllerRequestClient {
- private final HttpClient _httpClient;
- private final ControllerRequestURLBuilder _controllerRequestURLBuilder;
- private final Map<String, String> _headers;
-
- public ControllerRequestClient(ControllerRequestURLBuilder
controllerRequestUrlBuilder, HttpClient httpClient) {
- this(controllerRequestUrlBuilder, httpClient, Collections.emptyMap());
- }
-
- public ControllerRequestClient(ControllerRequestURLBuilder
controllerRequestUrlBuilder, HttpClient httpClient,
- Map<String, String> headers) {
- _controllerRequestURLBuilder = controllerRequestUrlBuilder;
- _httpClient = httpClient;
- _headers = headers;
- }
-
- public ControllerRequestURLBuilder getControllerRequestURLBuilder() {
- return _controllerRequestURLBuilder;
- }
- /**
- * Add a schema to the controller.
- */
- public void addSchema(Schema schema)
- throws IOException {
- String url = _controllerRequestURLBuilder.forSchemaCreate();
- try {
- HttpClient.wrapAndThrowHttpException(
- _httpClient.sendMultipartPostRequest(url,
schema.toSingleLineJsonString(), _headers));
- } catch (HttpErrorStatusException e) {
- throw new IOException(e);
- }
- }
-
- public Schema getSchema(String schemaName)
- throws IOException {
- String url = _controllerRequestURLBuilder.forSchemaGet(schemaName);
- try {
- SimpleHttpResponse resp =
- HttpClient.wrapAndThrowHttpException(_httpClient.sendGetRequest(new
URI(url), _headers));
- return Schema.fromString(resp.getResponse());
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public void updateSchema(Schema schema)
- throws IOException {
- String url =
_controllerRequestURLBuilder.forSchemaUpdate(schema.getSchemaName());
- try {
- HttpClient.wrapAndThrowHttpException(
- _httpClient.sendMultipartPutRequest(url,
schema.toSingleLineJsonString(), _headers));
- } catch (HttpErrorStatusException e) {
- throw new IOException(e);
- }
- }
-
- public void forceUpdateSchema(Schema schema)
- throws IOException {
- String url =
_controllerRequestURLBuilder.forSchemaUpdate(schema.getSchemaName()) +
"?force=true";
- try {
- HttpClient.wrapAndThrowHttpException(
- _httpClient.sendMultipartPutRequest(url,
schema.toSingleLineJsonString(), _headers));
- } catch (HttpErrorStatusException e) {
- throw new IOException(e);
- }
- }
-
- public void deleteSchema(String schemaName)
- throws IOException {
- String url = _controllerRequestURLBuilder.forSchemaDelete(schemaName);
- try {
- HttpClient.wrapAndThrowHttpException(_httpClient.sendDeleteRequest(new
URI(url), _headers));
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public void addTableConfig(TableConfig tableConfig)
- throws IOException {
- try {
- HttpClient.wrapAndThrowHttpException(
- _httpClient.sendJsonPostRequest(new
URI(_controllerRequestURLBuilder.forTableCreate()),
- tableConfig.toJsonString(), _headers));
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public void addLogicalTableConfig(LogicalTableConfig logicalTableConfig)
- throws IOException {
- try {
- HttpClient.wrapAndThrowHttpException(
- _httpClient.sendJsonPostRequest(new
URI(_controllerRequestURLBuilder.forLogicalTableCreate()),
- logicalTableConfig.toJsonString(), _headers));
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public void updateTableConfig(TableConfig tableConfig)
- throws IOException {
- try {
- HttpClient.wrapAndThrowHttpException(_httpClient.sendJsonPutRequest(
- new
URI(_controllerRequestURLBuilder.forUpdateTableConfig(tableConfig.getTableName())),
- tableConfig.toJsonString(), _headers));
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public void updateLogicalTableConfig(LogicalTableConfig logicalTableConfig)
- throws IOException {
- try {
- HttpClient.wrapAndThrowHttpException(_httpClient.sendJsonPutRequest(
- new
URI(_controllerRequestURLBuilder.forLogicalTableUpdate(logicalTableConfig.getTableName())),
- logicalTableConfig.toJsonString(), _headers));
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public void toggleTableState(String tableName, TableType type, boolean
enable)
- throws IOException {
- try {
- HttpClient.wrapAndThrowHttpException(_httpClient.sendPutRequest(
- new URI(_controllerRequestURLBuilder.forToggleTableState(tableName,
type, enable)), null, _headers));
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public void deleteTable(String tableNameWithType)
- throws IOException {
- deleteTable(tableNameWithType, null);
- }
-
- public void deleteTable(String tableNameWithType, String retentionPeriod)
- throws IOException {
- try {
- HttpClient.wrapAndThrowHttpException(
- _httpClient.sendDeleteRequest(
- new
URI(_controllerRequestURLBuilder.forTableDelete(tableNameWithType,
retentionPeriod)),
- _headers));
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public void deleteLogicalTable(String logicalTableName)
- throws IOException {
- try {
- HttpClient.wrapAndThrowHttpException(
- _httpClient.sendDeleteRequest(new
URI(_controllerRequestURLBuilder.forLogicalTableDelete(logicalTableName)),
- _headers));
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public String getLogicalTable(String logicalTableName)
- throws IOException {
- try {
- SimpleHttpResponse response = HttpClient.wrapAndThrowHttpException(
- _httpClient.sendGetRequest(
- new
URI(_controllerRequestURLBuilder.forLogicalTableGet(logicalTableName)),
_headers));
- return response.getResponse();
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public List<String> getLogicalTableNames()
- throws IOException {
- try {
- SimpleHttpResponse response = HttpClient.wrapAndThrowHttpException(
- _httpClient.sendGetRequest(
- new URI(_controllerRequestURLBuilder.forLogicalTableNamesGet()),
_headers));
- return JsonUtils.stringToObject(response.getResponse(), List.class);
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public TableConfig getTableConfig(String tableName, TableType tableType)
- throws IOException {
- try {
- SimpleHttpResponse response = HttpClient.wrapAndThrowHttpException(
- _httpClient.sendGetRequest(new
URI(_controllerRequestURLBuilder.forTableGet(tableName)), _headers));
- return
JsonUtils.jsonNodeToObject(JsonUtils.stringToJsonNode(response.getResponse()).get(tableType.toString()),
- TableConfig.class);
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public long getTableSize(String tableName)
- throws IOException {
- try {
- SimpleHttpResponse response = HttpClient.wrapAndThrowHttpException(
- _httpClient.sendGetRequest(new
URI(_controllerRequestURLBuilder.forTableSize(tableName)), _headers));
- return
Long.parseLong(JsonUtils.stringToJsonNode(response.getResponse()).get("reportedSizeInBytes").asText());
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public void resetTable(String tableNameWithType, String targetInstance)
- throws IOException {
- try {
- HttpClient.wrapAndThrowHttpException(_httpClient.sendJsonPostRequest(
- new
URI(_controllerRequestURLBuilder.forTableReset(tableNameWithType,
targetInstance)), null, _headers));
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public void resetSegment(String tableNameWithType, String segmentName,
String targetInstance)
- throws IOException {
- try {
- HttpClient.wrapAndThrowHttpException(_httpClient.sendJsonPostRequest(
- new
URI(_controllerRequestURLBuilder.forSegmentReset(tableNameWithType,
segmentName, targetInstance)), null,
- _headers));
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public String reloadTable(String tableName, TableType tableType, boolean
forceDownload)
- throws IOException {
- try {
- SimpleHttpResponse simpleHttpResponse =
HttpClient.wrapAndThrowHttpException(_httpClient.sendJsonPostRequest(
- new URI(_controllerRequestURLBuilder.forTableReload(tableName,
tableType, forceDownload)), null, _headers));
- return simpleHttpResponse.getResponse();
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public String checkIfReloadIsNeeded(String tableNameWithType, Boolean
verbose)
- throws IOException {
- try {
- SimpleHttpResponse simpleHttpResponse =
HttpClient.wrapAndThrowHttpException(_httpClient.sendGetRequest(
- new
URI(_controllerRequestURLBuilder.forTableNeedReload(tableNameWithType,
verbose)), _headers, null));
- return simpleHttpResponse.getResponse();
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public String reloadSegment(String tableName, String segmentName, boolean
forceReload)
- throws IOException {
- try {
- SimpleHttpResponse simpleHttpResponse =
HttpClient.wrapAndThrowHttpException(_httpClient.sendJsonPostRequest(
- new URI(_controllerRequestURLBuilder.forSegmentReload(tableName,
segmentName, forceReload)), null, _headers));
- return simpleHttpResponse.getResponse();
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public List<String> listSegments(String tableName, @Nullable String
tableType, boolean excludeReplacedSegments)
- throws IOException {
- String url = _controllerRequestURLBuilder.forSegmentListAPI(tableName,
tableType, excludeReplacedSegments);
- try {
- SimpleHttpResponse resp =
- HttpClient.wrapAndThrowHttpException(_httpClient.sendGetRequest(new
URI(url), _headers));
- // Example response: (list of map from table type to segments)
- //
[{"REALTIME":["mytable__0__0__20221012T1952Z","mytable__1__0__20221012T1952Z"]}]
- JsonNode jsonNode = JsonUtils.stringToJsonNode(resp.getResponse());
- List<String> segments = new ArrayList<>();
- for (JsonNode tableNode : jsonNode) {
- ArrayNode segmentsNode = (ArrayNode) tableNode.elements().next();
- for (JsonNode segmentNode : segmentsNode) {
- segments.add(segmentNode.asText());
- }
- }
- return segments;
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public Map<String, List<String>> getServersToSegmentsMap(String tableName,
TableType tableType)
- throws IOException {
- String url =
_controllerRequestURLBuilder.forServersToSegmentsMap(tableName,
tableType.toString());
- try {
- SimpleHttpResponse resp =
- HttpClient.wrapAndThrowHttpException(_httpClient.sendGetRequest(new
URI(url), _headers));
- JsonNode jsonNode = JsonUtils.stringToJsonNode(resp.getResponse());
- if (jsonNode == null || jsonNode.get(0) == null) {
- return Collections.emptyMap();
- }
-
- JsonNode serversMap = jsonNode.get(0).get("serverToSegmentsMap");
- if (serversMap == null) {
- return Collections.emptyMap();
- }
-
- HashMap<String, List<String>> result = new HashMap<>();
- for (Map.Entry<String, JsonNode> field : serversMap.properties()) {
- List<String> segments = new ArrayList<>();
-
- ArrayNode value = (ArrayNode) field.getValue();
- for (int i = 0, len = value.size(); i < len; i++) {
- segments.add(value.get(i).toString());
- }
-
- result.put(field.getKey(), segments);
- }
-
- return result;
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public void deleteSegment(String tableName, String segmentName)
- throws IOException {
- try {
- HttpClient.wrapAndThrowHttpException(_httpClient.sendDeleteRequest(
- new URI(_controllerRequestURLBuilder.forSegmentDelete(tableName,
segmentName)), _headers));
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public void deleteSegments(String tableName, TableType tableType)
- throws IOException {
- try {
- HttpClient.wrapAndThrowHttpException(_httpClient.sendDeleteRequest(
- new URI(_controllerRequestURLBuilder.forSegmentDeleteAll(tableName,
tableType.toString())), _headers));
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public PauseStatusDetails pauseConsumption(String tableName)
- throws IOException {
- try {
- SimpleHttpResponse response = HttpClient.wrapAndThrowHttpException(
- _httpClient.sendJsonPostRequest(new
URI(_controllerRequestURLBuilder.forPauseConsumption(tableName)), null,
- _headers));
- return JsonUtils.stringToObject(response.getResponse(),
PauseStatusDetails.class);
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public PauseStatusDetails resumeConsumption(String tableName)
- throws IOException {
- try {
- SimpleHttpResponse response = HttpClient.wrapAndThrowHttpException(
- _httpClient.sendJsonPostRequest(new
URI(_controllerRequestURLBuilder.forResumeConsumption(tableName)), null,
- _headers));
- return JsonUtils.stringToObject(response.getResponse(),
PauseStatusDetails.class);
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public PauseStatusDetails getPauseStatusDetails(String tableName)
- throws IOException {
- try {
- SimpleHttpResponse response = HttpClient.wrapAndThrowHttpException(
- _httpClient.sendGetRequest(new
URI(_controllerRequestURLBuilder.forPauseStatus(tableName)), _headers));
- return JsonUtils.stringToObject(response.getResponse(),
PauseStatusDetails.class);
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public void createBrokerTenant(String tenantName, int numBrokers)
- throws IOException {
- try {
- HttpClient.wrapAndThrowHttpException(
- _httpClient.sendJsonPostRequest(new
URI(_controllerRequestURLBuilder.forTenantCreate()),
- getBrokerTenantRequestPayload(tenantName, numBrokers),
_headers));
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public void updateBrokerTenant(String tenantName, int numBrokers)
- throws IOException {
- try {
- HttpClient.wrapAndThrowHttpException(
- _httpClient.sendJsonPutRequest(new
URI(_controllerRequestURLBuilder.forTenantCreate()),
- getBrokerTenantRequestPayload(tenantName, numBrokers),
_headers));
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public void deleteBrokerTenant(String tenantName)
- throws IOException {
- try {
- HttpClient.wrapAndThrowHttpException(
- _httpClient.sendDeleteRequest(new
URI(_controllerRequestURLBuilder.forBrokerTenantDelete(tenantName)),
- _headers));
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public void createServerTenant(String tenantName, int numOfflineServers, int
numRealtimeServers)
- throws IOException {
- try {
- HttpClient.wrapAndThrowHttpException(
- _httpClient.sendJsonPostRequest(new
URI(_controllerRequestURLBuilder.forTenantCreate()),
- getServerTenantRequestPayload(tenantName, numOfflineServers,
numRealtimeServers), _headers));
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public void updateServerTenant(String tenantName, int numOfflineServers, int
numRealtimeServers)
- throws IOException {
- try {
- HttpClient.wrapAndThrowHttpException(
- _httpClient.sendJsonPutRequest(new
URI(_controllerRequestURLBuilder.forTenantCreate()),
- getServerTenantRequestPayload(tenantName, numOfflineServers,
numRealtimeServers), _headers));
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public void runPeriodicTask(String taskName)
- throws IOException {
- try {
- HttpClient.wrapAndThrowHttpException(
- _httpClient.sendGetRequest(new
URI(_controllerRequestURLBuilder.forPeriodTaskRun(taskName)), _headers));
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- protected String getBrokerTenantRequestPayload(String tenantName, int
numBrokers) {
- return new Tenant(TenantRole.BROKER, tenantName, numBrokers, 0,
0).toJsonString();
- }
-
- protected static String getServerTenantRequestPayload(String tenantName, int
numOfflineServers,
- int numRealtimeServers) {
- return new Tenant(TenantRole.SERVER, tenantName, numOfflineServers +
numRealtimeServers, numOfflineServers,
- numRealtimeServers).toJsonString();
- }
-
- public void updateClusterConfig(Map<String, String> newConfigs)
- throws IOException {
- try {
- HttpClient.wrapAndThrowHttpException(_httpClient.sendJsonPostRequest(
- new URI(_controllerRequestURLBuilder.forClusterConfigUpdate()),
- JsonUtils.objectToString(newConfigs), _headers));
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public void deleteClusterConfig(String config)
- throws IOException {
- try {
- HttpClient.wrapAndThrowHttpException(_httpClient.sendDeleteRequest(
- new
URI(_controllerRequestURLBuilder.forClusterConfigDelete(config)), _headers));
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public void updateQueryWorkloadConfig(QueryWorkloadConfig
queryWorkloadConfig)
- throws IOException {
- try {
- HttpClient.wrapAndThrowHttpException(_httpClient.sendJsonPostRequest(
- new URI(_controllerRequestURLBuilder.forQueryWorkloadConfigUpdate()),
- JsonUtils.objectToString(queryWorkloadConfig), _headers));
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public void deleteQueryWorkloadConfig(String config)
- throws IOException {
- try {
- HttpClient.wrapAndThrowHttpException(_httpClient.sendDeleteRequest(
- new
URI(_controllerRequestURLBuilder.forBaseQueryWorkloadConfig(config)),
_headers));
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-
- public QueryWorkloadConfig getQueryWorkloadConfig(String config)
- throws IOException {
- try {
- SimpleHttpResponse response =
HttpClient.wrapAndThrowHttpException(_httpClient.sendGetRequest(
- new
URI(_controllerRequestURLBuilder.forBaseQueryWorkloadConfig(config)),
_headers));
- return JsonUtils.stringToObject(response.getResponse(),
QueryWorkloadConfig.class);
- } catch (HttpErrorStatusException | URISyntaxException e) {
- throw new IOException(e);
- }
- }
-}
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotAdminUserLogicalTableResourceTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotAdminUserLogicalTableResourceTest.java
index 16facae5cb0..9331e1ad61e 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotAdminUserLogicalTableResourceTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotAdminUserLogicalTableResourceTest.java
@@ -40,7 +40,7 @@ public class PinotAdminUserLogicalTableResourceTest extends
PinotLogicalTableRes
}
@Override
- protected Map<String, String> getControllerRequestClientHeaders() {
+ protected Map<String, String> getAdminClientHeaders() {
return AUTH_HEADER;
}
}
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotUserWithAccessLogicalTableResourceTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotUserWithAccessLogicalTableResourceTest.java
index aeee024104e..2f3c2c6156c 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotUserWithAccessLogicalTableResourceTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotUserWithAccessLogicalTableResourceTest.java
@@ -58,7 +58,7 @@ public class PinotUserWithAccessLogicalTableResourceTest
extends ControllerTest
}
@Override
- protected Map<String, String> getControllerRequestClientHeaders() {
+ protected Map<String, String> getAdminClientHeaders() {
return AUTH_HEADER;
}
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerTest.java
index abccf734bc6..ccff3ae96c2 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerTest.java
@@ -217,15 +217,13 @@ public class ControllerTest {
return _httpClient;
}
- /**
- * Retrieves the headers to be used for the `ControllerRequestClient`.
- *
- * <p>This method returns an empty map, indicating that no custom headers
- * are set by default for the `ControllerRequestClient`.
- *
- * @return A map of headers (key-value pairs) to be used for the
`ControllerRequestClient`.
- */
- protected Map<String, String> getControllerRequestClientHeaders() {
+ /// Retrieves the headers to be used for the [PinotAdminClient].
+ ///
+ /// This method returns an empty map, indicating that no custom headers
+ /// are set by default for the [PinotAdminClient].
+ ///
+ /// @return A map of headers (key-value pairs) to be used for the
[PinotAdminClient].
+ protected Map<String, String> getAdminClientHeaders() {
return Collections.emptyMap();
}
@@ -759,7 +757,7 @@ public class ControllerTest {
if (sslContext != null) {
org.apache.pinot.common.utils.tls.TlsUtils.setSslContext(sslContext);
}
- _pinotAdminClient = new PinotAdminClient(controllerAddress, properties,
getControllerRequestClientHeaders(),
+ _pinotAdminClient = new PinotAdminClient(controllerAddress, properties,
getAdminClientHeaders(),
sslContext);
return _pinotAdminClient;
} catch (PinotClientException e) {
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/CursorWithAuthIntegrationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/CursorWithAuthIntegrationTest.java
index a3cb9e13b16..d9502527f60 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/CursorWithAuthIntegrationTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/CursorWithAuthIntegrationTest.java
@@ -83,7 +83,7 @@ public class CursorWithAuthIntegrationTest extends
CursorIntegrationTest {
}
@Override
- protected Map<String, String> getControllerRequestClientHeaders() {
+ protected Map<String, String> getAdminClientHeaders() {
return AUTH_HEADER;
}
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RowLevelSecurityIntegrationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RowLevelSecurityIntegrationTest.java
index 98453035895..663c528d14e 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RowLevelSecurityIntegrationTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RowLevelSecurityIntegrationTest.java
@@ -101,7 +101,7 @@ public class RowLevelSecurityIntegrationTest extends
BaseClusterIntegrationTest
}
@Override
- protected Map<String, String> getControllerRequestClientHeaders() {
+ protected Map<String, String> getAdminClientHeaders() {
return AUTH_HEADER;
}
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/TimeSeriesAuthIntegrationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/TimeSeriesAuthIntegrationTest.java
index 41fa6e7ea18..5b18c1bf15e 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/TimeSeriesAuthIntegrationTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/TimeSeriesAuthIntegrationTest.java
@@ -38,7 +38,7 @@ public class TimeSeriesAuthIntegrationTest extends
TimeSeriesIntegrationTest {
}
@Override
- protected Map<String, String> getControllerRequestClientHeaders() {
+ protected Map<String, String> getAdminClientHeaders() {
return AUTH_HEADER;
}
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/TlsIntegrationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/TlsIntegrationTest.java
index 04acc15acd9..d14dced8578 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/TlsIntegrationTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/TlsIntegrationTest.java
@@ -281,7 +281,7 @@ public class TlsIntegrationTest extends
BaseClusterIntegrationTest {
}
@Override
- protected Map<String, String> getControllerRequestClientHeaders() {
+ protected Map<String, String> getAdminClientHeaders() {
return AUTH_HEADER;
}
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UrlAuthRealtimeIntegrationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UrlAuthRealtimeIntegrationTest.java
index c4130c4a96a..b7a41088280 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UrlAuthRealtimeIntegrationTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UrlAuthRealtimeIntegrationTest.java
@@ -137,7 +137,7 @@ public class UrlAuthRealtimeIntegrationTest extends
BaseClusterIntegrationTest {
}
@Override
- protected Map<String, String> getControllerRequestClientHeaders() {
+ protected Map<String, String> getAdminClientHeaders() {
return AUTH_HEADER;
}
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/multicluster/BaseMultiClusterIntegrationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/multicluster/BaseMultiClusterIntegrationTest.java
index ff0bc1dfe8e..91231292c9f 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/multicluster/BaseMultiClusterIntegrationTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/multicluster/BaseMultiClusterIntegrationTest.java
@@ -39,7 +39,6 @@ import org.apache.pinot.common.utils.FileUploadDownloadClient;
import org.apache.pinot.common.utils.ZkStarter;
import org.apache.pinot.controller.BaseControllerStarter;
import org.apache.pinot.controller.ControllerConf;
-import org.apache.pinot.controller.helix.ControllerRequestClient;
import org.apache.pinot.controller.helix.ControllerTest;
import org.apache.pinot.integration.tests.ClusterIntegrationTestUtils;
import org.apache.pinot.integration.tests.ClusterTest;
@@ -450,11 +449,9 @@ public abstract class BaseMultiClusterIntegrationTest
extends ClusterTest {
Map<String, PhysicalTableConfig> physicalTableConfigMap, String
brokerTenant, String controllerBaseApiUrl,
String logicalTable, String refOfflineTable, String refRealtimeTable)
throws IOException {
ControllerRequestURLBuilder urlBuilder =
ControllerRequestURLBuilder.baseUrl(controllerBaseApiUrl);
- ControllerRequestClient client = new ControllerRequestClient(urlBuilder,
getHttpClient(),
- getControllerRequestClientHeaders());
Schema schema = createSchema(schemaFile);
schema.setSchemaName(logicalTable);
- client.addSchema(schema);
+ addSchemaToCluster(schema, controllerBaseApiUrl);
LogicalTableConfig config = new LogicalTableConfigBuilder()
.setTableName(logicalTable)
.setBrokerTenant(brokerTenant)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
