This is an automated email from the ASF dual-hosted git repository.
swaminathanmanish 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 eaef9a2a30a Revert "Remove legacy per-segment fallback from segments
metadata API (#20222)" (#20237)
eaef9a2a30a is described below
commit eaef9a2a30ae2327b59150560653412cf30d5262
Author: Manak Bisht <[email protected]>
AuthorDate: Thu Nov 12 13:20:21 2026 +0530
Revert "Remove legacy per-segment fallback from segments metadata API
(#20222)" (#20237)
This reverts commit f8620508149a33b144a187ed935f1d9ab39850b6.
---
.../pinot/controller/util/TableMetadataReader.java | 90 +++++++++++++++++-----
.../controller/util/TableMetadataReaderTest.java | 56 +++++---------
2 files changed, 92 insertions(+), 54 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 c8dfdb60804..71ba2bf8c40 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,38 +131,92 @@ public class TableMetadataReader {
public JsonNode getSegmentsMetadata(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());
-
- List<String> tableUrls = new ArrayList<>(serverToSegs.size());
- for (String server : serverToSegs.keySet()) {
- tableUrls.add(reader.generateTableMetadataServerURL(
- tableNameWithType, columns, segments, endpoints.get(server)));
- }
+ return getSegmentsMetadataInternal(tableNameWithType, columns, segments,
timeoutMs);
+ }
- CompletionServiceHelper cs = new CompletionServiceHelper(_executor,
_connectionManager, endpoints.inverse(),
+ /// 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,
_pinotHelixResourceManager.getServerAdminAuthProvider());
CompletionServiceHelper.CompletionServiceResponse resp =
- cs.doMultiGetRequest(tableUrls, tableNameWithType, false, timeoutMs);
+ cs.doMultiGetRequest(urls, tableNameWithType, perSegmentJson,
timeoutMs);
+ // all requests will fail if new server endpoint is not available
if (resp._failedResponseCount > 0) {
- throw new RuntimeException(String.format("Got %d failed responses from
total %d server instances",
- resp._failedResponseCount, tableUrls.size()));
+ 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()));
}
ObjectMapper mapper = new ObjectMapper();
ObjectNode aggregatedNode = mapper.createObjectNode();
for (String body : resp._httpResponses.values()) {
JsonNode node = JsonUtils.stringToJsonNode(body);
- node.properties().forEach(entry -> aggregatedNode.set(entry.getKey(),
entry.getValue()));
+ // 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()));
+ }
}
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 d4d6a9cdd56..e2fd78e9fde 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,6 +35,7 @@ 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;
@@ -47,7 +48,6 @@ 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,6 +57,7 @@ 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() {
@@ -67,6 +68,7 @@ public class TableMetadataReaderTest {
public void stopServers() {
_httpServers.forEach(server -> server.stop(0));
_httpServers.clear();
+ _legacyRequestCount.set(0);
}
@AfterClass
@@ -87,6 +89,7 @@ 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.
@@ -102,6 +105,7 @@ 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.
@@ -116,29 +120,14 @@ public class TableMetadataReaderTest {
JsonNode metadata = getSegmentsMetadata(serverToSegments);
assertEquals(segmentNames(metadata), Set.of("seg0", "seg1", "seg2"));
- }
-
- /// 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")));
+ assertEquals(_legacyRequestCount.get(), 0);
}
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 =
- failingServers.contains(entry.getKey()) ? startFailingServer() :
startServer(entry.getValue());
+ HttpServer httpServer = startServer(entry.getValue());
serverToEndpoint.put(entry.getKey(), "http://localhost:"; +
httpServer.getAddress().getPort());
}
@@ -150,7 +139,8 @@ 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.
+ /// 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.
private HttpServer startServer(List<String> segments)
throws IOException {
ObjectNode tableLevelResponse = JsonUtils.newObjectNode();
@@ -160,10 +150,17 @@ public class TableMetadataReaderTest {
byte[] tableLevelBody =
JsonUtils.objectToString(tableLevelResponse).getBytes(StandardCharsets.UTF_8);
HttpServer httpServer = HttpServer.create(new
InetSocketAddress("localhost", 0), 0);
- httpServer.createContext("/tables/" + TABLE_NAME + "/segments/metadata",
exchange -> {
- exchange.sendResponseHeaders(200, tableLevelBody.length);
- try (OutputStream out = exchange.getResponseBody()) {
- out.write(tableLevelBody);
+ 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.start();
@@ -171,19 +168,6 @@ 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]
