This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new f8620508149 Remove legacy per-segment fallback from segments metadata
API (#20222)
f8620508149 is described below
commit f8620508149a33b144a187ed935f1d9ab39850b6
Author: Manak Bisht <[email protected]>
AuthorDate: Wed Nov 11 21:03:19 2026 +0530
Remove legacy per-segment fallback from segments metadata API (#20222)
---
.../pinot/controller/util/TableMetadataReader.java | 90 +++++-----------------
.../controller/util/TableMetadataReaderTest.java | 56 +++++++++-----
2 files changed, 54 insertions(+), 92 deletions(-)
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/util/TableMetadataReader.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/util/TableMetadataReader.java
index 71ba2bf8c40..c8dfdb60804 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/util/TableMetadataReader.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/util/TableMetadataReader.java
@@ -131,92 +131,38 @@ public class TableMetadataReader {
public JsonNode getSegmentsMetadata(String tableNameWithType, @Nullable
List<String> columns,
@Nullable List<String> segments, int timeoutMs)
throws InvalidConfigException, IOException {
- return getSegmentsMetadataInternal(tableNameWithType, columns, segments,
timeoutMs);
- }
+ Map<String, List<String>> serverToSegs =
+ _pinotHelixResourceManager.getServerToSegmentsMap(tableNameWithType);
+ BiMap<String, String> endpoints =
+
_pinotHelixResourceManager.getDataInstanceAdminEndpoints(serverToSegs.keySet());
+ ServerSegmentMetadataReader reader =
+ new ServerSegmentMetadataReader(_executor, _connectionManager,
+ _pinotHelixResourceManager.getServerAdminAuthProvider());
- /// Common helper used by both the new (server-level) and legacy
(segment-level) endpoints.
- private JsonNode fetchAndAggregateMetadata(List<String> urls, BiMap<String,
String> endpoints, boolean perSegmentJson,
- String tableNameWithType, int timeoutMs)
- throws InvalidConfigException, IOException {
- CompletionServiceHelper cs = new CompletionServiceHelper(_executor,
_connectionManager, endpoints,
+ List<String> tableUrls = new ArrayList<>(serverToSegs.size());
+ for (String server : serverToSegs.keySet()) {
+ tableUrls.add(reader.generateTableMetadataServerURL(
+ tableNameWithType, columns, segments, endpoints.get(server)));
+ }
+
+ CompletionServiceHelper cs = new CompletionServiceHelper(_executor,
_connectionManager, endpoints.inverse(),
_pinotHelixResourceManager.getServerAdminAuthProvider());
CompletionServiceHelper.CompletionServiceResponse resp =
- cs.doMultiGetRequest(urls, tableNameWithType, perSegmentJson,
timeoutMs);
- // all requests will fail if new server endpoint is not available
+ cs.doMultiGetRequest(tableUrls, tableNameWithType, false, timeoutMs);
if (resp._failedResponseCount > 0) {
- throw new RuntimeException(String.format("Got %d failed responses from
total %d server instances. "
- + "Falling back to legacy segment metadata api",
resp._failedResponseCount, urls.size()));
+ throw new RuntimeException(String.format("Got %d failed responses from
total %d server instances",
+ resp._failedResponseCount, tableUrls.size()));
}
ObjectMapper mapper = new ObjectMapper();
ObjectNode aggregatedNode = mapper.createObjectNode();
for (String body : resp._httpResponses.values()) {
JsonNode node = JsonUtils.stringToJsonNode(body);
- // legacy returns one JSON per segment; new returns one JSON with many
fields
- if (perSegmentJson) {
- String segmentName = node.get("segmentName").asText();
- aggregatedNode.set(segmentName, node);
- } else {
- node.properties().forEach(entry -> aggregatedNode.set(entry.getKey(),
entry.getValue()));
- }
+ node.properties().forEach(entry -> aggregatedNode.set(entry.getKey(),
entry.getValue()));
}
return aggregatedNode;
}
- private List<String> buildTableLevelUrls(Map<String, List<String>>
serverToSegs, BiMap<String, String> endpoints,
- String tableNameWithType, List<String> columns, List<String>
segmentsFilter, ServerSegmentMetadataReader reader) {
- List<String> urls = new ArrayList<>(serverToSegs.size());
- for (String server : serverToSegs.keySet()) {
- urls.add(reader.generateTableMetadataServerURL(
- tableNameWithType, columns, segmentsFilter, endpoints.get(server)));
- }
- return urls;
- }
-
- private List<String> buildSegmentLevelUrls(Map<String, List<String>>
serverToSegs, BiMap<String, String> endpoints,
- String tableNameWithType, List<String> columns, List<String>
segmentsFilter, ServerSegmentMetadataReader reader) {
- List<String> urls = new ArrayList<>();
- for (Map.Entry<String, List<String>> e : serverToSegs.entrySet()) {
- for (String segment : e.getValue()) {
- if (segmentsFilter == null || segmentsFilter.isEmpty()
- || segmentsFilter.contains(segment)) {
- urls.add(reader.generateSegmentMetadataServerURL(
- tableNameWithType, segment, columns, endpoints.get(e.getKey())));
- }
- }
- }
- return urls;
- }
-
- private JsonNode getSegmentsMetadataInternal(String tableNameWithType,
@Nullable List<String> columns,
- @Nullable List<String> segments, int timeoutMs)
- throws InvalidConfigException, IOException {
- Map<String, List<String>> serverToSegs =
- _pinotHelixResourceManager.getServerToSegmentsMap(tableNameWithType);
- BiMap<String, String> endpoints =
-
_pinotHelixResourceManager.getDataInstanceAdminEndpoints(serverToSegs.keySet());
- ServerSegmentMetadataReader reader =
- new ServerSegmentMetadataReader(_executor, _connectionManager,
- _pinotHelixResourceManager.getServerAdminAuthProvider());
-
- // try table level endpoint first
- try {
- List<String> tableUrls = buildTableLevelUrls(serverToSegs, endpoints,
- tableNameWithType, columns, segments, reader);
- return fetchAndAggregateMetadata(tableUrls, endpoints.inverse(),
/*perSegmentJson=*/false,
- tableNameWithType, timeoutMs);
- } catch (RuntimeException e) {
- log.warn("Failed to fetch table metadata for table {} using new server
endpoint, falling back to legacy "
- + "per-segment endpoint", tableNameWithType, e);
- }
-
- // legacy per segment endpoint
- List<String> segmentUrls = buildSegmentLevelUrls(serverToSegs, endpoints,
- tableNameWithType, columns, segments, reader);
- return fetchAndAggregateMetadata(segmentUrls, endpoints.inverse(),
/*perSegmentJson=*/true,
- tableNameWithType, timeoutMs);
- }
-
/// This method retrieves the full segment metadata for a given table and
segment
/// @return segment metadata
public JsonNode getSegmentMetadata(String tableNameWithType, String
segmentName, List<String> columns, int timeoutMs)
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/util/TableMetadataReaderTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/util/TableMetadataReaderTest.java
index e2fd78e9fde..d4d6a9cdd56 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/util/TableMetadataReaderTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/util/TableMetadataReaderTest.java
@@ -35,7 +35,6 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import java.util.concurrent.atomic.AtomicInteger;
import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.apache.pinot.spi.utils.JsonUtils;
@@ -48,6 +47,7 @@ import static org.mockito.ArgumentMatchers.anySet;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertThrows;
public class TableMetadataReaderTest {
@@ -57,7 +57,6 @@ public class TableMetadataReaderTest {
private final ExecutorService _executor = Executors.newFixedThreadPool(4);
private final PoolingHttpClientConnectionManager _connectionManager = new
PoolingHttpClientConnectionManager();
private final List<HttpServer> _httpServers = new ArrayList<>();
- private final AtomicInteger _legacyRequestCount = new AtomicInteger();
@BeforeClass
public void setUp() {
@@ -68,7 +67,6 @@ public class TableMetadataReaderTest {
public void stopServers() {
_httpServers.forEach(server -> server.stop(0));
_httpServers.clear();
- _legacyRequestCount.set(0);
}
@AfterClass
@@ -89,7 +87,6 @@ public class TableMetadataReaderTest {
JsonNode metadata = getSegmentsMetadata(serverToSegments);
assertEquals(segmentNames(metadata), Set.of("seg0", "seg1", "seg2"));
- assertEquals(_legacyRequestCount.get(), 0);
}
/// 4 servers with replication 2: each server holds only half of the
segments, so every server's response is needed.
@@ -105,7 +102,6 @@ public class TableMetadataReaderTest {
JsonNode metadata = getSegmentsMetadata(serverToSegments);
assertEquals(segmentNames(metadata), Set.of("seg0", "seg1", "seg2",
"seg3"));
- assertEquals(_legacyRequestCount.get(), 0);
}
/// Each server holds a distinct segment, so dropping any server's response
loses its segment.
@@ -120,14 +116,29 @@ public class TableMetadataReaderTest {
JsonNode metadata = getSegmentsMetadata(serverToSegments);
assertEquals(segmentNames(metadata), Set.of("seg0", "seg1", "seg2"));
- assertEquals(_legacyRequestCount.get(), 0);
+ }
+
+ /// A server whose table-level request fails must fail the whole call
instead of returning partial metadata.
+ @Test
+ public void testTableLevelMetadataFailsWhenServerReturnsError() {
+ Map<String, List<String>> serverToSegments = new LinkedHashMap<>();
+ serverToSegments.put("server0", List.of("seg0"));
+ serverToSegments.put("server1", List.of("seg1"));
+
+ assertThrows(RuntimeException.class, () ->
getSegmentsMetadata(serverToSegments, Set.of("server1")));
}
private JsonNode getSegmentsMetadata(Map<String, List<String>>
serverToSegments)
throws Exception {
+ return getSegmentsMetadata(serverToSegments, Set.of());
+ }
+
+ private JsonNode getSegmentsMetadata(Map<String, List<String>>
serverToSegments, Set<String> failingServers)
+ throws Exception {
BiMap<String, String> serverToEndpoint = HashBiMap.create();
for (Map.Entry<String, List<String>> entry : serverToSegments.entrySet()) {
- HttpServer httpServer = startServer(entry.getValue());
+ HttpServer httpServer =
+ failingServers.contains(entry.getKey()) ? startFailingServer() :
startServer(entry.getValue());
serverToEndpoint.put(entry.getKey(), "http://localhost:"; +
httpServer.getAddress().getPort());
}
@@ -139,8 +150,7 @@ public class TableMetadataReaderTest {
return reader.getSegmentsMetadata(TABLE_NAME, null, null, TIMEOUT_MS);
}
- /// Starts a fake server whose table-level endpoint returns metadata for the
given segments. The legacy per-segment
- /// endpoint returns 404, so tests can assert that the table-level path was
used.
+ /// Starts a fake server whose table-level endpoint returns metadata for the
given segments.
private HttpServer startServer(List<String> segments)
throws IOException {
ObjectNode tableLevelResponse = JsonUtils.newObjectNode();
@@ -150,17 +160,10 @@ public class TableMetadataReaderTest {
byte[] tableLevelBody =
JsonUtils.objectToString(tableLevelResponse).getBytes(StandardCharsets.UTF_8);
HttpServer httpServer = HttpServer.create(new
InetSocketAddress("localhost", 0), 0);
- httpServer.createContext("/tables/", exchange -> {
- String path = exchange.getRequestURI().getPath();
- if (path.equals("/tables/" + TABLE_NAME + "/segments/metadata")) {
- exchange.sendResponseHeaders(200, tableLevelBody.length);
- try (OutputStream out = exchange.getResponseBody()) {
- out.write(tableLevelBody);
- }
- } else {
- _legacyRequestCount.incrementAndGet();
- exchange.sendResponseHeaders(404, -1);
- exchange.close();
+ httpServer.createContext("/tables/" + TABLE_NAME + "/segments/metadata",
exchange -> {
+ exchange.sendResponseHeaders(200, tableLevelBody.length);
+ try (OutputStream out = exchange.getResponseBody()) {
+ out.write(tableLevelBody);
}
});
httpServer.start();
@@ -168,6 +171,19 @@ public class TableMetadataReaderTest {
return httpServer;
}
+ /// Starts a fake server whose table-level endpoint always returns 500.
+ private HttpServer startFailingServer()
+ throws IOException {
+ HttpServer httpServer = HttpServer.create(new
InetSocketAddress("localhost", 0), 0);
+ httpServer.createContext("/tables/" + TABLE_NAME + "/segments/metadata",
exchange -> {
+ exchange.sendResponseHeaders(500, -1);
+ exchange.close();
+ });
+ httpServer.start();
+ _httpServers.add(httpServer);
+ return httpServer;
+ }
+
private static Set<String> segmentNames(JsonNode metadata) {
Set<String> segmentNames = new HashSet<>();
metadata.fieldNames().forEachRemaining(segmentNames::add);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
