You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by xi...@apache.org on 2026/10/20 09:41:28 UTC
(pinot) branch xiangfu0/codex2/ratis-broker-routing created (now 386adf7e8b7)
This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a change to branch xiangfu0/codex2/ratis-broker-routing
in repository https://gitbox.apache.org/repos/asf/pinot.git
at 386adf7e8b7 Add authenticated native broker routing snapshots
This branch includes the following new commits:
new 386adf7e8b7 Add authenticated native broker routing snapshots
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
(pinot) 01/01: Add authenticated native broker routing snapshots
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch xiangfu0/codex2/ratis-broker-routing
in repository https://gitbox.apache.org/repos/asf/pinot.git
commit 386adf7e8b73f0130d5cc8181b05f1e3fc2e33d3
Author: Xiang Fu <xi...@apache.org>
AuthorDate: Tue Oct 20 02:41:14 2026 -0700
Add authenticated native broker routing snapshots
---
.../broker/broker/helix/BaseBrokerStarter.java | 8 +-
.../broker/helix/NativeBrokerParticipant.java | 122 +++++++++-
.../broker/routing/NativeBrokerRoutingImage.java | 270 +++++++++++++++++++++
.../broker/helix/NativeBrokerParticipantTest.java | 94 ++++++-
.../routing/NativeBrokerRoutingImageTest.java | 174 +++++++++++++
.../common/coordination/MetadataRpcClient.java | 19 ++
.../common/coordination/MetadataRpcConnection.java | 4 +
pinot-common/src/main/proto/metadata.proto | 50 ++++
.../coordination/NativeControllerMetadata.java | 12 +-
.../coordination/NativeControllerResourceTest.java | 21 +-
pinot-coordination-ratis/CONTROLLER-API-PARITY.md | 3 +
pinot-coordination-ratis/README.md | 59 ++++-
.../coordination/ratis/MetadataStateMachine.java | 7 +
.../coordination/ratis/RatisMetadataAccess.java | 12 +-
.../ratis/metadata/MetadataCatalog.java | 245 +++++++++++++++++++
.../ratis/metadata/NativeBrokerRoutingPolicy.java | 138 +++++++++++
.../ratis/metadata/OfflineRoutingProtocol.java | 69 ++++++
.../ratis/transport/MetadataGrpcServer.java | 14 ++
.../ratis/transport/RoutingAccess.java | 31 +++
.../ratis/transport/RoutingGrpcService.java | 134 ++++++++++
.../ratis/transport/RoutingRpcCodec.java | 88 +++++++
.../ratis/metadata/TaskCatalogTest.java | 170 +++++++++++++
.../ratis/transport/MetadataGrpcTransportTest.java | 130 +++++++++-
.../pinot/tools/service/RatisStartersTest.java | 64 ++++-
24 files changed, 1917 insertions(+), 21 deletions(-)
diff --git a/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java b/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java
index 13fb043f8f4..906b03c98df 100644
--- a/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java
+++ b/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java
@@ -305,7 +305,13 @@ public abstract class BaseBrokerStarter implements ServiceStartable {
@VisibleForTesting
NativeBrokerParticipant createNativeParticipant(NativeMetadataConfig config) throws IOException {
- return new NativeBrokerParticipant(config, config.openConnection());
+ var membership = config.openConnection();
+ try {
+ return new NativeBrokerParticipant(config, membership, config.openConnection());
+ } catch (IOException | RuntimeException e) {
+ membership.close();
+ throw e;
+ }
}
/// Can be overridden to apply custom configs to the broker conf.
diff --git a/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/NativeBrokerParticipant.java b/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/NativeBrokerParticipant.java
index ddc9a0b71cb..3aa7bcda5bf 100644
--- a/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/NativeBrokerParticipant.java
+++ b/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/NativeBrokerParticipant.java
@@ -22,11 +22,14 @@ import com.google.common.annotations.VisibleForTesting;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;
+import org.apache.pinot.broker.routing.NativeBrokerRoutingImage;
import org.apache.pinot.common.coordination.MetadataRpcConnection;
import org.apache.pinot.common.coordination.NativeMetadataConfig;
import org.apache.pinot.common.coordination.NativeOfflineSegmentWorker;
@@ -45,11 +48,17 @@ import static com.google.common.base.Preconditions.checkState;
/// Lifecycle operations and the single bounded heartbeat lane are serialized; status reads never perform RPCs.
final class NativeBrokerParticipant implements AutoCloseable, ServiceStatus.ServiceStatusCallback {
private static final Logger LOGGER = LoggerFactory.getLogger(NativeBrokerParticipant.class);
+ private static final int MAX_ROUTING_TABLES = 128;
+ private static final int MAX_ROUTING_BYTES = 8 * 1024 * 1024;
private final NativeMetadataConfig _config;
private final MetadataRpcConnection _connection;
+ private final MetadataRpcConnection _routingConnection;
+ private final Object _routingLock = new Object();
private final UUID _incarnation = UUID.randomUUID();
private final ScheduledExecutorService _heartbeats = Executors.newSingleThreadScheduledExecutor(
runnable -> new Thread(runnable, "native-broker-membership"));
+ private final ScheduledExecutorService _routingPoller = Executors.newSingleThreadScheduledExecutor(
+ runnable -> new Thread(runnable, "native-broker-routing"));
@Nullable
private MetadataRpc.ParticipantToken _participant;
@Nullable
@@ -58,12 +67,20 @@ final class NativeBrokerParticipant implements AutoCloseable, ServiceStatus.Serv
private long _appliedIndex;
private boolean _started;
private boolean _fenced;
+ private boolean _membershipHealthy;
+ private long _routingGeneration;
+ // This is only a read-regression floor on the separate routing lane, never a global applied revision.
+ private long _routingIndex;
+ private Map<String, NativeBrokerRoutingImage> _routingImages = Map.of();
private volatile boolean _closed;
private volatile String _description = "Native broker membership starting; query listeners disabled";
- NativeBrokerParticipant(NativeMetadataConfig config, MetadataRpcConnection connection) {
+ NativeBrokerParticipant(NativeMetadataConfig config, MetadataRpcConnection connection,
+ MetadataRpcConnection routingConnection) {
+ checkArgument(connection != routingConnection, "Routing hydration requires a separate membership connection");
_config = config;
_connection = connection;
+ _routingConnection = routingConnection;
}
synchronized void start() throws IOException {
@@ -95,9 +112,12 @@ final class NativeBrokerParticipant implements AutoCloseable, ServiceStatus.Serv
// Retain confirmed ownership before checking granted capabilities so failed startup can retire this session.
_participant = token;
validateOwnProjection(registered);
+ _membershipHealthy = true;
_description = "Native broker membership active; routing unavailable and query listeners disabled";
_heartbeats.scheduleWithFixedDelay(this::heartbeat, _config.heartbeatMillis(), _config.heartbeatMillis(),
TimeUnit.MILLISECONDS);
+ _routingPoller.scheduleWithFixedDelay(this::refreshRouting, _config.heartbeatMillis(), _config.heartbeatMillis(),
+ TimeUnit.MILLISECONDS);
} catch (IOException | RuntimeException e) {
close();
throw e;
@@ -122,6 +142,8 @@ final class NativeBrokerParticipant implements AutoCloseable, ServiceStatus.Serv
_config.rpcTimeoutMillis());
validateResponse(response);
if (response.getStatus() == ResultStatus.NOT_READY) {
+ invalidateRouting();
+ _membershipHealthy = false;
_description = "Native broker control epoch unavailable; query listeners disabled";
return;
}
@@ -135,8 +157,12 @@ final class NativeBrokerParticipant implements AutoCloseable, ServiceStatus.Serv
checkArgument(_nextRenewal < Long.MAX_VALUE, "Broker renewal sequences exhausted");
_nextRenewal++;
_pendingRenewal = null;
+ _membershipHealthy = renewal.getStatus() == ResultStatus.OK;
+ if (!_membershipHealthy) {
+ invalidateRouting();
+ }
_description = renewal.getStatus() == ResultStatus.OK
- ? "Native broker membership active; routing unavailable and query listeners disabled"
+ ? routingDescription()
: "Native broker membership projection must catch up; query listeners disabled";
} catch (StatusRuntimeException e) {
Status.Code code = e.getStatus().getCode();
@@ -144,6 +170,8 @@ final class NativeBrokerParticipant implements AutoCloseable, ServiceStatus.Serv
|| code == Status.Code.INVALID_ARGUMENT) {
fence("Broker membership authorization or request rejected: " + code);
} else {
+ invalidateRouting();
+ _membershipHealthy = false;
_description = "Native broker membership transport unavailable; query listeners disabled";
}
} catch (IOException | IllegalArgumentException e) {
@@ -153,6 +181,90 @@ final class NativeBrokerParticipant implements AutoCloseable, ServiceStatus.Serv
}
}
+ /// One bounded discovery/hydration cycle. RPCs never hold the membership monitor; a heartbeat failure, fencing or
+ /// shutdown invalidates any overlapping cycle before its result can be published.
+ @VisibleForTesting
+ void refreshRouting() {
+ synchronized (_routingLock) {
+ MetadataRpc.ParticipantToken participant;
+ long generation;
+ synchronized (this) {
+ if (_closed || _fenced || !_membershipHealthy) {
+ return;
+ }
+ participant = _participant;
+ generation = _routingGeneration;
+ }
+ try {
+ long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(_config.rpcTimeoutMillis());
+ var discovery = routingSnapshot(participant, "", MAX_ROUTING_TABLES + 2, deadline);
+ var tables = NativeBrokerRoutingImage.discover(discovery, participant);
+ checkArgument(tables.size() <= MAX_ROUTING_TABLES, "Routing cache exceeds its table limit");
+ _routingIndex = discovery.getRevision().getIndex();
+ Map<String, NativeBrokerRoutingImage> images = new HashMap<>();
+ int bytes = discovery.getSerializedSize();
+ for (String table : tables) {
+ var snapshot = routingSnapshot(participant, table, NativeBrokerRoutingImage.MAX_RECORDS, deadline);
+ NativeBrokerRoutingImage image = new NativeBrokerRoutingImage(snapshot, participant, table);
+ _routingIndex = snapshot.getRevision().getIndex();
+ checkArgument(image.serializedBytes() <= MAX_ROUTING_BYTES - bytes,
+ "Routing cache exceeds its aggregate byte limit");
+ bytes += image.serializedBytes();
+ images.put(table, image);
+ }
+ synchronized (this) {
+ if (!_closed && !_fenced && _membershipHealthy && generation == _routingGeneration
+ && participant.equals(_participant)) {
+ _routingImages = Map.copyOf(images);
+ _description = routingDescription();
+ }
+ }
+ } catch (IOException | RuntimeException e) {
+ synchronized (this) {
+ invalidateRouting();
+ if (!_closed && !_fenced) {
+ _description = "Native broker routing unavailable: " + e.getMessage() + "; query listeners disabled";
+ }
+ }
+ }
+ }
+ }
+
+ private MetadataRpc.RoutingSnapshot routingSnapshot(MetadataRpc.ParticipantToken participant, String table,
+ int maxRecords, long deadline) throws IOException {
+ long remaining = TimeUnit.NANOSECONDS.toMillis(deadline - System.nanoTime());
+ if (_closed || remaining <= 0) {
+ throw new IOException("Native routing refresh deadline expired or broker stopped");
+ }
+ var response = _routingConnection.routingSnapshot(MetadataRpc.RoutingSnapshotRequest.newBuilder()
+ .setHistoryId(_config.historyId().toString()).setSession(MetadataRpc.ParticipantSession.newBuilder()
+ .setIncarnation(participant.getIncarnation()).setRegistrationEpoch(participant.getRegistrationEpoch()))
+ .setTableName(table).setMaxRecords(maxRecords).setMaxBytes(NativeBrokerRoutingImage.MAX_BYTES).build(),
+ remaining);
+ checkArgument(response.hasRevision() && response.getRevision().getIndex() >= _routingIndex,
+ "Routing revision regressed");
+ if (deadline - System.nanoTime() <= 0) {
+ throw new IOException("Native routing refresh deadline expired");
+ }
+ return response;
+ }
+
+ @VisibleForTesting
+ synchronized NativeBrokerRoutingImage routingImage(String table) {
+ checkState(!_closed && !_fenced && _membershipHealthy && _routingImages.containsKey(table),
+ "Current native table routing is unavailable");
+ return _routingImages.get(table);
+ }
+
+ private void invalidateRouting() {
+ _routingImages = Map.of();
+ _routingGeneration++;
+ }
+
+ private String routingDescription() {
+ return "Native broker has " + _routingImages.size() + " table routing images; query listeners disabled";
+ }
+
private MetadataRpc.MembershipResponse read() throws IOException {
MetadataRpc.MembershipResponse response = _connection.membershipRead(MetadataRpc.MembershipReadRequest.newBuilder()
.setHistoryId(_config.historyId().toString()).build(), _config.rpcTimeoutMillis());
@@ -203,8 +315,11 @@ final class NativeBrokerParticipant implements AutoCloseable, ServiceStatus.Serv
private void fence(String detail) {
_fenced = true;
+ _membershipHealthy = false;
+ invalidateRouting();
_description = detail + "; query listeners disabled";
_heartbeats.shutdown();
+ _routingPoller.shutdownNow();
LOGGER.warn("{}", _description);
}
@@ -224,8 +339,11 @@ final class NativeBrokerParticipant implements AutoCloseable, ServiceStatus.Serv
return;
}
_closed = true;
+ invalidateRouting();
_description = "Native broker stopped; query listeners disabled";
_heartbeats.shutdownNow();
+ _routingPoller.shutdownNow();
+ _routingConnection.close();
try {
if (_participant != null) {
try {
diff --git a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/NativeBrokerRoutingImage.java b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/NativeBrokerRoutingImage.java
new file mode 100644
index 00000000000..a7943bca6c7
--- /dev/null
+++ b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/NativeBrokerRoutingImage.java
@@ -0,0 +1,270 @@
+/**
+ * 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.broker.routing;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.UUID;
+import org.apache.pinot.common.config.provider.NativeOfflineSegmentAction.ObjectVersion;
+import org.apache.pinot.common.config.provider.NativeOfflineSegmentAction.Participant;
+import org.apache.pinot.common.config.provider.NativeOfflineSegmentAction.Role;
+import org.apache.pinot.common.coordination.NativeOfflineSegmentWorker;
+import org.apache.pinot.common.request.context.NativeQueryContext;
+import org.apache.pinot.common.request.context.NativeQueryContext.ProcessIdentity;
+import org.apache.pinot.common.request.context.NativeQueryContext.Segment;
+import org.apache.pinot.common.request.context.NativeQueryDeadline;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.ParticipantRole;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.ParticipantToken;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.ReplicaRouting;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.RoutingSnapshot;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.utils.builder.TableNameBuilder;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+import static java.util.Objects.requireNonNull;
+
+
+/// Immutable complete metadata for one OFFLINE table at one committed revision. Installed replicas without an
+/// approved query endpoint remain visible but cannot be selected. Neither this image nor a produced request grants
+/// serving authority: authenticated transport, admission, qualified deadlines and delivery fences remain mandatory.
+public final class NativeBrokerRoutingImage {
+ public static final int MAX_RECORDS = 4096;
+ public static final int MAX_BYTES = 2 * 1024 * 1024;
+ private final RoutingSnapshot _snapshot;
+
+ public NativeBrokerRoutingImage(RoutingSnapshot snapshot, ParticipantToken broker, String tableName)
+ throws IOException {
+ validateHeader(snapshot, broker);
+ checkArgument(snapshot.hasTable() && snapshot.getTablesCount() == 0
+ && snapshot.getTable().getTableName().equals(tableName), "Routing snapshot table differs from request");
+ canonicalTable(tableName);
+ var table = snapshot.getTable();
+ checkArgument(table.hasTable() && table.hasSchema()
+ && table.getSchemaName().equals(TableNameBuilder.extractRawTableName(tableName)),
+ "Routing table/schema identity is incomplete");
+ Set<UUID> identities = new HashSet<>();
+ checkArgument(identities.add(object(table.getTable(), broker.getHistoryId()).objectId())
+ && identities.add(object(table.getSchema(), broker.getHistoryId()).objectId()),
+ "Routing table/schema identities overlap");
+ Set<String> names = new HashSet<>();
+ Set<UUID> materializations = new HashSet<>();
+ Set<UUID> installations = new HashSet<>();
+ Map<String, ParticipantToken> servers = new HashMap<>();
+ Map<String, String> endpoints = new HashMap<>();
+ Map<String, String> endpointOwners = new HashMap<>();
+ int records = 4;
+ for (var segment : table.getSegmentsList()) {
+ checkArgument(++records <= MAX_RECORDS && names.add(identifier(segment.getName())) && segment.hasContent()
+ && identities.add(object(segment.getContent(), broker.getHistoryId()).objectId())
+ && segment.getContentSha256().matches("[0-9a-f]{64}"), "Invalid or repeated routing segment");
+ Set<String> replicas = new HashSet<>();
+ for (ReplicaRouting replica : segment.getReplicasList()) {
+ checkArgument(++records <= MAX_RECORDS && replica.hasServer() && replica.hasAssignment(),
+ "Routing replica is incomplete or exceeds the record bound");
+ ParticipantToken server = replica.getServer();
+ participant(server, broker.getHistoryId(), ParticipantRole.SERVER);
+ checkArgument(replicas.add(server.getParticipantId())
+ && identities.add(object(replica.getAssignment(), broker.getHistoryId()).objectId())
+ && replica.getAssignmentEpoch() == replica.getAssignment().getVersion()
+ && materializations.add(uuid(replica.getMaterializationId()))
+ && installations.add(uuid(replica.getInstallationId())), "Invalid or repeated routing replica");
+ ParticipantToken previous = servers.putIfAbsent(server.getParticipantId(), server);
+ checkArgument(previous == null || previous.equals(server), "Routing image mixes server incarnations");
+ String endpoint = replica.hasEndpoint() ? replica.getEndpoint() : "";
+ if (replica.hasEndpoint()) {
+ endpoint(endpoint);
+ String owner = endpointOwners.putIfAbsent(endpoint, server.getParticipantId());
+ checkArgument(owner == null || owner.equals(server.getParticipantId()),
+ "Query endpoint identifies multiple servers");
+ }
+ String priorEndpoint = endpoints.putIfAbsent(server.getParticipantId(), endpoint);
+ checkArgument(priorEndpoint == null || priorEndpoint.equals(endpoint),
+ "Routing image mixes query endpoint approvals");
+ }
+ }
+ _snapshot = snapshot;
+ }
+
+ /// Validates the own-session discovery result. It is not a table image or evidence that any table is routable.
+ public static List<String> discover(RoutingSnapshot snapshot, ParticipantToken broker) throws IOException {
+ validateHeader(snapshot, broker);
+ checkArgument(!snapshot.hasTable() && snapshot.getTablesCount() <= MAX_RECORDS - 2,
+ "Invalid routing discovery response");
+ Set<String> tables = new HashSet<>();
+ for (String table : snapshot.getTablesList()) {
+ canonicalTable(table);
+ checkArgument(tables.add(table), "Repeated routing discovery table");
+ }
+ return tables.stream().sorted().toList();
+ }
+
+ public MetadataRpc.Revision revision() {
+ return _snapshot.getRevision();
+ }
+
+ public int serializedBytes() {
+ return _snapshot.getSerializedSize();
+ }
+
+ public String tableName() {
+ return _snapshot.getTable().getTableName();
+ }
+
+ public int segmentCount() {
+ return _snapshot.getTable().getSegmentsCount();
+ }
+
+ /// Selects exactly one approved replica for every segment, in deterministic order. The same originating deadline
+ /// is attached to every fanout request; callers must preserve it across retries and independently qualify its clock.
+ public List<ServerRequest> requests(UUID queryId, UUID attemptId, NativeQueryDeadline deadline) {
+ requireNonNull(queryId, "queryId");
+ requireNonNull(attemptId, "attemptId");
+ requireNonNull(deadline, "deadline");
+ var table = _snapshot.getTable();
+ checkState(table.getSegmentsCount() > 0, "Native empty-table query execution is not supported");
+ Map<String, List<Segment>> selected = new TreeMap<>();
+ Map<String, ReplicaRouting> targets = new HashMap<>();
+ for (var segment : table.getSegmentsList().stream()
+ .sorted(Comparator.comparing(MetadataRpc.SegmentRouting::getName)).toList()) {
+ ReplicaRouting replica = segment.getReplicasList().stream().filter(ReplicaRouting::hasEndpoint)
+ .min(Comparator.comparing(value -> value.getServer().getParticipantId())).orElseThrow(
+ () -> new IllegalStateException("No approved query replica for required segment: " + segment.getName()));
+ String server = replica.getServer().getParticipantId();
+ targets.put(server, replica);
+ selected.computeIfAbsent(server, ignored -> new ArrayList<>()).add(new Segment(segment.getName(),
+ object(segment.getContent(), _snapshot.getRevision().getHistoryId()), segment.getContentSha256(),
+ object(replica.getAssignment(), _snapshot.getRevision().getHistoryId()),
+ uuid(replica.getMaterializationId())));
+ }
+ ParticipantToken broker = _snapshot.getParticipant().getParticipant();
+ ProcessIdentity process = new ProcessIdentity(broker.getParticipantId(), uuid(broker.getIncarnation()),
+ broker.getRegistrationEpoch());
+ List<ServerRequest> requests = new ArrayList<>();
+ selected.forEach((id, segments) -> {
+ ReplicaRouting replica = targets.get(id);
+ ParticipantToken server = replica.getServer();
+ NativeQueryContext context = new NativeQueryContext(NativeQueryContext.VERSION, uuid(broker.getHistoryId()),
+ process, new Participant(id, Role.SERVER, uuid(server.getIncarnation()), server.getRegistrationEpoch()),
+ queryId, attemptId, table.getTableName(), object(table.getTable(), broker.getHistoryId()),
+ table.getSchemaName(), object(table.getSchema(), broker.getHistoryId()), segments, List.of())
+ .withDeadline(deadline);
+ requests.add(new ServerRequest(endpoint(replica.getEndpoint()), context));
+ });
+ return List.copyOf(requests);
+ }
+
+ /// An exact destination and identity envelope, not a dispatched or admitted query.
+ public record ServerRequest(URI endpoint, NativeQueryContext context) {
+ }
+
+ private static void validateHeader(RoutingSnapshot snapshot, ParticipantToken broker) throws IOException {
+ NativeOfflineSegmentWorker.validateMessage(snapshot);
+ participant(broker, broker.getHistoryId(), ParticipantRole.BROKER);
+ checkArgument(snapshot.getSerializedSize() <= MAX_BYTES && snapshot.getStatus() == MetadataRpc.ResultStatus.OK
+ && snapshot.hasRevision() && snapshot.getRevision().getIndex() > 0
+ && snapshot.getRevision().getHistoryId().equals(broker.getHistoryId())
+ && snapshot.hasParticipant() && snapshot.hasControl(), "Incomplete routing snapshot");
+ var state = snapshot.getParticipant();
+ var control = snapshot.getControl();
+ long index = snapshot.getRevision().getIndex();
+ checkArgument(state.getParticipant().equals(broker) && !state.getRetired() && !state.getDraining()
+ && state.getExpiryRound() > control.getRound() && state.getLastModifiedIndex() > 0
+ && state.getLastModifiedIndex() <= index && state.getLastRenewalSequence() >= 0
+ && state.hasAppliedRevision() && state.getAppliedRevision().getHistoryId().equals(broker.getHistoryId())
+ && state.getAppliedRevision().getIndex() >= 0 && state.getAppliedRevision().getIndex() <= index,
+ "Routing snapshot broker session is not live");
+ checkArgument(control.hasControlEpoch() && control.hasPolicy(), "Routing control state is incomplete");
+ var epoch = control.getControlEpoch();
+ checkArgument(epoch.getHistoryId().equals(broker.getHistoryId()) && epoch.getRaftTerm() > 0
+ && epoch.getBarrierIndex() > 0 && epoch.getBarrierIndex() <= index && !epoch.getLeaderId().isBlank()
+ && control.getRound() >= 0 && control.getClockBarrierIndex() >= epoch.getBarrierIndex()
+ && control.getClockBarrierIndex() <= index, "Invalid routing control barrier");
+ uuid(epoch.getActivationId());
+ var policy = control.getPolicy();
+ checkArgument(policy.getVersion() > 0 && policy.getMinTickIntervalMillis() > 0 && policy.getLeaseRounds() >= 2
+ && policy.getPermitLifetimeMillis() > 0 && policy.getMaxQueryLifetimeMillis() > 0
+ && policy.getMaxRevisionLag() > 0, "Invalid routing membership policy");
+ }
+
+ private static ObjectVersion object(MetadataRpc.ObjectVersion value, String history) {
+ checkArgument(value.getHistoryId().equals(history), "Routing object belongs to a different history");
+ return new ObjectVersion(uuid(value.getObjectId()), value.getVersion());
+ }
+
+ private static void participant(ParticipantToken value, String history, ParticipantRole role) {
+ checkArgument(value.getHistoryId().equals(history) && value.getRole() == role
+ && value.getRegistrationEpoch() > 0 && value.getParticipantId().length() <= 128,
+ "Invalid routing participant");
+ identifier(value.getParticipantId());
+ uuid(history);
+ uuid(value.getIncarnation());
+ }
+
+ private static void canonicalTable(String table) {
+ identifier(table);
+ String raw = TableNameBuilder.extractRawTableName(table);
+ int separator = raw.indexOf('.');
+ checkArgument(TableNameBuilder.getTableTypeFromTableName(table) == TableType.OFFLINE && !raw.isBlank()
+ && TableNameBuilder.getTableTypeFromTableName(raw) == null
+ && (separator < 0 || separator > 0 && separator < raw.length() - 1 && separator == raw.lastIndexOf('.'))
+ && table.chars().noneMatch(Character::isWhitespace), "Invalid canonical OFFLINE routing table");
+ }
+
+ private static String identifier(String value) {
+ checkArgument(value != null && !value.isBlank() && value.length() <= 1024
+ && value.chars().noneMatch(Character::isISOControl) && StandardCharsets.UTF_8.newEncoder().canEncode(value),
+ "Invalid routing identifier");
+ return value;
+ }
+
+ private static UUID uuid(String value) {
+ UUID parsed = UUID.fromString(value);
+ checkArgument(parsed.toString().equals(value), "Invalid canonical routing UUID");
+ return parsed;
+ }
+
+ private static URI endpoint(String value) {
+ URI uri = URI.create(value);
+ checkArgument(value.length() <= 2048 && "grpcs".equals(uri.getScheme()) && uri.getHost() != null
+ && uri.getHost().equals(uri.getHost().toLowerCase(Locale.ROOT)) && uri.getPort() > 0
+ && uri.getPort() <= 65535 && uri.getRawUserInfo() == null
+ && (uri.getRawPath() == null || uri.getRawPath().isEmpty())
+ && uri.getRawQuery() == null && uri.getRawFragment() == null, "Unapproved native query endpoint");
+ try {
+ String canonical = new URI("grpcs", null, uri.getHost(), uri.getPort(), null, null, null).toASCIIString();
+ checkArgument(value.equals(canonical), "Noncanonical native query endpoint");
+ } catch (URISyntaxException e) {
+ throw new IllegalArgumentException("Invalid native query endpoint", e);
+ }
+ return uri;
+ }
+}
diff --git a/pinot-broker/src/test/java/org/apache/pinot/broker/broker/helix/NativeBrokerParticipantTest.java b/pinot-broker/src/test/java/org/apache/pinot/broker/broker/helix/NativeBrokerParticipantTest.java
index 74e67f5d4db..da258c9ffd7 100644
--- a/pinot-broker/src/test/java/org/apache/pinot/broker/broker/helix/NativeBrokerParticipantTest.java
+++ b/pinot-broker/src/test/java/org/apache/pinot/broker/broker/helix/NativeBrokerParticipantTest.java
@@ -28,6 +28,9 @@ import java.util.Deque;
import java.util.List;
import java.util.Map;
import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
import org.apache.pinot.common.coordination.MetadataRpcConnection;
import org.apache.pinot.common.coordination.NativeMetadataConfig;
import org.apache.pinot.common.utils.ServiceStatus;
@@ -38,6 +41,7 @@ import org.testng.annotations.Test;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
@@ -49,7 +53,7 @@ import static org.testng.Assert.assertThrows;
import static org.testng.Assert.assertTrue;
-/// Covers membership-only broker startup and exact renewal recovery without opening query or metadata listeners.
+/// Covers isolated membership/routing lanes and exact renewal recovery without opening query or metadata listeners.
public class NativeBrokerParticipantTest {
@Test(timeOut = 10_000)
public void testNormalStarterBypassesZookeeperAndLeavesQueriesUnavailable() throws Exception {
@@ -189,11 +193,68 @@ public class NativeBrokerParticipantTest {
}
}
+ @Test(timeOut = 10_000)
+ public void testRoutingFailureInvalidatesImagesWithoutClaimingGlobalApplication() throws Exception {
+ try (Fixture fixture = new Fixture()) {
+ fixture._participant.start();
+ fixture._participant.refreshRouting();
+ long routingIndex = fixture._participant.routingImage("orders_OFFLINE").revision().getIndex();
+ fixture._participant.heartbeat();
+ var renewal = fixture._requests.getLast().getRenew();
+ assertTrue(renewal.getAppliedRevision().getIndex() < routingIndex,
+ "Per-table routing captures must not advance the membership-only applied revision");
+ fixture._routingFailure = true;
+ fixture._participant.refreshRouting();
+ assertThrows(IllegalStateException.class, () -> fixture._participant.routingImage("orders_OFFLINE"));
+ fixture._routingFailure = false;
+ fixture._participant.refreshRouting();
+ assertTrue(fixture._participant.routingImage("orders_OFFLINE").revision().getIndex() > routingIndex);
+ fixture._routingTables = List.of();
+ fixture._participant.refreshRouting();
+ assertThrows(IllegalStateException.class, () -> fixture._participant.routingImage("orders_OFFLINE"));
+ assertEquals(fixture._participant.getServiceStatus(), ServiceStatus.Status.BAD);
+ assertTrue(fixture._requests.stream().noneMatch(MetadataRpc.MembershipCommandRequest::hasReady));
+ }
+ }
+
+ @Test(timeOut = 10_000)
+ public void testSlowRoutingCannotBlockHeartbeatOrRepublishAnInvalidatedImage() throws Exception {
+ try (Fixture fixture = new Fixture(); var executor = Executors.newFixedThreadPool(2)) {
+ fixture._participant.start();
+ fixture._participant.refreshRouting();
+ CountDownLatch entered = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ doAnswer(invocation -> {
+ MetadataRpc.RoutingSnapshotRequest request = invocation.getArgument(0);
+ if (!request.getTableName().isEmpty()) {
+ entered.countDown();
+ assertTrue(release.await(3, TimeUnit.SECONDS));
+ }
+ return fixture.routingResponse(request);
+ }).when(fixture._routingConnection).routingSnapshot(any(), anyLong());
+ var refresh = executor.submit(fixture._participant::refreshRouting);
+ try {
+ assertTrue(entered.await(1, TimeUnit.SECONDS));
+ fixture._renewals.add("outer-not-ready");
+ executor.submit(fixture._participant::heartbeat).get(1, TimeUnit.SECONDS);
+ assertThrows(IllegalStateException.class, () -> fixture._participant.routingImage("orders_OFFLINE"));
+ } finally {
+ release.countDown();
+ }
+ refresh.get(1, TimeUnit.SECONDS);
+ assertThrows(IllegalStateException.class, () -> fixture._participant.routingImage("orders_OFFLINE"));
+ fixture._participant.heartbeat();
+ fixture._participant.refreshRouting();
+ assertEquals(fixture._participant.routingImage("orders_OFFLINE").segmentCount(), 0);
+ }
+ }
+
private static final class Fixture implements AutoCloseable {
private final String _history = UUID.randomUUID().toString();
private final String _id = "Broker_native_" + UUID.randomUUID();
private final Path _tls = Files.createTempFile("native-broker-test-", ".pem");
private final MetadataRpcConnection _connection = mock(MetadataRpcConnection.class);
+ private final MetadataRpcConnection _routingConnection = mock(MetadataRpcConnection.class);
private final PinotConfiguration _configuration;
private final NativeBrokerParticipant _participant;
private final List<MetadataRpc.MembershipCommandRequest> _requests = new ArrayList<>();
@@ -201,6 +262,9 @@ public class NativeBrokerParticipantTest {
private final MetadataRpc.MembershipControl _control;
private MetadataRpc.ParticipantToken _token;
private long _index;
+ private long _routingIndex = 100;
+ private List<String> _routingTables = List.of("orders_OFFLINE");
+ private boolean _routingFailure;
private boolean _grantCapabilities;
private Fixture() throws Exception {
@@ -256,7 +320,33 @@ public class NativeBrokerParticipantTest {
assertTrue(request.hasUnregister(), "Membership-only brokers must never set ready or advertise endpoints");
return response(ResultStatus.OK).build();
});
- _participant = new NativeBrokerParticipant(new NativeMetadataConfig(_configuration), _connection);
+ when(_routingConnection.routingSnapshot(any(), anyLong())).thenAnswer(invocation -> {
+ if (_routingFailure) {
+ throw Status.UNAVAILABLE.asRuntimeException();
+ }
+ return routingResponse(invocation.getArgument(0));
+ });
+ _participant = new NativeBrokerParticipant(new NativeMetadataConfig(_configuration), _connection,
+ _routingConnection);
+ }
+
+ private MetadataRpc.RoutingSnapshot routingResponse(MetadataRpc.RoutingSnapshotRequest request) {
+ assertEquals(request.getSession().getIncarnation(), _token.getIncarnation());
+ assertEquals(request.getSession().getRegistrationEpoch(), _token.getRegistrationEpoch());
+ var response = MetadataRpc.RoutingSnapshot.newBuilder().setStatus(ResultStatus.OK).setControl(_control)
+ .setParticipant(state()).setRevision(MetadataRpc.Revision.newBuilder().setHistoryId(_history)
+ .setIndex(++_routingIndex));
+ if (request.getTableName().isEmpty()) {
+ response.addAllTables(_routingTables);
+ } else {
+ response.setTable(MetadataRpc.TableRouting.newBuilder().setTableName(request.getTableName())
+ .setSchemaName("orders")
+ .setTable(MetadataRpc.ObjectVersion.newBuilder().setHistoryId(_history)
+ .setObjectId(UUID.randomUUID().toString()).setVersion(1))
+ .setSchema(MetadataRpc.ObjectVersion.newBuilder().setHistoryId(_history)
+ .setObjectId(UUID.randomUUID().toString()).setVersion(1)));
+ }
+ return response.build();
}
private MetadataRpc.MembershipResponse.Builder response(ResultStatus status) {
diff --git a/pinot-broker/src/test/java/org/apache/pinot/broker/routing/NativeBrokerRoutingImageTest.java b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/NativeBrokerRoutingImageTest.java
new file mode 100644
index 00000000000..80a4c46423d
--- /dev/null
+++ b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/NativeBrokerRoutingImageTest.java
@@ -0,0 +1,174 @@
+/**
+ * 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.broker.routing;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import org.apache.pinot.common.request.context.NativeQueryContext;
+import org.apache.pinot.common.request.context.NativeQueryDeadline;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.ParticipantRole;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.ParticipantToken;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.ReplicaRouting;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.RoutingSnapshot;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.SegmentRouting;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
+
+
+/// Ensures complete immutable fanout and rejects identities that could mix physical replicas across requests.
+public class NativeBrokerRoutingImageTest {
+ @Test
+ public void testCompleteFanoutPreservesPhysicalIdentityAndOneDeadline() throws Exception {
+ Fixture fixture = new Fixture();
+ ReplicaRouting first = fixture.replica("Server_a", "grpcs://server-a:8090");
+ ReplicaRouting second = fixture.replica("Server_b", "grpcs://server-b:8090");
+ SegmentRouting one = fixture.segment("one", first);
+ SegmentRouting two = fixture.segment("two", second);
+ RoutingSnapshot snapshot = fixture.snapshot(one, two);
+ NativeBrokerRoutingImage image = new NativeBrokerRoutingImage(snapshot, fixture._broker, "orders_OFFLINE");
+ UUID query = UUID.randomUUID();
+ UUID attempt = UUID.randomUUID();
+ NativeQueryDeadline deadline = new NativeQueryDeadline("qualified-test-clock", 100, 150);
+ var requests = image.requests(query, attempt, deadline);
+ assertEquals(requests.size(), 2);
+ Set<String> names = new HashSet<>();
+ for (var request : requests) {
+ NativeQueryContext context = request.context();
+ assertEquals(context.version(), NativeQueryContext.DEADLINE_VERSION);
+ assertEquals(context.queryId(), query);
+ assertEquals(context.attemptId(), attempt);
+ assertSame(context.deadline(), deadline);
+ assertTrue(context.optionalSegments().isEmpty());
+ assertTrue(names.addAll(context.requiredNames()));
+ assertEquals(context.broker().incarnation().toString(), fixture._broker.getIncarnation());
+ SegmentRouting source = context.requiredNames().contains("one") ? one : two;
+ ReplicaRouting replica = source.getReplicas(0);
+ assertEquals(context.server().incarnation().toString(), replica.getServer().getIncarnation());
+ assertEquals(context.segment(source.getName()).content().objectId().toString(),
+ source.getContent().getObjectId());
+ assertEquals(context.segment(source.getName()).assignment().objectId().toString(),
+ replica.getAssignment().getObjectId());
+ assertEquals(context.segment(source.getName()).materializationId().toString(), replica.getMaterializationId());
+ assertEquals(request.endpoint().toString(), replica.getEndpoint());
+ }
+ assertEquals(names, Set.of("one", "two"));
+ assertThrows(UnsupportedOperationException.class, () -> requests.clear());
+ // Mutating a protobuf builder cannot alter the captured image or already produced contexts.
+ RoutingSnapshot.Builder changed = snapshot.toBuilder();
+ changed.getTableBuilder().clearSegments();
+ assertEquals(image.segmentCount(), 2);
+ assertEquals(image.requests(query, attempt, deadline).getFirst().context().toProto(),
+ requests.getFirst().context().toProto());
+ }
+
+ @Test
+ public void testUninstalledUnapprovedAndEmptyTablesCannotProducePartialRequests() throws Exception {
+ Fixture fixture = new Fixture();
+ SegmentRouting installed = fixture.segment("installed",
+ fixture.replica("Server_a", "grpcs://server-a:8090"));
+ SegmentRouting missing = fixture.segment("missing");
+ var image = new NativeBrokerRoutingImage(fixture.snapshot(installed, missing), fixture._broker, "orders_OFFLINE");
+ assertEquals(image.segmentCount(), 2, "Missing replicas must remain represented in complete table metadata");
+ assertThrows(IllegalStateException.class, () -> image.requests(UUID.randomUUID(), UUID.randomUUID(),
+ new NativeQueryDeadline("test-clock", 100, 150)));
+ ReplicaRouting unapproved = fixture.replica("Server_b", "grpcs://server-b:8090").toBuilder()
+ .clearEndpoint().build();
+ var notEligible = new NativeBrokerRoutingImage(fixture.snapshot(fixture.segment("not-eligible", unapproved)),
+ fixture._broker, "orders_OFFLINE");
+ assertThrows(IllegalStateException.class, () -> notEligible.requests(UUID.randomUUID(), UUID.randomUUID(),
+ new NativeQueryDeadline("test-clock", 100, 150)));
+ var empty = new NativeBrokerRoutingImage(fixture.snapshot(), fixture._broker, "orders_OFFLINE");
+ assertThrows(IllegalStateException.class, () -> empty.requests(UUID.randomUUID(), UUID.randomUUID(),
+ new NativeQueryDeadline("test-clock", 100, 150)));
+ }
+
+ @Test
+ public void testCrossServerIdentityCollisionsAndEndpointConfusionAreRejected() {
+ Fixture fixture = new Fixture();
+ ReplicaRouting first = fixture.replica("Server_a", "grpcs://server-a:8090");
+ ReplicaRouting second = fixture.replica("Server_b", "grpcs://server-b:8090");
+ SegmentRouting one = fixture.segment("one", first);
+ SegmentRouting two = fixture.segment("two", second);
+ assertThrows(IllegalArgumentException.class, () -> new NativeBrokerRoutingImage(
+ fixture.snapshot(one, two.toBuilder().setContent(one.getContent()).build()), fixture._broker,
+ "orders_OFFLINE"));
+ assertThrows(IllegalArgumentException.class, () -> new NativeBrokerRoutingImage(fixture.snapshot(one,
+ two.toBuilder().setReplicas(0, second.toBuilder().setAssignment(first.getAssignment())).build()),
+ fixture._broker, "orders_OFFLINE"));
+ assertThrows(IllegalArgumentException.class, () -> new NativeBrokerRoutingImage(fixture.snapshot(one,
+ two.toBuilder().setReplicas(0, second.toBuilder().setMaterializationId(first.getMaterializationId())).build()),
+ fixture._broker, "orders_OFFLINE"));
+ assertThrows(IllegalArgumentException.class, () -> new NativeBrokerRoutingImage(fixture.snapshot(one,
+ two.toBuilder().setReplicas(0, second.toBuilder().setEndpoint(first.getEndpoint())).build()),
+ fixture._broker, "orders_OFFLINE"));
+ assertThrows(IllegalArgumentException.class, () -> new NativeBrokerRoutingImage(fixture.snapshot(one,
+ two.toBuilder().setReplicas(0, second.toBuilder().setServer(first.getServer())).build()),
+ fixture._broker, "orders_OFFLINE"));
+ }
+
+ private static final class Fixture {
+ private final String _history = UUID.randomUUID().toString();
+ private final ParticipantToken _broker = token("Broker_native", ParticipantRole.BROKER);
+ private final MetadataRpc.ObjectVersion _table = object();
+ private final MetadataRpc.ObjectVersion _schema = object();
+
+ private ParticipantToken token(String name, ParticipantRole role) {
+ return ParticipantToken.newBuilder().setHistoryId(_history).setParticipantId(name).setRole(role)
+ .setIncarnation(UUID.randomUUID().toString()).setRegistrationEpoch(1).build();
+ }
+
+ private MetadataRpc.ObjectVersion object() {
+ return MetadataRpc.ObjectVersion.newBuilder().setHistoryId(_history)
+ .setObjectId(UUID.randomUUID().toString()).setVersion(1).build();
+ }
+
+ private ReplicaRouting replica(String name, String endpoint) {
+ return ReplicaRouting.newBuilder().setServer(token(name, ParticipantRole.SERVER)).setAssignment(object())
+ .setAssignmentEpoch(1).setMaterializationId(UUID.randomUUID().toString())
+ .setInstallationId(UUID.randomUUID().toString()).setEndpoint(endpoint).build();
+ }
+
+ private SegmentRouting segment(String name, ReplicaRouting... replicas) {
+ return SegmentRouting.newBuilder().setName(name).setContent(object()).setContentSha256("a".repeat(64))
+ .addAllReplicas(List.of(replicas)).build();
+ }
+
+ private RoutingSnapshot snapshot(SegmentRouting... segments) {
+ var revision = MetadataRpc.Revision.newBuilder().setHistoryId(_history).setIndex(10);
+ var control = MetadataRpc.MembershipControl.newBuilder().setControlEpoch(MetadataRpc.ControlToken.newBuilder()
+ .setHistoryId(_history).setRaftTerm(1).setBarrierIndex(1).setLeaderId("controller")
+ .setActivationId(UUID.randomUUID().toString())).setClockBarrierIndex(1).setMinimumTickDelayMillis(100)
+ .setMaxConfiguredPermitLifetimeMillis(200).setPolicy(MetadataRpc.MembershipPolicy.newBuilder().setVersion(1)
+ .setMinTickIntervalMillis(100).setLeaseRounds(3).setPermitLifetimeMillis(200)
+ .setMaxQueryLifetimeMillis(300).setMaxRevisionLag(100));
+ return RoutingSnapshot.newBuilder().setStatus(MetadataRpc.ResultStatus.OK).setRevision(revision)
+ .setParticipant(MetadataRpc.ParticipantState.newBuilder().setParticipant(_broker).setExpiryRound(3)
+ .setLastModifiedIndex(1).setAppliedRevision(revision))
+ .setControl(control).setTable(MetadataRpc.TableRouting.newBuilder().setTableName("orders_OFFLINE")
+ .setTable(_table).setSchemaName("orders").setSchema(_schema).addAllSegments(List.of(segments))).build();
+ }
+ }
+}
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/coordination/MetadataRpcClient.java b/pinot-common/src/main/java/org/apache/pinot/common/coordination/MetadataRpcClient.java
index c04f5aa2010..da3c838071e 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/coordination/MetadataRpcClient.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/coordination/MetadataRpcClient.java
@@ -27,6 +27,7 @@ import java.io.IOException;
import java.nio.file.Path;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
+import org.apache.pinot.coordination.ratis.rpc.BrokerRoutingServiceGrpc;
import org.apache.pinot.coordination.ratis.rpc.MetadataRpc;
import org.apache.pinot.coordination.ratis.rpc.MetadataServiceGrpc;
import org.apache.pinot.coordination.ratis.rpc.MinionTaskServiceGrpc;
@@ -45,11 +46,13 @@ public class MetadataRpcClient implements AutoCloseable {
public static final int MAX_MEMBERSHIP_BYTES = 64 * 1024;
public static final int MAX_INSTALLATION_REQUEST_BYTES = 64 * 1024;
public static final int MAX_TASK_REQUEST_BYTES = 64 * 1024;
+ public static final int MAX_ROUTING_REQUEST_BYTES = 64 * 1024;
private final ManagedChannel _channel;
private final MetadataServiceGrpc.MetadataServiceBlockingStub _stub;
private final ParticipantMembershipServiceGrpc.ParticipantMembershipServiceBlockingStub _membership;
private final ServerInstallationServiceGrpc.ServerInstallationServiceBlockingStub _installation;
private final MinionTaskServiceGrpc.MinionTaskServiceBlockingStub _task;
+ private final BrokerRoutingServiceGrpc.BrokerRoutingServiceBlockingStub _routing;
private final long _timeoutMillis;
public MetadataRpcClient(String host, int port, Path certificateChain, Path privateKey, Path trustCertificates,
@@ -67,6 +70,22 @@ public class MetadataRpcClient implements AutoCloseable {
_membership = ParticipantMembershipServiceGrpc.newBlockingStub(_channel);
_installation = ServerInstallationServiceGrpc.newBlockingStub(_channel);
_task = MinionTaskServiceGrpc.newBlockingStub(_channel);
+ _routing = BrokerRoutingServiceGrpc.newBlockingStub(_channel);
+ }
+
+ public MetadataRpc.RoutingSnapshot routingSnapshot(MetadataRpc.RoutingSnapshotRequest request) {
+ return routingSnapshot(request, _timeoutMillis);
+ }
+
+ public MetadataRpc.RoutingSnapshot routingSnapshot(MetadataRpc.RoutingSnapshotRequest request, long timeoutMillis) {
+ if (request == null || request.getSerializedSize() > MAX_ROUTING_REQUEST_BYTES) {
+ throw Status.RESOURCE_EXHAUSTED.withDescription("Routing request exceeds transport byte limit")
+ .asRuntimeException();
+ }
+ if (timeoutMillis <= 0) {
+ throw Status.DEADLINE_EXCEEDED.withDescription("Routing request deadline expired").asRuntimeException();
+ }
+ return _routing.withDeadlineAfter(Math.min(_timeoutMillis, timeoutMillis), TimeUnit.MILLISECONDS).snapshot(request);
}
public MetadataRpc.CommandResult taskExecute(MetadataRpc.TaskCommandRequest request, long timeoutMillis) {
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/coordination/MetadataRpcConnection.java b/pinot-common/src/main/java/org/apache/pinot/common/coordination/MetadataRpcConnection.java
index 69aeb714ab9..3561788e179 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/coordination/MetadataRpcConnection.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/coordination/MetadataRpcConnection.java
@@ -78,6 +78,10 @@ public class MetadataRpcConnection implements AutoCloseable {
return call(timeoutMillis, (client, timeout) -> client.membershipRead(request, timeout));
}
+ public MetadataRpc.RoutingSnapshot routingSnapshot(MetadataRpc.RoutingSnapshotRequest request, long timeoutMillis) {
+ return call(timeoutMillis, (client, timeout) -> client.routingSnapshot(request, timeout));
+ }
+
public MetadataRpc.MembershipResponse membershipExecute(MetadataRpc.MembershipCommandRequest request,
long timeoutMillis) {
return call(timeoutMillis, (client, timeout) -> client.membershipExecute(request, timeout));
diff --git a/pinot-common/src/main/proto/metadata.proto b/pinot-common/src/main/proto/metadata.proto
index cc7fdf1e78a..ed2e38fa486 100644
--- a/pinot-common/src/main/proto/metadata.proto
+++ b/pinot-common/src/main/proto/metadata.proto
@@ -57,6 +57,56 @@ service MinionTaskService {
rpc Outcome(TaskOutcomeRequest) returns (CommandResult);
}
+// BROKER-only complete routing observations. TLS policy selects the stable identity; committed policy authorizes
+// its table tenants. Responses contain identities and installation evidence, never catalog records or storage URIs.
+service BrokerRoutingService {
+ rpc Snapshot(RoutingSnapshotRequest) returns (RoutingSnapshot);
+}
+
+message RoutingSnapshotRequest {
+ string history_id = 1;
+ ParticipantSession session = 2;
+ // Empty means discovery of authorized table names.
+ string table_name = 3;
+ int32 max_records = 4;
+ int32 max_bytes = 5;
+}
+
+message RoutingSnapshot {
+ ResultStatus status = 1;
+ Revision revision = 2;
+ string detail = 3;
+ ParticipantState participant = 4;
+ MembershipControl control = 5;
+ repeated string tables = 6;
+ TableRouting table = 7;
+}
+
+message TableRouting {
+ string table_name = 1;
+ ObjectVersion table = 2;
+ string schema_name = 3;
+ ObjectVersion schema = 4;
+ repeated SegmentRouting segments = 5;
+}
+
+message SegmentRouting {
+ string name = 1;
+ ObjectVersion content = 2;
+ string content_sha256 = 3;
+ repeated ReplicaRouting replicas = 4;
+}
+
+message ReplicaRouting {
+ ParticipantToken server = 1;
+ ObjectVersion assignment = 2;
+ int64 assignment_epoch = 3;
+ string materialization_id = 4;
+ string installation_id = 5;
+ // Present only for an approved advertised query endpoint with the required current server capability.
+ optional string endpoint = 6;
+}
+
enum RecordKind {
RECORD_KIND_UNSPECIFIED = 0;
CLUSTER_CONFIG = 1;
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeControllerMetadata.java b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeControllerMetadata.java
index 71022e6e3fe..2312dccff87 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeControllerMetadata.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeControllerMetadata.java
@@ -54,6 +54,7 @@ import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Result;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Revision;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Status;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.StreamTipResult;
+import org.apache.pinot.coordination.ratis.metadata.NativeBrokerRoutingPolicy;
import org.apache.pinot.coordination.ratis.metadata.NativeReplicaPlacementPolicy;
import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol;
import org.apache.pinot.coordination.ratis.metadata.OperationProtocol;
@@ -354,15 +355,20 @@ public final class NativeControllerMetadata implements PinotClusterConfigProvide
if (placement != null) {
NativeReplicaPlacementPolicy.parse(placement);
}
+ String routing = values.get(NativeBrokerRoutingPolicy.CONFIG_KEY);
+ if (routing != null) {
+ NativeBrokerRoutingPolicy.parse(routing);
+ }
if (!snapshot.nativeTables().isEmpty()) {
Set<String> changed = new HashSet<>(snapshot.clusterConfig().keySet());
changed.addAll(values.keySet());
changed.removeIf(key -> Objects.equals(snapshot.clusterConfig().get(key), values.get(key)));
- // Endpoint approvals are rechecked before observations return. The native output reconciler validates each
- // placement policy change in replicated apply; unrelated settings retain the existing lifecycle gate.
+ // Endpoint and broker routing approvals are rechecked by observations. The native output reconciler validates
+ // each placement policy change in replicated apply; unrelated settings retain the existing lifecycle gate.
if (changed.isEmpty() || changed.stream().anyMatch(
key -> !key.startsWith(NativeTableSizeMetadataProvider.ENDPOINT_APPROVAL_PREFIX)
- && !NativeReplicaPlacementPolicy.CONFIG_KEY.equals(key))) {
+ && !NativeReplicaPlacementPolicy.CONFIG_KEY.equals(key)
+ && !NativeBrokerRoutingPolicy.CONFIG_KEY.equals(key))) {
throw new UnsupportedOperationException("Cluster config changes require native table reconciliation");
}
}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/coordination/NativeControllerResourceTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/coordination/NativeControllerResourceTest.java
index 135567829bb..8a4c2341101 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/coordination/NativeControllerResourceTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/coordination/NativeControllerResourceTest.java
@@ -49,6 +49,7 @@ import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.FeedCursor;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.FeedScope;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.RecordKind;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Revision;
+import org.apache.pinot.coordination.ratis.metadata.NativeBrokerRoutingPolicy;
import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.Content;
import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.LoaderDescriptor;
import org.apache.pinot.coordination.ratis.transport.MetadataAccess;
@@ -132,13 +133,27 @@ public class NativeControllerResourceTest {
HttpHeaders clusterHeaders = headers(null);
when(clusterHeaders.getHeaderString(HttpHeaders.IF_MATCH)).thenReturn(clusterTag);
assertStatus(501, () -> resource.updateClusterConfig("{}", clusterHeaders));
- // Read-only endpoint approvals may rotate while segment state exists; loader settings still require reconcile.
+ // Observation approvals may rotate or be revoked while segment state exists; loader settings require reconcile.
Map<String, String> approved = new HashMap<>(fixture._metadata.snapshot().clusterConfig());
approved.put(NativeTableSizeMetadataProvider.ENDPOINT_APPROVAL_PREFIX + "Server_native", "https://host:8443");
fixture._metadata.validateClusterConfig(fixture._metadata.snapshot(), approved);
- approved.put("loader.setting", "changed");
+ approved.put(NativeBrokerRoutingPolicy.CONFIG_KEY,
+ "{\"brokers\":{\"Broker_native\":[\"DefaultTenant\"]},\"servers\":{}}");
+ fixture._metadata.validateClusterConfig(fixture._metadata.snapshot(), approved);
+ fixture.config(approved, 11);
+ Map<String, String> revoked = new HashMap<>(approved);
+ revoked.put(NativeBrokerRoutingPolicy.CONFIG_KEY, "{\"brokers\":{},\"servers\":{}}");
+ fixture._metadata.validateClusterConfig(fixture._metadata.snapshot(), revoked);
+ Map<String, String> malformed = new HashMap<>(revoked);
+ malformed.put(NativeBrokerRoutingPolicy.CONFIG_KEY, "{}");
+ assertThrows(IllegalArgumentException.class,
+ () -> fixture._metadata.validateClusterConfig(fixture._metadata.snapshot(), malformed));
+ fixture.config(revoked, 12);
+ revoked.remove(NativeBrokerRoutingPolicy.CONFIG_KEY);
+ fixture._metadata.validateClusterConfig(fixture._metadata.snapshot(), revoked);
+ revoked.put("loader.setting", "changed");
assertThrows(UnsupportedOperationException.class,
- () -> fixture._metadata.validateClusterConfig(fixture._metadata.snapshot(), approved));
+ () -> fixture._metadata.validateClusterConfig(fixture._metadata.snapshot(), revoked));
}
}
diff --git a/pinot-coordination-ratis/CONTROLLER-API-PARITY.md b/pinot-coordination-ratis/CONTROLLER-API-PARITY.md
index f8e72652a06..a4b83d4f87c 100644
--- a/pinot-coordination-ratis/CONTROLLER-API-PARITY.md
+++ b/pinot-coordination-ratis/CONTROLLER-API-PARITY.md
@@ -36,6 +36,9 @@ The standard resource package has **264 annotated HTTP methods across 43 resourc
has bounded automatic replica placement from explicit administrator-owned tenant/domain policy, durable installation
intent and exact server receipts. The task API reports publication, desired and installed replicas separately.
This does not provide replacement/deletion, general rebalance, physical cleanup or query-serving qualification.
+- Native broker routing: authenticated, tenant-scoped table discovery and complete OFFLINE routing observations,
+ immutable broker images and query-identity construction. Query planning, dispatch and serving activation remain
+ outstanding; installed replicas without query authority are not selectable.
- Outstanding: remaining offline lifecycle, full existing task API/plugin parity, realtime,
remaining feature families and query-serving qualification. A full replacement is not yet implemented.
- ZooKeeper tree APIs are not applicable to the replacement; a mutable compatibility tree will not be introduced.
diff --git a/pinot-coordination-ratis/README.md b/pinot-coordination-ratis/README.md
index c4d11a9dc64..f91706b7f9b 100644
--- a/pinot-coordination-ratis/README.md
+++ b/pinot-coordination-ratis/README.md
@@ -362,11 +362,55 @@ content must match. Logical-table acquisition also releases earlier handles when
Native names-only, realtime and logical queries remain rejected. A native request with no acquired content cannot
yet obtain a version-checked schema-only catalog snapshot and is rejected instead of returning an empty scan.
-Normal broker production, pinned routing images, feature-level negotiation, authenticated process admission,
-production clock qualification, broker deadline propagation, transport response-delivery fencing and whole-query
-retry/drain remain unimplemented. MSE/time-series and other unsupported paths cannot silently drop the context. Native
+Broker routing observations and immutable identity construction are described below. Normal query planning and
+dispatch, feature-level negotiation, authenticated process admission, production clock qualification, broker
+deadline origination, transport response-delivery fencing and whole-query retry/drain remain unimplemented.
+MSE/time-series and other unsupported paths cannot silently drop the context. Native
`ServerInstance` query resources remain disabled until those gates are implemented and tested together.
+## Broker routing observations
+
+The metadata endpoint exposes a separate, read-only `BrokerRoutingService.Snapshot` RPC to certificate-bound BROKER
+sessions. An empty table name discovers authorized OFFLINE tables; a canonical table name captures one complete
+table image after a consensus read barrier. Bootstrap observation requires a live session, not query readiness or
+`BROKER_QUERY` authority. Other participant roles cannot use this endpoint.
+
+Configure the cluster key `pinot.coordination.ratis.broker.routing` with a JSON string such as:
+
+```json
+{
+ "brokers": {"Broker_native": ["DefaultTenant"]},
+ "servers": {"Server_native": "grpcs://server.example:8090"}
+}
+```
+
+Broker IDs are mapped explicitly to allowed broker tenants. Server IDs are mapped to canonical TLS query addresses.
+There are no wildcard or implicit policy grants. Normal omitted table tenants retain Pinot's `DefaultTenant`
+convention. Policy changes take effect through subsequent committed reads; participant advertisements cannot grant
+tenant access or destination approval.
+
+Each table image carries exact table/schema versions and every published immutable segment, including segments
+with zero installed replicas. Installed candidates require current configuration and assignment identities, a live
+ready server incarnation, placement-policy eligibility and matching installation evidence from the durable
+reconciliation operation. A completed proof does not expire merely because its former controller owner expires.
+Archive locations, loader configuration, task configuration and unrelated cluster metadata are not exposed.
+
+A candidate includes a query destination only if its server also has `SERVER_QUERY`, advertises the exact approved
+`queryGrpc` address and declares `nativeQueryVersion=2`. This is routing metadata, not a serving grant or an
+authenticated query handshake. Normal server starters still advertise neither query capability nor query endpoints.
+Consequently their installed replicas are visible as observations but cannot be selected for query dispatch.
+
+Snapshots allow at most 4,096 logical records and 2 MiB, rejecting excessive counts without returning partial
+routing data. The broker bounds its cache to 128 tables and 8 MiB and uses a separate connection and polling lane
+from membership heartbeats. It validates immutable per-table images and invalidates cached selection on refresh
+failure. Discovery and per-table revisions are not a cluster-wide applied watermark.
+The selector requires coverage for every published segment and constructs per-server `NativeQueryContext` values
+using one caller-supplied deadline for the whole attempt. Missing coverage and empty-table execution remain
+unsupported; no fallback can silently turn required segments into optional ones.
+
+These observations and contexts do not authorize queries. Connecting the normal query handlers requires the
+remaining serving-permit, routing-revision, transport-identity, clock and response-delivery checks together.
+
## Normal starters (experimental, query listeners disabled)
`StartController`, `StartBroker`, and `StartServer` use their existing starter classes. Set
@@ -382,13 +426,14 @@ selection, process identity, peers, TLS and local storage are bootstrap settings
data directory, metadata hydration and installation reconciliation. It requests no query capabilities and advertises
no query endpoints. The existing server query-start gate stays closed.
- Broker registers and renews an empty-capability membership session. It remains `ready=false`, with service status
- explicitly reporting that routing and query listeners are unavailable. Its applied revision covers only its own
- membership/control projection, not a routing image.
+ reporting routing-cache observations and disabled query listeners. Its membership applied revision covers only
+ its own participant/control projection, not a routing image. Routing snapshots retain their own table revisions.
No broker HTTP/gRPC/MSE or server Netty/TLS/gRPC/MSE query listeners are allocated, even if legacy query-port settings
are present. Helix-dependent REST resources are not started on the native path. An explicitly configured server admin listener
-serves the native read-only size endpoint described below. Minions
-and multi-cluster brokers are unsupported. These entry points exercise coordination and offline installation; they
+serves the native read-only size endpoint described below. The native minion path is described in
+[MINION-TASK-PLAN.md](MINION-TASK-PLAN.md); multi-cluster brokers remain unsupported.
+These entry points exercise coordination and offline installation; they
do not yet provide a queryable replacement cluster.
All native components need `pinot.coordination.ratis.historyId`. Controllers retain the voter settings in the next
diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/MetadataStateMachine.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/MetadataStateMachine.java
index 27c38caf755..8129e266cc5 100644
--- a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/MetadataStateMachine.java
+++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/MetadataStateMachine.java
@@ -34,6 +34,7 @@ import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.ChangePage;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.FeedCursor;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.FeedScope;
import org.apache.pinot.coordination.ratis.metadata.OfflineInstallationProtocol.InstallationSnapshot;
+import org.apache.pinot.coordination.ratis.metadata.OfflineRoutingProtocol.RoutingSnapshot;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.WorkSnapshot;
import org.apache.ratis.proto.RaftProtos.LogEntryProto;
import org.apache.ratis.protocol.Message;
@@ -175,6 +176,12 @@ final class MetadataStateMachine extends BaseStateMachine {
return _catalog.installationSnapshot(server, maxRecords, maxBytes);
}
+ synchronized RoutingSnapshot routingSnapshot(ParticipantToken broker, String tableName, int maxRecords, int maxBytes,
+ long minimumIndex, long deadlineNanos) throws IOException {
+ awaitApplied(minimumIndex, deadlineNanos);
+ return _catalog.routingSnapshot(broker, tableName, maxRecords, maxBytes);
+ }
+
synchronized WorkSnapshot taskSnapshot(ParticipantToken minion, int maxRecords, int maxBytes, long minimumIndex,
long deadlineNanos) throws IOException {
awaitApplied(minimumIndex, deadlineNanos);
diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/RatisMetadataAccess.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/RatisMetadataAccess.java
index 4ba416cf0dc..96d4e488607 100644
--- a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/RatisMetadataAccess.java
+++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/RatisMetadataAccess.java
@@ -40,10 +40,12 @@ import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Status;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.StreamTipResult;
import org.apache.pinot.coordination.ratis.metadata.OfflineInstallationProtocol;
import org.apache.pinot.coordination.ratis.metadata.OfflineInstallationProtocol.InstallationSnapshot;
+import org.apache.pinot.coordination.ratis.metadata.OfflineRoutingProtocol.RoutingSnapshot;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.WorkSnapshot;
import org.apache.pinot.coordination.ratis.transport.InstallationAccess;
import org.apache.pinot.coordination.ratis.transport.MembershipAccess;
import org.apache.pinot.coordination.ratis.transport.MetadataAccess;
+import org.apache.pinot.coordination.ratis.transport.RoutingAccess;
import org.apache.pinot.coordination.ratis.transport.TaskAccess;
@@ -51,7 +53,7 @@ import org.apache.pinot.coordination.ratis.transport.TaskAccess;
/// coherent capture; no full catalog image crosses the Raft client transport. The enclosing service authorizes actors.
/// This class is thread-safe and owns its clients, but does not own the voter.
public final class RatisMetadataAccess
- implements MetadataAccess, MembershipAccess, InstallationAccess, TaskAccess, AutoCloseable {
+ implements MetadataAccess, MembershipAccess, InstallationAccess, TaskAccess, RoutingAccess, AutoCloseable {
private final MetadataStateMachine _stateMachine;
private final RatisMetadataStore.Client _client;
private final RatisMetadataStore.Client _membershipClient;
@@ -128,6 +130,14 @@ public final class RatisMetadataAccess
return _stateMachine.installationSnapshot(server, maxRecords, maxBytes, index, deadline);
}
+ @Override
+ public RoutingSnapshot routingSnapshot(ParticipantToken broker, String tableName, int maxRecords, int maxBytes,
+ long timeoutMillis) throws IOException {
+ long deadline = deadline(timeoutMillis);
+ long index = readBarrier(deadline);
+ return _stateMachine.routingSnapshot(broker, tableName, maxRecords, maxBytes, index, deadline);
+ }
+
@Override
public WorkSnapshot taskSnapshot(ParticipantToken minion, int maxRecords, int maxBytes, long timeoutMillis)
throws IOException {
diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/MetadataCatalog.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/MetadataCatalog.java
index b91569187a5..c1b9ed0ae4b 100644
--- a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/MetadataCatalog.java
+++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/MetadataCatalog.java
@@ -33,7 +33,11 @@ import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.UUID;
+import javax.annotation.Nullable;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ControlState;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ParticipantState;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ParticipantToken;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Role;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Acknowledge;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Action;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.CatalogImage;
@@ -78,7 +82,12 @@ import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.TaskSession
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Transaction;
import org.apache.pinot.coordination.ratis.metadata.OfflineInstallationProtocol.InstallationSnapshot;
import org.apache.pinot.coordination.ratis.metadata.OfflineInstallationProtocol.WorkResult;
+import org.apache.pinot.coordination.ratis.metadata.OfflineRoutingProtocol.ReplicaRouting;
+import org.apache.pinot.coordination.ratis.metadata.OfflineRoutingProtocol.RoutingSnapshot;
+import org.apache.pinot.coordination.ratis.metadata.OfflineRoutingProtocol.SegmentRouting;
+import org.apache.pinot.coordination.ratis.metadata.OfflineRoutingProtocol.TableRouting;
import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.Assignment;
+import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.Content;
import org.apache.pinot.coordination.ratis.metadata.OperationProtocol.AssignedOfflineSegmentReload;
import org.apache.pinot.coordination.ratis.metadata.OperationProtocol.OperationState;
import org.apache.pinot.coordination.ratis.metadata.OperationProtocol.State;
@@ -397,6 +406,242 @@ public final class MetadataCatalog {
return nextBytes;
}
+ /// A complete table observation at one committed revision. Discovery is broker-tenant scoped; individual images
+ /// retain segments without replicas so missing installations cannot silently remove data from a query plan.
+ public synchronized RoutingSnapshot routingSnapshot(ParticipantToken broker, String tableName, int maxRecords,
+ int maxBytes) throws IOException {
+ try {
+ require(tableName != null && maxRecords > 0 && maxRecords <= OfflineRoutingProtocol.MAX_SNAPSHOT_RECORDS
+ && maxBytes >= OfflineRoutingProtocol.MIN_SNAPSHOT_BYTES
+ && maxBytes <= OfflineRoutingProtocol.MAX_SNAPSHOT_BYTES, Status.INVALID_REQUEST,
+ "Invalid routing snapshot bounds or table name");
+ if (!tableName.isEmpty()) {
+ validateRoutingTableName(tableName);
+ }
+ OperationCatalog.View view = coordinationView();
+ require(broker != null && _historyId.equals(broker.historyId()), Status.HISTORY_MISMATCH,
+ "Routing participant history differs");
+ require(broker.role() == Role.BROKER && broker.incarnation() != null && broker.registrationEpoch() > 0,
+ Status.FENCED, "Routing requires a complete BROKER registration");
+ OfflineSegmentCatalog.identifier(broker.participantId(), 1024);
+ ParticipantState participant = view.participant(broker.participantId());
+ ControlState control = view.control();
+ require(participant != null && participant.participant().equals(broker) && !participant.retired()
+ && !participant.draining() && control != null && control.controlEpoch() != null && control.policy() != null
+ && participant.expiryRound() > control.round(), Status.FENCED,
+ "Routing participant is replaced, retired or expired");
+ CatalogRecord cluster = view.record(RecordKind.CLUSTER_CONFIG, MetadataProtocol.CLUSTER_CONFIG_NAME);
+ NativeBrokerRoutingPolicy policy;
+ try {
+ policy = NativeBrokerRoutingPolicy.parse(routingPolicyValue(cluster, NativeBrokerRoutingPolicy.CONFIG_KEY));
+ } catch (IllegalArgumentException e) {
+ throw new Rejection(Status.NOT_READY, "Native broker routing policy is absent or invalid");
+ }
+ require(policy.brokers().containsKey(broker.participantId()), Status.FENCED,
+ "Broker has no routing tenant grant");
+ RoutingSnapshot header = new RoutingSnapshot(Status.OK, revision(_lastAppliedIndex), "", participant, control,
+ List.of(), null);
+ RoutingProjection projection = new RoutingProjection(header, maxRecords, maxBytes);
+ if (tableName.isEmpty()) {
+ List<String> tables = new ArrayList<>();
+ for (UUID objectId : _names.values()) {
+ CatalogRecord record = _records.get(objectId);
+ if (record.kind() == RecordKind.TABLE_CONFIG) {
+ TableConfig table = JsonUtils.jsonNodeToObject(record.value(), TableConfig.class);
+ if (table.getTableType() == TableType.OFFLINE && policy.permits(broker.participantId(), table)) {
+ validateRoutingTableName(record.name());
+ projection.add(record.name(), 1);
+ tables.add(record.name());
+ }
+ }
+ }
+ tables.sort(Comparator.naturalOrder());
+ return new RoutingSnapshot(Status.OK, header.revision(), "", participant, control, tables, null);
+ }
+ CatalogRecord table = view.record(RecordKind.TABLE_CONFIG, tableName);
+ require(table != null && !table.deleted(), Status.NOT_FOUND, "Routing table is absent");
+ TableConfig config = JsonUtils.jsonNodeToObject(table.value(), TableConfig.class);
+ require(policy.permits(broker.participantId(), config), Status.FENCED, "Broker lacks the table tenant grant");
+ NativeReplicaPlacementPolicy placement;
+ try {
+ NativeReplicaPlacementPolicy.validateTable(config);
+ placement = NativeReplicaPlacementPolicy.parse(routingPolicyValue(cluster,
+ NativeReplicaPlacementPolicy.CONFIG_KEY));
+ } catch (IllegalArgumentException e) {
+ throw new Rejection(Status.NOT_READY, "Native table placement or its policy is unsupported or unavailable");
+ }
+ String schemaName = TableNameBuilder.extractRawTableName(tableName);
+ CatalogRecord schema = view.record(RecordKind.SCHEMA, schemaName);
+ require(schema != null && !schema.deleted(), Status.NOT_READY, "Routing schema is absent");
+ TableRouting tableHeader = new TableRouting(tableName, routingVersion(table), schemaName,
+ routingVersion(schema), List.of());
+ projection.add(tableHeader, 2);
+ List<SegmentRouting> segments = new ArrayList<>();
+ String contentPrefix = table.objectId() + "/";
+ String serverTenant = NativeReplicaPlacementPolicy.tenant(config);
+ List<String> approvedServers = placement.servers().keySet().stream().sorted().toList();
+ for (UUID objectId : _names.values()) {
+ CatalogRecord record = _records.get(objectId);
+ if (record.kind() != RecordKind.OFFLINE_SEGMENT_CONTENT || !record.name().startsWith(contentPrefix)) {
+ continue;
+ }
+ Content content = OfflineSegmentProtocol.decodeContent(record.value());
+ if (!content.tableObjectId().equals(table.objectId())) {
+ continue;
+ }
+ require(content.tableName().equals(tableName) && content.segmentObjectId().equals(record.objectId()),
+ Status.NOT_READY, "Routing content identity differs from its table");
+ SegmentRouting segment = new SegmentRouting(content.segmentName(), routingVersion(record),
+ content.contentSha256(), List.of());
+ projection.add(segment, 1);
+ List<ReplicaRouting> replicas = new ArrayList<>();
+ for (String serverId : approvedServers) {
+ if (!placement.servers().get(serverId).tenant().equals(serverTenant)) {
+ continue;
+ }
+ ReplicaRouting replica = routingReplica(view, policy, content, record, table, schema, serverId);
+ if (replica != null) {
+ projection.add(replica, 1);
+ replicas.add(replica);
+ }
+ }
+ segments.add(new SegmentRouting(segment.name(), segment.content(), segment.contentSha256(), replicas));
+ }
+ segments.sort(Comparator.comparing(SegmentRouting::name));
+ return new RoutingSnapshot(Status.OK, header.revision(), "", participant, control, List.of(),
+ new TableRouting(tableName, tableHeader.table(), schemaName, tableHeader.schema(), segments));
+ } catch (Rejection e) {
+ return routingSnapshotError(e._status, e.getMessage());
+ } catch (CoordinationRejection e) {
+ return routingSnapshotError(e.status(), e.getMessage());
+ }
+ }
+
+ private static void validateRoutingTableName(String tableName) {
+ OfflineSegmentCatalog.identifier(tableName, 1024);
+ String raw = TableNameBuilder.extractRawTableName(tableName);
+ int separator = raw.indexOf('.');
+ require(TableNameBuilder.getTableTypeFromTableName(tableName) == TableType.OFFLINE
+ && TableNameBuilder.getTableTypeFromTableName(raw) == null
+ && tableName.chars().noneMatch(Character::isWhitespace) && !raw.isEmpty()
+ && (separator < 0 || separator > 0 && separator < raw.length() - 1 && separator == raw.lastIndexOf('.')),
+ Status.INVALID_REQUEST, "Routing requires a canonical database-qualified OFFLINE table name");
+ }
+
+ @Nullable
+ private static String routingPolicyValue(@Nullable CatalogRecord cluster, String key) {
+ JsonNode value = cluster == null ? null : cluster.value().get(key);
+ return value != null && value.isTextual() ? value.textValue() : null;
+ }
+
+ private ObjectVersion routingVersion(CatalogRecord record) {
+ return new ObjectVersion(_historyId, record.objectId(), record.version());
+ }
+
+ private RoutingSnapshot routingSnapshotError(Status status, String detail) {
+ return new RoutingSnapshot(status, revision(_lastAppliedIndex), detail, null, null, List.of(), null);
+ }
+
+ @Nullable
+ private ReplicaRouting routingReplica(OperationCatalog.View view, NativeBrokerRoutingPolicy policy, Content content,
+ CatalogRecord contentRecord, CatalogRecord table, CatalogRecord schema, String serverId) throws IOException {
+ ParticipantState server = view.participant(serverId);
+ if (server == null || server.participant().role() != Role.SERVER || !server.ready() || server.draining()
+ || server.retired() || server.expiryRound() <= view.control().round()) {
+ return null;
+ }
+ CatalogRecord assignmentRecord = view.record(RecordKind.OFFLINE_REPLICA_ASSIGNMENT,
+ OfflineSegmentProtocol.assignmentName(content.segmentObjectId(), serverId));
+ if (assignmentRecord == null || assignmentRecord.deleted()) {
+ return null;
+ }
+ Assignment assignment = OfflineSegmentProtocol.decodeAssignment(assignmentRecord.value());
+ if (!assignment.desiredOnline() || !assignment.server().equals(server.participant())
+ || !assignment.tableName().equals(content.tableName())
+ || !assignment.tableObjectId().equals(content.tableObjectId())
+ || !assignment.segmentName().equals(content.segmentName())
+ || !assignment.segmentObjectId().equals(content.segmentObjectId())
+ || !assignment.contentSha256().equals(content.contentSha256())
+ || !assignment.compatibility().table().equals(routingVersion(table))
+ || !assignment.compatibility().schema().equals(routingVersion(schema))) {
+ return null;
+ }
+ // This slice recognizes the durable normal-placement reconcile intent only. Arbitrary retained reload receipts
+ // cannot introduce routing candidates; future lifecycle work must define its authoritative proof selection.
+ UUID operationId = NativeReplicaPlacementPolicy.operationId(_historyId, assignment.materializationId());
+ CatalogRecord operationRecord = view.record(RecordKind.CONTROLLER_OPERATION, operationId.toString());
+ if (operationRecord == null || operationRecord.deleted()) {
+ return null;
+ }
+ OperationState operation = OperationProtocol.decodeOperation(operationRecord.value());
+ if (!(operation.input() instanceof AssignedOfflineSegmentReload assigned)
+ || (operation.status() != State.RUNNING && operation.status() != State.SUCCEEDED)
+ || operation.owner() == null || operation.attempt() <= 0) {
+ return null;
+ }
+ var reload = assigned.reload();
+ var target = new OperationProtocol.TargetMaterialization(assignment.server(), assignment.assignmentEpoch(),
+ assignment.materializationId());
+ List<ObjectVersion> dependencies = List.of(routingVersion(contentRecord), routingVersion(table),
+ routingVersion(schema), routingVersion(assignmentRecord));
+ if (!reload.reconcile() || !reload.tableName().equals(content.tableName())
+ || !reload.tableObjectId().equals(content.tableObjectId())
+ || !reload.segmentName().equals(content.segmentName())
+ || !reload.segmentObjectId().equals(content.segmentObjectId())
+ || !reload.contentSha256().equals(content.contentSha256()) || !reload.targets().equals(List.of(target))
+ || !assigned.dependencies().containsAll(dependencies) || !operation.dependencies().containsAll(dependencies)) {
+ return null;
+ }
+ CatalogRecord receiptRecord = view.record(RecordKind.OFFLINE_INSTALLATION,
+ OfflineInstallationProtocol.recordName(operationId, operation.attempt(), serverId));
+ if (receiptRecord == null || receiptRecord.deleted()) {
+ return null;
+ }
+ var receipt = OfflineInstallationProtocol.decodeReceipt(receiptRecord.value());
+ var report = receipt.installation();
+ var expected = new OfflineInstallationProtocol.Materialization(content.tableName(), content.tableObjectId(),
+ content.segmentName(), content.segmentObjectId(), content.contentSha256(), server.participant(),
+ routingVersion(assignmentRecord), assignment.materializationId(), assignment.compatibility());
+ if (!_historyId.equals(receipt.historyId()) || !operationId.equals(report.operationId())
+ || operation.attempt() != report.attempt() || !operation.owner().equals(report.owner())
+ || !expected.equals(report.installed()) || operation.status() == State.SUCCEEDED
+ && (operation.completionProof() == null
+ || !operation.completionProof().installations().contains(routingVersion(receiptRecord)))) {
+ return null;
+ }
+ // A completed proof outlives the former controller owner's lease. Current assignment/server fences still apply.
+ return new ReplicaRouting(server.participant(), routingVersion(assignmentRecord), assignment.assignmentEpoch(),
+ assignment.materializationId(), report.installationId(), policy.endpoint(server));
+ }
+
+ /// Only bounded identity DTOs are encoded while holding apply. Config and archive payloads never enter this image;
+ /// reject before retaining each additional element and let transport separately check the final protobuf size.
+ private static final class RoutingProjection {
+ private final int _maxRecords;
+ private final int _maxBytes;
+ private int _records = 2; // Own participant and committed control.
+ private long _bytes;
+
+ private RoutingProjection(RoutingSnapshot header, int maxRecords, int maxBytes) throws IOException {
+ _maxRecords = maxRecords;
+ _maxBytes = maxBytes;
+ _bytes = MetadataProtocol.encodeValue(header).length;
+ checkBounds();
+ }
+
+ private void add(Object value, int records) throws IOException {
+ _records += records;
+ // Empty arrays/nulls in the skeleton remain charged, conservatively covering replacement and separators.
+ _bytes += MetadataProtocol.encodeValue(value).length + 1;
+ checkBounds();
+ }
+
+ private void checkBounds() {
+ require(_records <= _maxRecords && _bytes <= _maxBytes, Status.RESPONSE_TOO_LARGE,
+ "Routing snapshot exceeds its record or byte bound");
+ }
+ }
+
/// Captures one complete own-session projection after the caller's consensus read barrier. Readiness is not
/// required for bootstrap. The revision describes this whole projection, not arbitrary operation lookup progress.
/// Fixed bounds deliberately reject a large image; pagination must not be simulated by dropping records here.
diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/NativeBrokerRoutingPolicy.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/NativeBrokerRoutingPolicy.java
new file mode 100644
index 00000000000..6a567020d96
--- /dev/null
+++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/NativeBrokerRoutingPolicy.java
@@ -0,0 +1,138 @@
+/**
+ * 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.coordination.ratis.metadata;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.charset.StandardCharsets;
+import java.util.HashSet;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import javax.annotation.Nullable;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Capability;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ParticipantState;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Status;
+import org.apache.pinot.spi.config.table.TableConfig;
+
+
+/// Administrator-owned broker tenant access and query endpoint approval. Participant advertisements cannot create
+/// these grants; both policy and advertisements must agree in the same committed routing observation.
+public record NativeBrokerRoutingPolicy(Map<String, Set<String>> brokers, Map<String, String> servers) {
+ public static final String CONFIG_KEY = "pinot.coordination.ratis.broker.routing";
+ private static final int MAX_IDENTITIES = 64;
+ private static final int MAX_TENANTS = 64;
+
+ public NativeBrokerRoutingPolicy {
+ Map<String, Set<String>> copy = new TreeMap<>();
+ brokers.forEach((id, tenants) -> copy.put(id, Set.copyOf(tenants)));
+ brokers = Map.copyOf(copy);
+ servers = Map.copyOf(servers);
+ }
+
+ public static NativeBrokerRoutingPolicy parse(String json) {
+ require(json != null && !json.isBlank(), "Native broker routing policy is not configured");
+ try {
+ JsonNode root = MetadataProtocol.decodeValue(json.getBytes(StandardCharsets.UTF_8), JsonNode.class);
+ require(root.isObject() && root.size() == 2 && root.path("brokers").isObject()
+ && root.path("servers").isObject(), "Routing policy requires only brokers and servers objects");
+ require(root.path("brokers").size() <= MAX_IDENTITIES && root.path("servers").size() <= MAX_IDENTITIES,
+ "Routing policy exceeds its identity bound");
+ Map<String, Set<String>> brokers = new TreeMap<>();
+ for (var entry : root.path("brokers").properties()) {
+ identifier(entry.getKey());
+ JsonNode values = entry.getValue();
+ require(values.isArray() && values.size() > 0 && values.size() <= MAX_TENANTS,
+ "Broker policy requires a bounded nonempty tenant array");
+ Set<String> tenants = new HashSet<>();
+ for (JsonNode value : values) {
+ require(value.isTextual(), "Broker tenants must be strings");
+ identifier(value.textValue());
+ require(tenants.add(value.textValue()), "Duplicate broker tenant");
+ }
+ brokers.put(entry.getKey(), tenants);
+ }
+ Map<String, String> servers = new TreeMap<>();
+ Set<String> endpoints = new HashSet<>();
+ for (var entry : root.path("servers").properties()) {
+ identifier(entry.getKey());
+ require(entry.getValue().isTextual(), "Server endpoint must be a string");
+ String endpoint = entry.getValue().textValue();
+ validateEndpoint(endpoint);
+ require(endpoints.add(endpoint), "Query endpoint cannot identify multiple servers");
+ servers.put(entry.getKey(), endpoint);
+ }
+ return new NativeBrokerRoutingPolicy(brokers, servers);
+ } catch (IOException e) {
+ throw new CoordinationRejection(Status.INVALID_REQUEST, "Invalid native broker routing policy JSON");
+ }
+ }
+
+ public boolean permits(String brokerId, TableConfig table) {
+ return brokers.getOrDefault(brokerId, Set.of()).contains(brokerTenant(table));
+ }
+
+ public static String brokerTenant(TableConfig table) {
+ require(table != null && table.getTenantConfig() != null, "Routing requires a broker tenant");
+ String tenant = table.getTenantConfig().getBroker();
+ // Match Pinot's existing omitted-tenant convention, without granting any default broker policy entry.
+ tenant = tenant == null ? "DefaultTenant" : tenant;
+ identifier(tenant);
+ return tenant;
+ }
+
+ @Nullable
+ public String endpoint(ParticipantState server) {
+ String approved = servers.get(server.participant().participantId());
+ return approved != null && server.capabilities().contains(Capability.SERVER_QUERY)
+ && approved.equals(server.endpoints().get("queryGrpc"))
+ && "2".equals(server.endpoints().get("nativeQueryVersion")) ? approved : null;
+ }
+
+ private static void validateEndpoint(String value) {
+ require(value != null && value.length() <= MembershipProtocol.MAX_ENDPOINT_VALUE_LENGTH,
+ "Query endpoint exceeds its bound");
+ try {
+ URI uri = new URI(value);
+ require("grpcs".equals(uri.getScheme()) && uri.getHost() != null
+ && uri.getHost().equals(uri.getHost().toLowerCase(Locale.ROOT))
+ && uri.getPort() > 0 && uri.getPort() <= 65535 && uri.getRawUserInfo() == null
+ && uri.getRawQuery() == null && uri.getRawFragment() == null
+ && (uri.getRawPath() == null || uri.getRawPath().isEmpty())
+ && value.equals(new URI("grpcs", null, uri.getHost(), uri.getPort(), null, null, null).toASCIIString()),
+ "Query endpoint requires a canonical grpcs host and port without path or credentials");
+ } catch (URISyntaxException e) {
+ throw new CoordinationRejection(Status.INVALID_REQUEST, "Invalid native query endpoint");
+ }
+ }
+
+ private static void identifier(String value) {
+ OfflineSegmentCatalog.identifier(value, 1024);
+ require(!value.contains("*"), "Routing policy does not accept wildcards");
+ }
+
+ private static void require(boolean condition, String message) {
+ if (!condition) {
+ throw new CoordinationRejection(Status.INVALID_REQUEST, message);
+ }
+ }
+}
diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/OfflineRoutingProtocol.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/OfflineRoutingProtocol.java
new file mode 100644
index 00000000000..69fe8bb2657
--- /dev/null
+++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/OfflineRoutingProtocol.java
@@ -0,0 +1,69 @@
+/**
+ * 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.coordination.ratis.metadata;
+
+import java.util.List;
+import java.util.UUID;
+import javax.annotation.Nullable;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ControlState;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ParticipantState;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ParticipantToken;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.ObjectVersion;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Revision;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Status;
+
+
+/// Complete bounded OFFLINE routing observations, never serving grants. Every current segment is represented,
+/// including segments without an installed replica. A null endpoint means installed evidence is not selectable.
+/// Config payloads, archive locations and other participants' private metadata are deliberately not exposed.
+public final class OfflineRoutingProtocol {
+ public static final int MAX_SNAPSHOT_RECORDS = 4096;
+ public static final int MIN_SNAPSHOT_BYTES = 1024;
+ public static final int MAX_SNAPSHOT_BYTES = MetadataProtocol.MAX_CHANGE_BYTES;
+
+ private OfflineRoutingProtocol() {
+ }
+
+ public record ReplicaRouting(ParticipantToken server, ObjectVersion assignment, long assignmentEpoch,
+ UUID materializationId, UUID installationId, @Nullable String endpoint) {
+ }
+
+ public record SegmentRouting(String name, ObjectVersion content, String contentSha256,
+ List<ReplicaRouting> replicas) {
+ public SegmentRouting {
+ replicas = List.copyOf(replicas);
+ }
+ }
+
+ public record TableRouting(String tableName, ObjectVersion table, String schemaName, ObjectVersion schema,
+ List<SegmentRouting> segments) {
+ public TableRouting {
+ segments = List.copyOf(segments);
+ }
+ }
+
+ /// Empty requested table name discovers authorized tables. A table image is complete at this revision; consumers
+ /// must atomically replace it, reject incomplete coverage and separately enforce query lifetime and serving fences.
+ public record RoutingSnapshot(Status status, Revision revision, String detail, @Nullable ParticipantState participant,
+ @Nullable ControlState control, List<String> tables, @Nullable TableRouting table) {
+ public RoutingSnapshot {
+ tables = List.copyOf(tables);
+ }
+ }
+}
diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/MetadataGrpcServer.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/MetadataGrpcServer.java
index 8acd3ba10d0..045d9166c97 100644
--- a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/MetadataGrpcServer.java
+++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/MetadataGrpcServer.java
@@ -47,6 +47,7 @@ import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;
import javax.net.ssl.SSLSession;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol;
+import org.apache.pinot.coordination.ratis.rpc.BrokerRoutingServiceGrpc;
import org.apache.pinot.coordination.ratis.rpc.MinionTaskServiceGrpc;
import org.apache.pinot.coordination.ratis.rpc.ParticipantMembershipServiceGrpc;
import org.apache.pinot.coordination.ratis.rpc.ServerInstallationServiceGrpc;
@@ -63,6 +64,7 @@ public final class MetadataGrpcServer implements AutoCloseable {
private final ThreadPoolExecutor _membershipExecutor;
private final ThreadPoolExecutor _installationExecutor;
private final ThreadPoolExecutor _taskExecutor;
+ private final ThreadPoolExecutor _routingExecutor;
private final ThreadPoolExecutor _dispatchExecutor;
private final ScheduledThreadPoolExecutor _deadlines;
@@ -87,6 +89,8 @@ public final class MetadataGrpcServer implements AutoCloseable {
Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());
_taskExecutor = new ThreadPoolExecutor(2, 2, 0, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(16),
Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());
+ _routingExecutor = new ThreadPoolExecutor(2, 2, 0, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(16),
+ Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());
// gRPC performs method lookup before callExecutor selection. Keep that short dispatch off both handler pools;
// its SerializingExecutor switches to the selected pool before starting the application call.
_dispatchExecutor = new ThreadPoolExecutor(2, 2, 0, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(64),
@@ -109,6 +113,9 @@ public final class MetadataGrpcServer implements AutoCloseable {
if (MinionTaskServiceGrpc.SERVICE_NAME.equals(service)) {
return _taskExecutor;
}
+ if (BrokerRoutingServiceGrpc.SERVICE_NAME.equals(service)) {
+ return _routingExecutor;
+ }
return ServerInstallationServiceGrpc.SERVICE_NAME.equals(service) ? _installationExecutor : _executor;
}
}).maxInboundMessageSize(MetadataProtocol.MAX_COMMAND_BYTES).maxInboundMetadataSize(8192)
@@ -128,6 +135,11 @@ public final class MetadataGrpcServer implements AutoCloseable {
new TaskGrpcService(access, taskAccess, participantPolicy, audit, limits.maxRpcMillis()),
new PeerInterceptor(limits.maxRpcMillis(), audit)));
}
+ if (access instanceof RoutingAccess routingAccess) {
+ builder.addService(ServerInterceptors.intercept(
+ new RoutingGrpcService(routingAccess, participantPolicy, audit, limits.maxRpcMillis()),
+ new PeerInterceptor(limits.maxRpcMillis(), audit)));
+ }
}
server = builder.build().start();
_server = server;
@@ -139,6 +151,7 @@ public final class MetadataGrpcServer implements AutoCloseable {
_membershipExecutor.shutdownNow();
_installationExecutor.shutdownNow();
_taskExecutor.shutdownNow();
+ _routingExecutor.shutdownNow();
_dispatchExecutor.shutdownNow();
_deadlines.shutdownNow();
throw e;
@@ -157,6 +170,7 @@ public final class MetadataGrpcServer implements AutoCloseable {
_membershipExecutor.shutdownNow();
_installationExecutor.shutdownNow();
_taskExecutor.shutdownNow();
+ _routingExecutor.shutdownNow();
_dispatchExecutor.shutdownNow();
_deadlines.shutdownNow();
try {
diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/RoutingAccess.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/RoutingAccess.java
new file mode 100644
index 00000000000..42d9a8323d2
--- /dev/null
+++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/RoutingAccess.java
@@ -0,0 +1,31 @@
+/**
+ * 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.coordination.ratis.transport;
+
+import java.io.IOException;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ParticipantToken;
+import org.apache.pinot.coordination.ratis.metadata.OfflineRoutingProtocol.RoutingSnapshot;
+
+
+/// Authenticated BROKER discovery and complete per-table routing observations after a consensus read barrier.
+/// Implementations return no partial image when a requested bound is exceeded.
+public interface RoutingAccess {
+ RoutingSnapshot routingSnapshot(ParticipantToken broker, String tableName, int maxRecords, int maxBytes,
+ long timeoutMillis) throws IOException;
+}
diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/RoutingGrpcService.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/RoutingGrpcService.java
new file mode 100644
index 00000000000..b61cb979a17
--- /dev/null
+++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/RoutingGrpcService.java
@@ -0,0 +1,134 @@
+/**
+ * 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.coordination.ratis.transport;
+
+import io.grpc.Context;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import io.grpc.stub.StreamObserver;
+import java.io.IOException;
+import java.util.Objects;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Role;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol;
+import org.apache.pinot.coordination.ratis.metadata.OfflineRoutingProtocol;
+import org.apache.pinot.coordination.ratis.rpc.BrokerRoutingServiceGrpc;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc;
+import org.apache.pinot.coordination.ratis.transport.ParticipantAccessPolicy.ParticipantIdentity;
+
+
+/// BROKER-only routing reads on an independently bounded lane. Certificate authorization is rechecked at delivery;
+/// committed tenant policy and exact participant sessions are checked in the coherent catalog projection.
+final class RoutingGrpcService extends BrokerRoutingServiceGrpc.BrokerRoutingServiceImplBase {
+ static final int MAX_REQUEST_BYTES = 64 * 1024;
+ private final RoutingAccess _routing;
+ private final ParticipantAccessPolicy _policy;
+ private final MetadataAudit _audit;
+ private final long _maxRpcMillis;
+ private final Semaphore _calls = new Semaphore(2);
+
+ RoutingGrpcService(RoutingAccess routing, ParticipantAccessPolicy policy, MetadataAudit audit, long maxRpcMillis) {
+ _routing = Objects.requireNonNull(routing);
+ _policy = Objects.requireNonNull(policy);
+ _audit = Objects.requireNonNull(audit);
+ _maxRpcMillis = maxRpcMillis;
+ }
+
+ @Override
+ public void snapshot(MetadataRpc.RoutingSnapshotRequest request,
+ StreamObserver<MetadataRpc.RoutingSnapshot> observer) {
+ ParticipantIdentity identity = null;
+ String outcome = "UNAVAILABLE";
+ boolean acquired = false;
+ try {
+ identity = authorize();
+ remainingMillis();
+ if (request.getSerializedSize() > MAX_REQUEST_BYTES || !_calls.tryAcquire()) {
+ throw Status.RESOURCE_EXHAUSTED.withDescription("Routing request exceeds admission bounds")
+ .asRuntimeException();
+ }
+ acquired = true;
+ var broker = RoutingRpcCodec.broker(request, identity);
+ var snapshot = _routing.routingSnapshot(broker, request.getTableName(), request.getMaxRecords(),
+ request.getMaxBytes(), remainingMillis());
+ if (snapshot.status() == MetadataProtocol.Status.OK) {
+ if (snapshot.participant() == null || !broker.equals(snapshot.participant().participant())
+ || snapshot.control() == null || !broker.historyId().equals(snapshot.revision().historyId())
+ || request.getTableName().isEmpty() != (snapshot.table() == null)
+ || snapshot.table() != null && !request.getTableName().equals(snapshot.table().tableName())) {
+ throw Status.INTERNAL.withDescription("Routing projection identity differs").asRuntimeException();
+ }
+ } else if (snapshot.participant() != null || snapshot.control() != null || snapshot.table() != null
+ || !snapshot.tables().isEmpty()) {
+ throw Status.INTERNAL.withDescription("Failed routing projection contains data").asRuntimeException();
+ }
+ MetadataRpc.RoutingSnapshot response = RoutingRpcCodec.snapshot(snapshot);
+ if (response.getSerializedSize() > request.getMaxBytes()
+ || response.getSerializedSize() > OfflineRoutingProtocol.MAX_SNAPSHOT_BYTES) {
+ response = MetadataRpc.RoutingSnapshot.newBuilder().setStatus(MetadataRpc.ResultStatus.RESPONSE_TOO_LARGE)
+ .setRevision(response.getRevision()).setDetail("Routing projection exceeds its wire byte bound").build();
+ }
+ remainingMillis();
+ if (!identity.equals(authorize())) {
+ throw Status.PERMISSION_DENIED.withDescription("Broker authorization changed during routing request")
+ .asRuntimeException();
+ }
+ outcome = response.getStatus().name();
+ observer.onNext(response);
+ observer.onCompleted();
+ } catch (StatusRuntimeException e) {
+ outcome = e.getStatus().getCode().name();
+ observer.onError(e);
+ } catch (IllegalArgumentException e) {
+ outcome = "INVALID_ARGUMENT";
+ observer.onError(Status.INVALID_ARGUMENT.withDescription("Invalid routing request").asRuntimeException());
+ } catch (IOException e) {
+ outcome = "UNAVAILABLE";
+ observer.onError(Status.UNAVAILABLE.withDescription("Routing projection is unavailable").asRuntimeException());
+ } catch (RuntimeException e) {
+ outcome = "INTERNAL";
+ observer.onError(Status.INTERNAL.withDescription("Routing projection failed").asRuntimeException());
+ } finally {
+ if (acquired) {
+ _calls.release();
+ }
+ MetadataGrpcService.audit(_audit, identity == null ? "" : identity.participantId(), "RoutingSnapshot", outcome);
+ }
+ }
+
+ private ParticipantIdentity authorize() {
+ String fingerprint = MetadataGrpcServer.PEER_FINGERPRINT.get();
+ ParticipantIdentity identity = fingerprint == null ? null : _policy.authorize(fingerprint);
+ if (identity == null || identity.role() != Role.BROKER) {
+ throw Status.PERMISSION_DENIED.withDescription("Broker routing access is not authorized").asRuntimeException();
+ }
+ return identity;
+ }
+
+ private long remainingMillis() {
+ Context context = Context.current();
+ long remaining = context.getDeadline() == null ? _maxRpcMillis
+ : context.getDeadline().timeRemaining(TimeUnit.MILLISECONDS);
+ if (context.isCancelled() || remaining <= 0) {
+ throw Status.DEADLINE_EXCEEDED.withDescription("Routing request deadline expired").asRuntimeException();
+ }
+ return Math.min(remaining, _maxRpcMillis);
+ }
+}
diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/RoutingRpcCodec.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/RoutingRpcCodec.java
new file mode 100644
index 00000000000..eb5d4169d7d
--- /dev/null
+++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/RoutingRpcCodec.java
@@ -0,0 +1,88 @@
+/**
+ * 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.coordination.ratis.transport;
+
+import java.util.UUID;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ParticipantToken;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Role;
+import org.apache.pinot.coordination.ratis.metadata.OfflineRoutingProtocol;
+import org.apache.pinot.coordination.ratis.metadata.OfflineRoutingProtocol.RoutingSnapshot;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc;
+import org.apache.pinot.coordination.ratis.transport.ParticipantAccessPolicy.ParticipantIdentity;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+
+/// Closed, identity-only broker routing vocabulary. The authenticated certificate supplies the stable broker ID;
+/// the request can select only its session and one table or authorized discovery.
+final class RoutingRpcCodec {
+ private RoutingRpcCodec() {
+ }
+
+ static ParticipantToken broker(MetadataRpc.RoutingSnapshotRequest request, ParticipantIdentity identity) {
+ MetadataRpcCodec.rejectUnknown(request);
+ checkArgument(identity.role() == Role.BROKER && request.hasSession()
+ && request.getSession().getRegistrationEpoch() > 0, "Current BROKER session required");
+ checkArgument(request.getMaxRecords() > 0
+ && request.getMaxRecords() <= OfflineRoutingProtocol.MAX_SNAPSHOT_RECORDS
+ && request.getMaxBytes() >= OfflineRoutingProtocol.MIN_SNAPSHOT_BYTES
+ && request.getMaxBytes() <= OfflineRoutingProtocol.MAX_SNAPSHOT_BYTES
+ && request.getTableName().length() <= 1024, "Invalid routing projection bounds");
+ return new ParticipantToken(UUID.fromString(request.getHistoryId()), identity.participantId(), Role.BROKER,
+ UUID.fromString(request.getSession().getIncarnation()), request.getSession().getRegistrationEpoch());
+ }
+
+ static MetadataRpc.RoutingSnapshot snapshot(RoutingSnapshot snapshot) {
+ var response = MetadataRpc.RoutingSnapshot.newBuilder()
+ .setStatus(MetadataRpc.ResultStatus.valueOf(snapshot.status().name())).setDetail(snapshot.detail())
+ .setRevision(MetadataRpc.Revision.newBuilder().setHistoryId(snapshot.revision().historyId().toString())
+ .setIndex(snapshot.revision().index())).addAllTables(snapshot.tables());
+ if (snapshot.participant() != null) {
+ response.setParticipant(MembershipRpcCodec.participant(snapshot.participant()));
+ }
+ if (snapshot.control() != null) {
+ response.setControl(MembershipRpcCodec.control(snapshot.control()));
+ }
+ if (snapshot.table() != null) {
+ var table = snapshot.table();
+ var encodedTable = MetadataRpc.TableRouting.newBuilder().setTableName(table.tableName())
+ .setTable(OfflineSegmentRpcCodec.version(table.table())).setSchemaName(table.schemaName())
+ .setSchema(OfflineSegmentRpcCodec.version(table.schema()));
+ for (var segment : table.segments()) {
+ var encodedSegment = MetadataRpc.SegmentRouting.newBuilder().setName(segment.name())
+ .setContent(OfflineSegmentRpcCodec.version(segment.content())).setContentSha256(segment.contentSha256());
+ for (var replica : segment.replicas()) {
+ var encodedReplica = MetadataRpc.ReplicaRouting.newBuilder()
+ .setServer(MembershipRpcCodec.participant(replica.server()))
+ .setAssignment(OfflineSegmentRpcCodec.version(replica.assignment()))
+ .setAssignmentEpoch(replica.assignmentEpoch())
+ .setMaterializationId(replica.materializationId().toString())
+ .setInstallationId(replica.installationId().toString());
+ if (replica.endpoint() != null) {
+ encodedReplica.setEndpoint(replica.endpoint());
+ }
+ encodedSegment.addReplicas(encodedReplica);
+ }
+ encodedTable.addSegments(encodedSegment);
+ }
+ response.setTable(encodedTable);
+ }
+ return response.build();
+ }
+}
diff --git a/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/TaskCatalogTest.java b/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/TaskCatalogTest.java
index ccbec5da657..189d089c379 100644
--- a/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/TaskCatalogTest.java
+++ b/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/TaskCatalogTest.java
@@ -64,6 +64,7 @@ import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Status;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.TaskCommand;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.TaskSessionCommand;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Transaction;
+import org.apache.pinot.coordination.ratis.metadata.OfflineRoutingProtocol.RoutingSnapshot;
import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.Assignment;
import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.Content;
import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.LoaderDescriptor;
@@ -721,6 +722,136 @@ public class TaskCatalogTest {
f.restore();
}
+ @Test
+ public void testRoutingDiscoveryCoverageBoundsAndNoPrivatePayloads() throws Exception {
+ Fixture f = new Fixture();
+ ParticipantToken broker = f.routingBroker(0);
+ f.routingPolicies("DefaultTenant", "grpcs://localhost:8090", true);
+ UUID hiddenTable = UUID.randomUUID();
+ assertEquals(f.mutate(Mutation.createTableConfig(f._history, hiddenTable,
+ new TableConfigBuilder(TableType.OFFLINE).setTableName("private").setBrokerTenant("PrivateTenant").build()))
+ .status(), Status.OK);
+ Content installed = f.publishRoutingContent("installed");
+ f.publishRoutingContent("pending");
+ ParticipantToken server = f.routingServer(Set.of());
+ ReconcileContent reconcile = new ReconcileContent(f._owner, installed.segmentObjectId());
+ assertEquals(f.segment(reconcile).status(), Status.OK);
+ assertEquals(f.segment(reconcile).status(), Status.OK);
+ assertEquals(f.routing(broker, TABLE).table().segments().getFirst().replicas().size(), 0,
+ "Desired assignment is not installation evidence");
+ f.install(f.assignments(installed).getFirst());
+ RoutingSnapshot discovery = f.routing(broker, "");
+ assertEquals(discovery.status(), Status.OK);
+ assertEquals(discovery.tables(), List.of(TABLE));
+ assertFalse(discovery.participant().ready(), "Routing bootstrap must not manufacture broker readiness");
+ assertTrue(discovery.participant().capabilities().isEmpty());
+ RoutingSnapshot image = f.routing(broker, TABLE);
+ assertEquals(image.status(), Status.OK);
+ assertEquals(image.table().segments().size(), 2, "Uninstalled published content must remain explicit");
+ assertEquals(image.table().segments().get(1).name(), "pending");
+ assertTrue(image.table().segments().get(1).replicas().isEmpty());
+ var replica = image.table().segments().getFirst().replicas().getFirst();
+ assertEquals(replica.server(), server);
+ assertEquals(replica.endpoint(), null, "Installed SERVER without SERVER_QUERY is not selectable");
+ String json = new String(MetadataProtocol.encodeValue(image), StandardCharsets.UTF_8);
+ assertFalse(json.contains("private-archive"));
+ assertFalse(json.contains("never-expose"));
+ assertFalse(json.contains("PrivateTenant"));
+ assertFalse(json.contains("loader"));
+ assertEquals(f.routing(broker, "private_OFFLINE").status(), Status.FENCED);
+ assertEquals(f.routing(broker, "orders").status(), Status.INVALID_REQUEST);
+ assertEquals(f.routing(broker, "orders_REALTIME").status(), Status.INVALID_REQUEST);
+ assertEquals(f.routing(broker, "orders_OFFLINE_OFFLINE").status(), Status.INVALID_REQUEST);
+ assertEquals(f.routing(broker, ".orders_OFFLINE").status(), Status.INVALID_REQUEST);
+ for (RoutingSnapshot rejected : List.of(
+ f._catalog.routingSnapshot(broker, TABLE, 6, OfflineRoutingProtocol.MAX_SNAPSHOT_BYTES),
+ f._catalog.routingSnapshot(broker, TABLE, 4096, 1024),
+ f._catalog.routingSnapshot(broker, "", 2, OfflineRoutingProtocol.MAX_SNAPSHOT_BYTES))) {
+ assertEquals(rejected.status(), Status.RESPONSE_TOO_LARGE);
+ assertEquals(rejected.table(), null);
+ assertEquals(rejected.participant(), null);
+ assertTrue(rejected.tables().isEmpty());
+ }
+ f.restore();
+ assertEquals(f.routing(broker, TABLE), image);
+ }
+
+ @Test
+ public void testRoutingExactProofEndpointAndCurrentSessionFences() throws Exception {
+ Fixture f = new Fixture();
+ ParticipantToken broker = f.routingBroker(0);
+ f.routingPolicies("DefaultTenant", "grpcs://localhost:8090", true);
+ Content content = f.publishRoutingContent("installed");
+ ParticipantToken server = f.routingServer(Set.of(Capability.SERVER_QUERY));
+ ReconcileContent reconcile = new ReconcileContent(f._owner, content.segmentObjectId());
+ assertEquals(f.segment(reconcile).status(), Status.OK);
+ assertEquals(f.segment(reconcile).status(), Status.OK);
+ f.install(f.assignments(content).getFirst());
+ assertEquals(f.segment(reconcile).status(), Status.OK);
+ var replica = f.routing(broker, TABLE).table().segments().getFirst().replicas().getFirst();
+ assertEquals(replica.endpoint(), "grpcs://localhost:8090");
+ assertEquals(f.member(new Unregister(f._owner.participant())).status(), Status.OK);
+ assertEquals(f.routing(broker, TABLE).table().segments().getFirst().replicas().getFirst(), replica,
+ "Completed proof must survive the former controller owner's retirement");
+ f.restore();
+ assertEquals(f.routing(broker, TABLE).table().segments().getFirst().replicas().getFirst(), replica);
+ f.routingPolicies("DefaultTenant", "grpcs://other:8090", true);
+ assertEquals(f.routing(broker, TABLE).table().segments().getFirst().replicas().getFirst().endpoint(), null,
+ "A self-advertised endpoint cannot bypass changed administrator approval");
+ f.routingPolicies("DefaultTenant", "grpcs://localhost:8090", false);
+ assertTrue(f.routing(broker, TABLE).table().segments().getFirst().replicas().isEmpty(),
+ "Revoking placement approval invalidates the candidate before reconciliation");
+ f.routingPolicies("DefaultTenant", "grpcs://localhost:8090", true);
+ assertEquals(f.member(new BeginDrain(server)).status(), Status.OK);
+ assertTrue(f.routing(broker, TABLE).table().segments().getFirst().replicas().isEmpty());
+ assertEquals(f.member(new Unregister(server)).status(), Status.OK);
+ var replacement = f.member(new Register(server.participantId(), Role.SERVER, UUID.randomUUID(), 1,
+ Map.of("queryGrpc", "grpcs://localhost:8090", "nativeQueryVersion", "2"), Set.of(Capability.SERVER_QUERY)));
+ assertEquals(replacement.status(), Status.OK);
+ assertEquals(f.member(new SetReady(replacement.participant().participant(), new Revision(f._history, f._index),
+ true)).status(), Status.OK);
+ assertTrue(f.routing(broker, TABLE).table().segments().getFirst().replicas().isEmpty(),
+ "Replacement server cannot inherit another incarnation's receipt");
+ assertEquals(f.member(new Unregister(broker)).status(), Status.OK);
+ ParticipantToken newBroker = f.routingBroker(1);
+ assertEquals(f.routing(broker, TABLE).status(), Status.FENCED);
+ assertEquals(f.routing(newBroker, TABLE).status(), Status.OK);
+ f.routingPolicies("OtherTenant", "grpcs://localhost:8090", true);
+ assertEquals(f.routing(newBroker, TABLE).status(), Status.FENCED);
+ assertTrue(f.routing(newBroker, "").tables().isEmpty());
+ assertEquals(f.member(new BeginDrain(newBroker)).status(), Status.OK);
+ assertEquals(f.routing(newBroker, "").status(), Status.FENCED);
+ }
+
+ @Test
+ public void testRoutingConfigurationChangesAndPolicyValidationFailClosed() throws Exception {
+ Fixture f = new Fixture();
+ ParticipantToken broker = f.routingBroker(0);
+ assertEquals(f.routing(broker, TABLE).status(), Status.NOT_READY);
+ f.routingPolicies("DefaultTenant", "grpcs://localhost:8090", true);
+ Content content = f.publishRoutingContent("installed");
+ f.routingServer(Set.of(Capability.SERVER_QUERY));
+ ReconcileContent reconcile = new ReconcileContent(f._owner, content.segmentObjectId());
+ assertEquals(f.segment(reconcile).status(), Status.OK);
+ assertEquals(f.segment(reconcile).status(), Status.OK);
+ f.install(f.assignments(content).getFirst());
+ CatalogRecord schema = f.record(RecordKind.SCHEMA, "orders");
+ assertEquals(f.mutate(Mutation.updateSchema(f.objectVersion(schema), new Schema.SchemaBuilder()
+ .setSchemaName("orders").addSingleValueDimension("id", DataType.INT)
+ .addSingleValueDimension("newField", DataType.STRING).build())).status(), Status.OK);
+ assertTrue(f.routing(broker, TABLE).table().segments().getFirst().replicas().isEmpty(),
+ "A proof against a previous exact schema version is not a current candidate");
+ for (String policy : List.of("{}", "{\"brokers\":{},\"servers\":{},\"extra\":{}}",
+ "{\"brokers\":{\"routing-broker\":[\"*\"]},\"servers\":{}}",
+ "{\"brokers\":{\"routing-broker\":[\"DefaultTenant\",\"DefaultTenant\"]},\"servers\":{}}",
+ "{\"brokers\":{},\"servers\":{\"server\":\"http://localhost:8090\"}}",
+ "{\"brokers\":{},\"servers\":{\"server\":\"grpcs://localhost:8090/path\"}}",
+ "{\"brokers\":{},\"servers\":{\"server\":\"grpcs://user@localhost:8090\"}}")) {
+ f.cluster(Map.of(NativeBrokerRoutingPolicy.CONFIG_KEY, policy));
+ assertEquals(f.routing(broker, TABLE).status(), Status.NOT_READY);
+ }
+ }
+
private static final class Fixture {
private final boolean _nativeTask;
private final UUID _cluster = UUID.randomUUID();
@@ -760,6 +891,45 @@ public class TaskCatalogTest {
.addSingleValueDimension("id", DataType.INT).build())).status(), Status.OK);
}
+ private ParticipantToken routingBroker(long epoch) throws IOException {
+ var registered = member(
+ new Register("routing-broker", Role.BROKER, UUID.randomUUID(), epoch, Map.of(), Set.of()));
+ assertEquals(registered.status(), Status.OK);
+ return registered.participant().participant();
+ }
+
+ private ParticipantToken routingServer(Set<Capability> capabilities) throws IOException {
+ var registered = member(new Register("routing-server", Role.SERVER, UUID.randomUUID(), 0,
+ Map.of("queryGrpc", "grpcs://localhost:8090", "nativeQueryVersion", "2"), capabilities));
+ assertEquals(registered.status(), Status.OK);
+ ParticipantToken server = registered.participant().participant();
+ assertEquals(member(new SetReady(server, new Revision(_history, _index), true)).status(), Status.OK);
+ return server;
+ }
+
+ private void routingPolicies(String tenant, String endpoint, boolean placed) throws IOException {
+ cluster(Map.of(NativeReplicaPlacementPolicy.CONFIG_KEY, JsonUtils.objectToString(
+ new NativeReplicaPlacementPolicy(placed
+ ? Map.of("routing-server", new NativeReplicaPlacementPolicy.Server("DefaultTenant", "zone-a"))
+ : Map.of())),
+ NativeBrokerRoutingPolicy.CONFIG_KEY, JsonUtils.objectToString(new NativeBrokerRoutingPolicy(
+ Map.of("routing-broker", Set.of(tenant)), Map.of("routing-server", endpoint))),
+ "controller.password", "never-expose"));
+ }
+
+ private Content publishRoutingContent(String name) throws IOException {
+ Content content = new Content(TABLE, _table, name, UUID.randomUUID(), "a".repeat(64),
+ "file:///private-archive/segment.tar.gz", new LoaderDescriptor(1, -1, false, null, null, 1, -1, Map.of()));
+ assertEquals(segment(new PublishValidatedContent(_owner, objectVersion(record(RecordKind.TABLE_CONFIG, TABLE)),
+ objectVersion(record(RecordKind.SCHEMA, "orders")), content)).status(), Status.OK);
+ return content;
+ }
+
+ private RoutingSnapshot routing(ParticipantToken broker, String table) throws IOException {
+ return _catalog.routingSnapshot(broker, table, OfflineRoutingProtocol.MAX_SNAPSHOT_RECORDS,
+ OfflineRoutingProtocol.MAX_SNAPSHOT_BYTES);
+ }
+
private TableConfig table() {
return new TableConfigBuilder(TableType.OFFLINE).setTableName("orders").build();
}
diff --git a/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/transport/MetadataGrpcTransportTest.java b/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/transport/MetadataGrpcTransportTest.java
index 95968775247..330913223af 100644
--- a/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/transport/MetadataGrpcTransportTest.java
+++ b/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/transport/MetadataGrpcTransportTest.java
@@ -77,9 +77,12 @@ import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.FeedCursor;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.FeedScope;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Query;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Result;
+import org.apache.pinot.coordination.ratis.metadata.NativeBrokerRoutingPolicy;
+import org.apache.pinot.coordination.ratis.metadata.NativeReplicaPlacementPolicy;
import org.apache.pinot.coordination.ratis.metadata.OfflineInstallationProtocol;
import org.apache.pinot.coordination.ratis.metadata.OfflineInstallationProtocol.ReportInstallation;
import org.apache.pinot.coordination.ratis.metadata.OfflineInstallationProtocol.Work;
+import org.apache.pinot.coordination.ratis.metadata.OfflineRoutingProtocol;
import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol;
import org.apache.pinot.coordination.ratis.metadata.OperationProtocol;
import org.apache.pinot.coordination.ratis.metadata.OwnershipProtocol;
@@ -1530,6 +1533,123 @@ public class MetadataGrpcTransportTest {
}
}
+ @Test(timeOut = 90_000)
+ public void testBrokerRoutingIdentityBoundsAndProtectedRenewal() throws Exception {
+ ParticipantIdentity broker = new ParticipantIdentity("Broker_worker", Role.BROKER, Set.of());
+ AtomicReference<ParticipantIdentity> binding = new AtomicReference<>(broker);
+ ParticipantAccessPolicy policy = fingerprint -> {
+ if (fingerprint.equals(_certificates.get("worker").fingerprint())) {
+ return binding.get();
+ }
+ if (fingerprint.equals(_certificates.get("controller-a").fingerprint())) {
+ return new ParticipantIdentity("Controller_routing", Role.CONTROLLER, Set.of());
+ }
+ return fingerprint.equals(_certificates.get("controller-b").fingerprint())
+ ? new ParticipantIdentity("Broker_other", Role.BROKER, Set.of()) : null;
+ };
+ try (Quorum quorum = new Quorum()) {
+ quorum.startMembership();
+ BlockingAccess access = new BlockingAccess(quorum._access);
+ try (Fixture fixture = new Fixture(MetadataGrpcService.Limits.defaults(), quorum._history, access,
+ quorum._access, policy);
+ MetadataGrpcClient controller = fixture.client("controller-a");
+ MetadataGrpcClient worker = fixture.client("worker");
+ MetadataGrpcClient other = fixture.client("controller-b")) {
+ UUID history = quorum._history;
+ MetadataRpc.ParticipantToken workerSession = registerReady(worker, history);
+ registerReady(other, history);
+ assertStatus(Status.Code.PERMISSION_DENIED, () -> worker.read(fixture.revision()));
+ assertStatus(Status.Code.PERMISSION_DENIED, () -> worker.openSnapshot(fixture.open()));
+ assertEquals(quorum._access.execute(new MetadataProtocol.RegisterClient(history, "native-test",
+ UUID.randomUUID(), 0), 5000).status(), MetadataProtocol.Status.OK);
+ commitNativeTest(quorum, new MetadataProtocol.Transaction(history, "native-test", 1, "config", 1,
+ List.of(MetadataProtocol.Mutation.createTableConfig(history, UUID.randomUUID(),
+ new TableConfigBuilder(TableType.OFFLINE).setTableName("routes").setNumReplicas(1).build()),
+ MetadataProtocol.Mutation.createSchema(history, UUID.randomUUID(),
+ new Schema.SchemaBuilder().setSchemaName("routes").addSingleValueDimension("id", DataType.INT)
+ .build()),
+ MetadataProtocol.Mutation.createClusterConfig(history, UUID.randomUUID(), Map.of(
+ NativeBrokerRoutingPolicy.CONFIG_KEY,
+ "{\"brokers\":{\"Broker_worker\":[\"DefaultTenant\"]},\"servers\":{}}",
+ NativeReplicaPlacementPolicy.CONFIG_KEY, "{\"servers\":{}}")))));
+ MetadataRpc.RoutingSnapshotRequest request = MetadataRpc.RoutingSnapshotRequest.newBuilder()
+ .setHistoryId(history.toString()).setSession(session(workerSession)).setMaxRecords(128)
+ .setMaxBytes(OfflineRoutingProtocol.MAX_SNAPSHOT_BYTES).build();
+ MetadataRpc.RoutingSnapshot discovery = worker.routingSnapshot(request);
+ assertEquals(discovery.getStatus(), MetadataRpc.ResultStatus.OK);
+ assertEquals(discovery.getParticipant().getParticipant(), workerSession);
+ assertEquals(discovery.getTablesList(), List.of("routes_OFFLINE"));
+ assertFalse(discovery.hasTable());
+ MetadataRpc.RoutingSnapshot table = worker.routingSnapshot(
+ request.toBuilder().setTableName("routes_OFFLINE").build(), 5000);
+ assertEquals(table.getStatus(), MetadataRpc.ResultStatus.OK);
+ assertTrue(table.hasTable());
+ assertTrue(table.getTable().hasTable());
+ assertTrue(table.getTable().hasSchema());
+ assertEquals(table.getTable().getTableName(), "routes_OFFLINE");
+ assertEquals(table.getTable().getSegmentsCount(), 0);
+ assertEquals(other.routingSnapshot(request).getStatus(), MetadataRpc.ResultStatus.FENCED);
+ assertStatus(Status.Code.PERMISSION_DENIED, () -> controller.routingSnapshot(request));
+ assertEquals(worker.routingSnapshot(request.toBuilder().setHistoryId(UUID.randomUUID().toString()).build())
+ .getStatus(), MetadataRpc.ResultStatus.HISTORY_MISMATCH);
+ assertStatus(Status.Code.INVALID_ARGUMENT, () -> worker.routingSnapshot(request.toBuilder()
+ .setSession(request.getSession().toBuilder().setRegistrationEpoch(0)).build()));
+ assertStatus(Status.Code.INVALID_ARGUMENT, () -> worker.routingSnapshot(request.toBuilder()
+ .setMaxBytes(OfflineRoutingProtocol.MIN_SNAPSHOT_BYTES - 1).build()));
+ assertStatus(Status.Code.INVALID_ARGUMENT, () -> worker.routingSnapshot(request.toBuilder()
+ .setSession(request.getSession().toBuilder().setUnknownFields(UnknownFieldSet.newBuilder()
+ .addField(100, UnknownFieldSet.Field.newBuilder().addVarint(1).build()).build())).build()));
+ MetadataRpc.RoutingSnapshot oversized = worker.routingSnapshot(request.toBuilder().setMaxRecords(1).build());
+ assertEquals(oversized.getStatus(), MetadataRpc.ResultStatus.RESPONSE_TOO_LARGE);
+ assertFalse(oversized.hasParticipant());
+ assertFalse(oversized.hasControl());
+ assertFalse(oversized.hasTable());
+ assertEquals(oversized.getTablesCount(), 0);
+
+ CountDownLatch entered = new CountDownLatch(2);
+ CountDownLatch release = new CountDownLatch(1);
+ access._beforeRouting.set(() -> {
+ entered.countDown();
+ try {
+ if (!release.await(10, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("Routing test gate was not released");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(e);
+ }
+ });
+ ExecutorService requests = Executors.newFixedThreadPool(2);
+ try {
+ Future<MetadataRpc.RoutingSnapshot> first = requests.submit(() -> worker.routingSnapshot(request));
+ Future<MetadataRpc.RoutingSnapshot> second = requests.submit(() -> worker.routingSnapshot(request));
+ assertTrue(entered.await(5, TimeUnit.SECONDS));
+ assertStatus(Status.Code.DEADLINE_EXCEEDED, () -> worker.routingSnapshot(request, 50));
+ MetadataRpc.MembershipResponse renewed = worker.membershipExecute(memberRenewal(history,
+ session(workerSession), 1, discovery.getRevision()));
+ assertEquals(renewed.getRenewals(0).getStatus(), MetadataRpc.ResultStatus.OK);
+ assertEquals(renewed.getRenewals(0).getPermit().getCapabilitiesCount(), 0);
+ assertEquals(release.getCount(), 1L, "Membership renewal must not wait for routing handlers");
+ binding.set(new ParticipantIdentity(broker.participantId(), Role.SERVER, Set.of()));
+ release.countDown();
+ for (Future<MetadataRpc.RoutingSnapshot> pending : List.of(first, second)) {
+ ExecutionException denied = expectThrows(ExecutionException.class, () -> pending.get(5, TimeUnit.SECONDS));
+ assertEquals(Status.fromThrowable(denied.getCause()).getCode(), Status.Code.PERMISSION_DENIED);
+ }
+ } finally {
+ release.countDown();
+ requests.shutdownNow();
+ access._beforeRouting.set(() -> { });
+ binding.set(broker);
+ }
+ assertEquals(worker.membershipExecute(MetadataRpc.MembershipCommandRequest.newBuilder()
+ .setHistoryId(history.toString()).setUnregister(session(workerSession)).build()).getStatus(),
+ MetadataRpc.ResultStatus.OK);
+ assertEquals(worker.routingSnapshot(request).getStatus(), MetadataRpc.ResultStatus.FENCED);
+ }
+ }
+ }
+
private static MetadataRpc.ParticipantToken registerReady(MetadataGrpcClient client, UUID history)
throws IOException {
return registerReady(client, history, 0);
@@ -1810,10 +1930,11 @@ public class MetadataGrpcTransportTest {
}
/// Blocks the four catalog handler threads without blocking membership access to the same real quorum.
- private static final class BlockingAccess implements MetadataAccess, InstallationAccess, TaskAccess {
+ private static final class BlockingAccess implements MetadataAccess, InstallationAccess, TaskAccess, RoutingAccess {
private final MetadataAccess _delegate;
private final AtomicReference<Runnable> _afterExecute = new AtomicReference<>(() -> { });
private final AtomicReference<Runnable> _beforeWork = new AtomicReference<>(() -> { });
+ private final AtomicReference<Runnable> _beforeRouting = new AtomicReference<>(() -> { });
private final CountDownLatch _entered = new CountDownLatch(4);
private final CountDownLatch _release = new CountDownLatch(1);
private volatile boolean _block;
@@ -1844,6 +1965,13 @@ public class MetadataGrpcTransportTest {
return ((TaskAccess) _delegate).taskSnapshot(minion, maxRecords, maxBytes, timeoutMillis);
}
+ @Override
+ public OfflineRoutingProtocol.RoutingSnapshot routingSnapshot(MembershipProtocol.ParticipantToken broker,
+ String tableName, int maxRecords, int maxBytes, long timeoutMillis) throws IOException {
+ _beforeRouting.get().run();
+ return ((RoutingAccess) _delegate).routingSnapshot(broker, tableName, maxRecords, maxBytes, timeoutMillis);
+ }
+
@Override
public OfflineInstallationProtocol.InstallationSnapshot installationSnapshot(
MembershipProtocol.ParticipantToken server, int maxRecords, int maxBytes, long timeoutMillis)
diff --git a/pinot-tools/src/test/java/org/apache/pinot/tools/service/RatisStartersTest.java b/pinot-tools/src/test/java/org/apache/pinot/tools/service/RatisStartersTest.java
index c5c28420c8e..3ee3b056e6a 100644
--- a/pinot-tools/src/test/java/org/apache/pinot/tools/service/RatisStartersTest.java
+++ b/pinot-tools/src/test/java/org/apache/pinot/tools/service/RatisStartersTest.java
@@ -64,6 +64,7 @@ import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Role;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.CatalogRecord;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.ObjectVersion;
+import org.apache.pinot.coordination.ratis.metadata.NativeBrokerRoutingPolicy;
import org.apache.pinot.coordination.ratis.metadata.NativeReplicaPlacementPolicy;
import org.apache.pinot.coordination.ratis.metadata.OfflineInstallationProtocol;
import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol;
@@ -395,14 +396,24 @@ public class RatisStartersTest {
assertEquals(updated.statusCode(), 200, updated.body());
String placementPolicy = JsonUtils.objectToString(Map.of("servers", Map.of(SERVER,
Map.of("tenant", "DefaultTenant", "failureDomain", "zone-a"))));
+ String routingPolicy = JsonUtils.objectToString(Map.of("brokers", Map.of(BROKER, List.of("DefaultTenant")),
+ "servers", Map.of(SERVER, "grpcs://127.0.0.1:" + ports[7])));
clusterTag = get(http, controllerUrl + "/cluster/configs", true).headers().firstValue("ETag").orElseThrow();
updated = write(http, controllerUrl + "/cluster/configs", "POST",
- JsonUtils.objectToString(Map.of(PREFIX + "offline.placement", placementPolicy)),
+ JsonUtils.objectToString(Map.of(PREFIX + "offline.placement", placementPolicy,
+ NativeBrokerRoutingPolicy.CONFIG_KEY, routingPolicy)),
Map.of("If-Match", clusterTag));
assertEquals(updated.statusCode(), 200, updated.body());
+ MetadataRpc.RoutingSnapshot discovered = brokerClient.routingSnapshot(routingRequest(
+ brokerState.getParticipant(), ""));
+ assertEquals(discovered.getStatus(), ResultStatus.OK, discovered.getDetail());
+ assertTrue(discovered.getTablesList().contains("orders_OFFLINE"));
+ assertFalse(discovered.hasTable(), "Discovery must not include table or replica payloads");
PinotConfiguration voterConfig = new PinotConfiguration(controllers.getFirst());
InstalledOutput uploaded = uploadSizeTestSegment(directory, voterConfig, http, controllerUrl,
secondControllerUrl, serverState.getParticipant(), table, schema);
+ assertRoutingInstallation(brokerClient, brokerState.getParticipant(), uploaded.content(),
+ serverState.getParticipant(), uploaded.installation());
long expectedBytes = uploaded.bytes();
awaitSize(http, serverAdmin + "/tables/orders_OFFLINE/size", "diskSizeInBytes", expectedBytes);
assertMissingSize(http, controllerUrl);
@@ -431,6 +442,8 @@ public class RatisStartersTest {
assertEquals(ServiceStatus.getServiceStatus(SERVER), ServiceStatus.Status.GOOD);
InstalledOutput generatedOutput = runNativeTask(manager, directory, history, endpoints, minion, controller,
http, controllerUrl, taskOutput, voterConfig, serverState.getParticipant(), serverAdmin);
+ assertRoutingInstallation(brokerClient, brokerState.getParticipant(), generatedOutput.content(),
+ serverState.getParticipant(), generatedOutput.installation());
for (int i = 6; i < 8; i++) {
try (ServerSocket unused = new ServerSocket(ports[i])) {
assertTrue(unused.isBound(), "Native query listener must remain closed");
@@ -439,6 +452,12 @@ public class RatisStartersTest {
assertTrue(manager.stopPinotInstanceById(SERVER));
assertTrue(serverClient.membershipRead(read).getParticipant().getRetired());
+ MetadataRpc.RoutingSnapshot unavailable = brokerClient.routingSnapshot(
+ routingRequest(brokerState.getParticipant(), "orders_OFFLINE"));
+ assertEquals(unavailable.getStatus(), ResultStatus.OK, unavailable.getDetail());
+ assertEquals(unavailable.getTable().getSegmentsCount(), 1,
+ "A published segment must remain visible when its only server retires");
+ assertEquals(unavailable.getTable().getSegments(0).getReplicasCount(), 0);
try (ServerSocket released = new ServerSocket(ports[8])) {
assertTrue(released.isBound(), "Server shutdown must release its administrative listener");
}
@@ -452,6 +471,8 @@ public class RatisStartersTest {
assertNotEquals(uploadedAgain.assignment().materializationId(),
uploaded.installation().assignment().materializationId());
assertNotEquals(uploadedAgain.installationId(), uploaded.installation().installationId());
+ MetadataRpc.TableRouting recoveredRouting = assertRoutingInstallation(brokerClient,
+ brokerState.getParticipant(), uploaded.content(), replacement, uploadedAgain);
awaitSize(http, serverAdmin + "/tables/orders_OFFLINE/size", "diskSizeInBytes", expectedBytes);
awaitSize(http, controllerUrl + "/tables/orders/size", "reportedSizeInBytes", expectedBytes);
ReplicaInstallation reinstalled = awaitInstallation(voterConfig, replacement, generatedOutput.content());
@@ -479,6 +500,8 @@ public class RatisStartersTest {
assertEquals(recovered, reinstalled, "Controller restart must retain the current installation and proof");
assertEquals(awaitInstallation(voterConfig, replacement, uploaded.content()), uploadedAgain,
"Controller restart must not create another materialization for an uploaded segment");
+ assertEquals(assertRoutingInstallation(brokerClient, brokerState.getParticipant(), uploaded.content(),
+ replacement, uploadedAgain), recoveredRouting, "Routing must retain the recovered installation proof");
var taskState = JsonUtils.stringToJsonNode(get(http, controllerUrl + "/tasks/native/nativeTasks", true).body());
assertEquals(taskState.path("tasks").size(), 1);
assertEquals(taskState.path("publications").size(), 1);
@@ -490,6 +513,15 @@ public class RatisStartersTest {
assertEquals(updated.statusCode(), 200, updated.body());
assertFalse(JsonUtils.stringToJsonNode(get(http, secondControllerUrl + "/cluster/configs", true).body())
.has(approvalKey), "Restart must recover the bounded HTTP writer stream");
+ clusterTag = get(http, controllerUrl + "/cluster/configs", true).headers().firstValue("ETag").orElseThrow();
+ updated = write(http, controllerUrl + "/cluster/configs/" + NativeBrokerRoutingPolicy.CONFIG_KEY, "DELETE", "",
+ Map.of("If-Match", clusterTag));
+ assertEquals(updated.statusCode(), 200, updated.body());
+ MetadataRpc.RoutingSnapshot revoked = brokerClient.routingSnapshot(
+ routingRequest(brokerState.getParticipant(), "orders_OFFLINE"));
+ assertNotEquals(revoked.getStatus(), ResultStatus.OK, "Revoking routing policy must deny the next snapshot");
+ assertFalse(revoked.hasTable());
+ assertEquals(revoked.getTablesCount(), 0);
// Stop participants before losing quorum so the test does not wait for failed retirements.
manager.stopPinotInstanceById(BROKER);
manager.stopPinotInstanceById(SERVER);
@@ -523,6 +555,36 @@ public class RatisStartersTest {
private record InstalledOutput(OfflineSegmentProtocol.Content content, long bytes, ReplicaInstallation installation) {
}
+ private static MetadataRpc.RoutingSnapshotRequest routingRequest(MetadataRpc.ParticipantToken broker,
+ String tableName) {
+ return MetadataRpc.RoutingSnapshotRequest.newBuilder().setHistoryId(broker.getHistoryId())
+ .setSession(MetadataRpc.ParticipantSession.newBuilder().setIncarnation(broker.getIncarnation())
+ .setRegistrationEpoch(broker.getRegistrationEpoch()))
+ .setTableName(tableName).setMaxRecords(4096).setMaxBytes(2 * 1024 * 1024).build();
+ }
+
+ private static MetadataRpc.TableRouting assertRoutingInstallation(MetadataRpcClient client,
+ MetadataRpc.ParticipantToken broker, OfflineSegmentProtocol.Content content,
+ MetadataRpc.ParticipantToken server, ReplicaInstallation installed) {
+ MetadataRpc.RoutingSnapshot snapshot = client.routingSnapshot(routingRequest(broker, content.tableName()));
+ assertEquals(snapshot.getStatus(), ResultStatus.OK, snapshot.getDetail());
+ assertEquals(snapshot.getParticipant().getParticipant(), broker);
+ assertEquals(snapshot.getTable().getTable().getObjectId(), content.tableObjectId().toString());
+ assertEquals(snapshot.getTable().getSegmentsCount(), 1);
+ MetadataRpc.SegmentRouting segment = snapshot.getTable().getSegments(0);
+ assertEquals(segment.getName(), content.segmentName());
+ assertEquals(segment.getContent().getObjectId(), content.segmentObjectId().toString());
+ assertEquals(segment.getContentSha256(), content.contentSha256());
+ assertEquals(segment.getReplicasCount(), 1);
+ MetadataRpc.ReplicaRouting replica = segment.getReplicas(0);
+ assertEquals(replica.getServer(), server);
+ assertEquals(replica.getAssignmentEpoch(), installed.assignment().assignmentEpoch());
+ assertEquals(replica.getMaterializationId(), installed.assignment().materializationId().toString());
+ assertEquals(replica.getInstallationId(), installed.installationId().toString());
+ assertFalse(replica.hasEndpoint(), "An approved address cannot grant a missing server query capability");
+ return snapshot.getTable();
+ }
+
private static InstalledOutput runNativeTask(PinotServiceManager manager, Path directory, String history,
String endpoints, CertificateFiles minion, CertificateFiles controller, HttpClient http, String controllerUrl,
Path output, PinotConfiguration voterConfig, MetadataRpc.ParticipantToken server, String serverAdmin)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org