This is an automated email from the ASF dual-hosted git repository. xiangfu0 pushed a commit to branch xiangfu0/codex2/ratis-broker-admission in repository https://gitbox.apache.org/repos/asf/pinot.git
commit d7da29e2c2b0a6f496d8d8439b2d52262a2d431d Author: Xiang Fu <[email protected]> AuthorDate: Tue Oct 20 03:55:10 2026 -0700 Fence native broker admissions and local delivery --- .../broker/helix/NativeBrokerParticipant.java | 508 +++++++++++++++++++-- .../broker/helix/NativeBrokerParticipantTest.java | 433 +++++++++++++++++- pinot-coordination-ratis/CONTROLLER-API-PARITY.md | 5 +- pinot-coordination-ratis/README.md | 31 +- 4 files changed, 922 insertions(+), 55 deletions(-) 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 3aa7bcda5bf..aff48e88640 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,17 +22,32 @@ import com.google.common.annotations.VisibleForTesting; import io.grpc.Status; import io.grpc.StatusRuntimeException; import java.io.IOException; +import java.util.ArrayList; import java.util.HashMap; +import java.util.List; import java.util.Map; +import java.util.Set; import java.util.UUID; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.ReentrantReadWriteLock; 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; +import org.apache.pinot.common.coordination.QueryDeadlineGuard; +import org.apache.pinot.common.coordination.ServingPermitGuard; +import org.apache.pinot.common.coordination.ServingPermitGuard.LeaseObservation; +import org.apache.pinot.common.coordination.ServingPermitGuard.MembershipView; +import org.apache.pinot.common.coordination.ServingPermitGuard.MonotonicClock; +import org.apache.pinot.common.coordination.ServingPermitGuard.PeerLeaseBound; +import org.apache.pinot.common.coordination.ServingPermitGuard.QualifiedLifetimePolicy; +import org.apache.pinot.common.coordination.ServingPermitGuard.QueryTicket; +import org.apache.pinot.common.coordination.ServingPermitGuard.ReadOnlyCapability; +import org.apache.pinot.common.coordination.ServingPermitGuard.ServingDeniedException; +import org.apache.pinot.common.request.context.NativeQueryDeadline; import org.apache.pinot.common.utils.ServiceStatus; import org.apache.pinot.coordination.ratis.rpc.MetadataRpc; import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.ResultStatus; @@ -41,9 +56,11 @@ import org.slf4j.LoggerFactory; import static com.google.common.base.Preconditions.checkArgument; import static com.google.common.base.Preconditions.checkState; +import static java.util.Objects.requireNonNull; -/// Owns a broker membership session without advertising query endpoints, readiness, or serving capabilities. +/// Owns a broker membership session without opening query listeners. The default is membership-only; explicitly +/// qualified serving settings enable local admission and committed readiness, never a network query resource. /// The applied revision covers only the committed own-participant/control projection, never table or routing state. /// Lifecycle operations and the single bounded heartbeat lane are serialized; status reads never perform RPCs. final class NativeBrokerParticipant implements AutoCloseable, ServiceStatus.ServiceStatusCallback { @@ -54,6 +71,15 @@ final class NativeBrokerParticipant implements AutoCloseable, ServiceStatus.Serv private final MetadataRpcConnection _connection; private final MetadataRpcConnection _routingConnection; private final Object _routingLock = new Object(); + private final ReentrantReadWriteLock _admission = new ReentrantReadWriteLock(); + @Nullable + private final ServingSettings _servingSettings; + @Nullable + private ServingPermitGuard _serving; + // Query paths use only this immutable projection and the admission lock, never the membership monitor. + private volatile ServingMembership _servingMembership = new ServingMembership(null, false, 0); + private Map<String, RoutingCapture> _queryRoutes = Map.of(); + private long _queryGeneration; private final UUID _incarnation = UUID.randomUUID(); private final ScheduledExecutorService _heartbeats = Executors.newSingleThreadScheduledExecutor( runnable -> new Thread(runnable, "native-broker-membership")); @@ -66,6 +92,7 @@ final class NativeBrokerParticipant implements AutoCloseable, ServiceStatus.Serv private long _nextRenewal = 1; private long _appliedIndex; private boolean _started; + private boolean _cleanedUp; private boolean _fenced; private boolean _membershipHealthy; private long _routingGeneration; @@ -77,10 +104,62 @@ final class NativeBrokerParticipant implements AutoCloseable, ServiceStatus.Serv NativeBrokerParticipant(NativeMetadataConfig config, MetadataRpcConnection connection, MetadataRpcConnection routingConnection) { + this(config, connection, routingConnection, null); + } + + NativeBrokerParticipant(NativeMetadataConfig config, MetadataRpcConnection connection, + MetadataRpcConnection routingConnection, @Nullable ServingSettings servingSettings) { checkArgument(connection != routingConnection, "Routing hydration requires a separate membership connection"); - _config = config; - _connection = connection; - _routingConnection = routingConnection; + _config = requireNonNull(config); + _connection = requireNonNull(connection); + _routingConnection = requireNonNull(routingConnection); + _servingSettings = servingSettings; + } + + /// Explicit qualified clock inputs, with no platform defaults and no listener activation. + record ServingSettings(MonotonicClock clock, QualifiedLifetimePolicy lifetimePolicy, + QueryDeadlineGuard queryDeadlines) { + ServingSettings { + requireNonNull(clock); + requireNonNull(lifetimePolicy); + requireNonNull(queryDeadlines); + } + } + + /// A request-owned immutable plan and fanout, bound to its original finite serving and selected-server leases. + static final class QueryAdmission { + private final NativeBrokerRoutingImage _image; + private final List<NativeBrokerRoutingImage.ServerRequest> _requests; + private final List<PeerLeaseBound> _serverLeases; + private final QueryTicket _ticket; + private final long _generation; + + private QueryAdmission(NativeBrokerRoutingImage image, List<NativeBrokerRoutingImage.ServerRequest> requests, + List<PeerLeaseBound> serverLeases, QueryTicket ticket, long generation) { + _image = image; + _requests = List.copyOf(requests); + _serverLeases = List.copyOf(serverLeases); + _ticket = ticket; + _generation = generation; + } + + NativeBrokerRoutingImage image() { + return _image; + } + + List<NativeBrokerRoutingImage.ServerRequest> requests() { + return _requests; + } + + NativeBrokerRoutingImage.PlanningMetadata planningMetadata(boolean ignoreCase) { + return _image.planningMetadata(ignoreCase); + } + } + + private record ServingMembership(@Nullable MembershipView view, boolean available, long minimumRevision) { + } + + private record RoutingCapture(NativeBrokerRoutingImage image, Map<String, PeerLeaseBound> serverLeases) { } synchronized void start() throws IOException { @@ -111,8 +190,35 @@ final class NativeBrokerParticipant implements AutoCloseable, ServiceStatus.Serv && token.getRegistrationEpoch() == previousEpoch + 1, "Broker registration response differs"); // Retain confirmed ownership before checking granted capabilities so failed startup can retire this session. _participant = token; - validateOwnProjection(registered); + MembershipView view = validateOwnProjection(registered); + _admission.writeLock().lock(); + try { + checkState(!_closed, "Native broker stopped during registration"); + if (_servingSettings != null) { + _serving = new ServingPermitGuard(token, Set.of(ReadOnlyCapability.BROKER_QUERY), + _servingSettings.clock(), _servingSettings.lifetimePolicy(), this::servingView); + } + publishMembership(view, false, null); + } finally { + _admission.writeLock().unlock(); + } + if (_servingSettings != null) { + var ready = command().setReady(MetadataRpc.ParticipantReadiness.newBuilder().setSession(session()) + .setAppliedRevision(registered.getRevision()).setReady(true)).build(); + var response = _connection.membershipExecute(ready, _config.rpcTimeoutMillis()); + validateResponse(response); + checkArgument(response.getStatus() == ResultStatus.OK && response.hasParticipant() + && response.getParticipant().getParticipant().equals(token) && response.getParticipant().getReady(), + "Native broker readiness was not committed"); + observeServingResponse(response); + // Read a complete own projection; never patch the bootstrap state with the readiness response. + var currentView = read(); + checkArgument(currentView.getStatus() == ResultStatus.OK, "Native broker readiness image is unavailable"); + view = validateOwnProjection(currentView); + checkArgument(view.ready() && !view.draining(), "Native broker readiness image is not ready"); + } _membershipHealthy = true; + publishMembership(view, true, null); _description = "Native broker membership active; routing unavailable and query listeners disabled"; _heartbeats.scheduleWithFixedDelay(this::heartbeat, _config.heartbeatMillis(), _config.heartbeatMillis(), TimeUnit.MILLISECONDS); @@ -133,16 +239,35 @@ final class NativeBrokerParticipant implements AutoCloseable, ServiceStatus.Serv if (_pendingRenewal == null) { MetadataRpc.MembershipResponse image = read(); checkArgument(image.getStatus() == ResultStatus.OK, "Broker membership is unavailable: %s", image.getStatus()); - validateOwnProjection(image); + publishMembership(validateOwnProjection(image), _membershipHealthy + && _servingMembership.available(), null); + if (_closed) { + return; + } _pendingRenewal = command().setRenew(MetadataRpc.ParticipantRenewal.newBuilder().setSession(session()) .setSequence(_nextRenewal).setAppliedRevision(image.getRevision())).build(); } + _admission.writeLock().lock(); + try { + if (_closed) { + return; + } + if (_serving != null) { + _serving.firstSend(_nextRenewal); + } + } finally { + _admission.writeLock().unlock(); + } // No read or rewrite may change a logical renewal after an unknown outcome, including its applied revision. MetadataRpc.MembershipResponse response = _connection.membershipExecute(_pendingRenewal, _config.rpcTimeoutMillis()); validateResponse(response); + observeServingResponse(response); + if (_closed) { + return; + } if (response.getStatus() == ResultStatus.NOT_READY) { - invalidateRouting(); + membershipUnavailable(); _membershipHealthy = false; _description = "Native broker control epoch unavailable; query listeners disabled"; return; @@ -153,13 +278,29 @@ final class NativeBrokerParticipant implements AutoCloseable, ServiceStatus.Serv checkArgument(renewal.getParticipant().equals(_participant) && renewal.getSequence() == _nextRenewal, "Broker renewal identity or sequence differs"); checkArgument((renewal.getStatus() == ResultStatus.OK || renewal.getStatus() == ResultStatus.NOT_READY) - && !renewal.hasPermit(), "Membership-only broker received unexpected authority or outcome"); + && (renewal.getStatus() == ResultStatus.OK || !renewal.hasPermit()) + && (_servingSettings != null || !renewal.hasPermit()), + "Broker received unexpected authority or outcome"); checkArgument(_nextRenewal < Long.MAX_VALUE, "Broker renewal sequences exhausted"); + // The durable outcome is consumed even if the subsequent coherent read fails. _nextRenewal++; _pendingRenewal = null; _membershipHealthy = renewal.getStatus() == ResultStatus.OK; if (!_membershipHealthy) { - invalidateRouting(); + membershipUnavailable(); + } + if (_servingSettings != null) { + var image = read(); + checkArgument(image.getStatus() == ResultStatus.OK, "Broker renewal image is unavailable: %s", + image.getStatus()); + MembershipView view = validateOwnProjection(image); + boolean draining = _membershipHealthy && !renewal.hasPermit() && view.draining(); + if (!renewal.hasPermit() && !draining) { + invalidateRouting(); + } + // The grant's minimum revision includes its own renewal command, so the pre-renewal read is insufficient. + publishMembership(view, _membershipHealthy && (renewal.hasPermit() || draining), + renewal.hasPermit() ? renewal.getPermit() : null); } _description = renewal.getStatus() == ResultStatus.OK ? routingDescription() @@ -170,11 +311,17 @@ final class NativeBrokerParticipant implements AutoCloseable, ServiceStatus.Serv || code == Status.Code.INVALID_ARGUMENT) { fence("Broker membership authorization or request rejected: " + code); } else { - invalidateRouting(); + membershipUnavailable(); _membershipHealthy = false; _description = "Native broker membership transport unavailable; query listeners disabled"; } - } catch (IOException | IllegalArgumentException e) { + } catch (ServingDeniedException e) { + fence("Native broker serving authority failed: " + e.getMessage()); + } catch (IOException e) { + membershipUnavailable(); + _membershipHealthy = false; + _description = "Native broker membership transport unavailable; query listeners disabled"; + } catch (IllegalArgumentException e) { fence("Invalid broker membership response: " + e.getMessage()); } catch (RuntimeException e) { fence("Broker membership failed: " + e.getClass().getSimpleName()); @@ -189,7 +336,8 @@ final class NativeBrokerParticipant implements AutoCloseable, ServiceStatus.Serv MetadataRpc.ParticipantToken participant; long generation; synchronized (this) { - if (_closed || _fenced || !_membershipHealthy) { + if (_closed || _fenced || !_membershipHealthy + || _servingMembership.view() != null && _servingMembership.view().draining()) { return; } participant = _participant; @@ -202,8 +350,10 @@ final class NativeBrokerParticipant implements AutoCloseable, ServiceStatus.Serv checkArgument(tables.size() <= MAX_ROUTING_TABLES, "Routing cache exceeds its table limit"); _routingIndex = discovery.getRevision().getIndex(); Map<String, NativeBrokerRoutingImage> images = new HashMap<>(); + Map<String, RoutingCapture> captures = new HashMap<>(); int bytes = discovery.getSerializedSize(); for (String table : tables) { + LeaseObservation observation = beginLeaseObservation(); var snapshot = routingSnapshot(participant, table, NativeBrokerRoutingImage.MAX_RECORDS, deadline); NativeBrokerRoutingImage image = new NativeBrokerRoutingImage(snapshot, participant, table); _routingIndex = snapshot.getRevision().getIndex(); @@ -211,16 +361,36 @@ final class NativeBrokerParticipant implements AutoCloseable, ServiceStatus.Serv "Routing cache exceeds its aggregate byte limit"); bytes += image.serializedBytes(); images.put(table, image); + captures.put(table, captureRouting(image, observation)); } synchronized (this) { if (!_closed && !_fenced && _membershipHealthy && generation == _routingGeneration && participant.equals(_participant)) { - _routingImages = Map.copyOf(images); + _admission.writeLock().lock(); + try { + if (_closed) { + return; + } + if (!sameRoutes(captures)) { + _queryGeneration = Math.incrementExact(_queryGeneration); + } + _queryRoutes = Map.copyOf(captures); + _routingImages = Map.copyOf(images); + } finally { + _admission.writeLock().unlock(); + } _description = routingDescription(); } } } catch (IOException | RuntimeException e) { synchronized (this) { + if (generation != _routingGeneration || _closed || _fenced) { + return; + } + if (e instanceof ServingDeniedException) { + fence("Native broker routing clock lost serving qualification"); + return; + } invalidateRouting(); if (!_closed && !_fenced) { _description = "Native broker routing unavailable: " + e.getMessage() + "; query listeners disabled"; @@ -257,8 +427,159 @@ final class NativeBrokerParticipant implements AutoCloseable, ServiceStatus.Serv } private void invalidateRouting() { - _routingImages = Map.of(); - _routingGeneration++; + _admission.writeLock().lock(); + try { + _routingImages = Map.of(); + _queryRoutes = Map.of(); + _routingGeneration++; + _queryGeneration = Math.incrementExact(_queryGeneration); + } finally { + _admission.writeLock().unlock(); + } + } + + private void membershipUnavailable() { + _admission.writeLock().lock(); + try { + invalidateRouting(); + ServingMembership current = _servingMembership; + _servingMembership = new ServingMembership(current.view(), false, current.minimumRevision()); + } finally { + _admission.writeLock().unlock(); + } + } + + @Nullable + private LeaseObservation beginLeaseObservation() { + _admission.writeLock().lock(); + try { + checkState(!_closed, "Native broker stopped before routing hydration"); + return _serving == null ? null : _serving.beginLeaseObservation(); + } finally { + _admission.writeLock().unlock(); + } + } + + private RoutingCapture captureRouting(NativeBrokerRoutingImage image, @Nullable LeaseObservation observation) { + Map<String, PeerLeaseBound> bounds = new HashMap<>(); + _admission.writeLock().lock(); + try { + checkState(!_closed, "Native broker stopped during routing hydration"); + if (_serving != null) { + var control = image.control(); + for (var server : image.serverAuthorities().values()) { + bounds.put(server.getParticipant().getParticipantId(), _serving.boundPeerLease( + requireNonNull(observation), control.getRound(), server.getExpiryRound(), + control.getPolicy().getMinTickIntervalMillis(), control.getPolicy().getPermitLifetimeMillis())); + } + } + } finally { + _admission.writeLock().unlock(); + } + return new RoutingCapture(image, Map.copyOf(bounds)); + } + + private boolean sameRoutes(Map<String, RoutingCapture> next) { + if (!_queryRoutes.keySet().equals(next.keySet())) { + return false; + } + return next.entrySet().stream().allMatch(entry -> { + NativeBrokerRoutingImage previous = _queryRoutes.get(entry.getKey()).image(); + NativeBrokerRoutingImage image = entry.getValue().image(); + return previous.queryIdentity().equals(image.queryIdentity()) + && previous.control().getControlEpoch().equals(image.control().getControlEpoch()) + && previous.control().getPolicy().getVersion() == image.control().getPolicy().getVersion(); + }); + } + + /// Originate once at authenticated request entry, before planning. Retries retain the returned deadline. + NativeQueryDeadline originateDeadline(long requestedMillis) { + _admission.readLock().lock(); + try { + serving(); + return requireNonNull(_servingSettings).queryDeadlines().originate(requestedMillis); + } finally { + _admission.readLock().unlock(); + } + } + + QueryAdmission admitQuery(String table, UUID queryId, UUID attemptId, NativeQueryDeadline deadline) { + _admission.readLock().lock(); + try { + ServingPermitGuard guard = serving(); + QueryTicket ticket = guard.admit(ReadOnlyCapability.BROKER_QUERY, deadline, + requireNonNull(_servingSettings).queryDeadlines()); + RoutingCapture capture = _queryRoutes.get(table); + checkState(capture != null && capture.image().hasPlanningMetadata(), + "Complete native query planning metadata is unavailable"); + checkControl(capture.image()); + var requests = capture.image().requests(queryId, attemptId, deadline); + List<PeerLeaseBound> leases = new ArrayList<>(requests.size()); + for (var request : requests) { + PeerLeaseBound bound = capture.serverLeases().get(request.context().server().participantId()); + guard.checkPeerLease(bound); + leases.add(bound); + } + guard.checkDelivery(ticket); + return new QueryAdmission(capture.image(), requests, leases, ticket, _queryGeneration); + } finally { + _admission.readLock().unlock(); + } + } + + void checkQueryDelivery(QueryAdmission admission) { + _admission.readLock().lock(); + try { + validateDelivery(admission); + } finally { + _admission.readLock().unlock(); + } + } + + /// The callback must be bounded and must not call lifecycle methods or acquire membership/routing monitors. + /// It executes under the admission read lock so a committed local revocation cannot race check-then-delivery. + void deliverQuery(QueryAdmission admission, Runnable callback) { + requireNonNull(callback); + _admission.readLock().lock(); + try { + validateDelivery(admission); + callback.run(); + } finally { + _admission.readLock().unlock(); + } + } + + private void validateDelivery(QueryAdmission admission) { + requireNonNull(admission); + ServingPermitGuard guard = serving(); + checkState(admission._generation == _queryGeneration, "Native query routing was invalidated"); + checkControl(admission._image); + guard.checkDelivery(admission._ticket); + for (PeerLeaseBound bound : admission._serverLeases) { + guard.checkPeerLease(bound); + } + // Peer-clock observations can consume time; recheck the original ticket immediately before the callback. + guard.checkDelivery(admission._ticket); + } + + private void checkControl(NativeBrokerRoutingImage image) { + MembershipView own = servingView(); + checkState(image.control().getControlEpoch().equals(own.control()) + && image.control().getPolicy().getVersion() == own.policyVersion(), + "Native query routing and serving control differ"); + } + + private ServingPermitGuard serving() { + checkState(!_closed && _serving != null, "Native broker serving requires explicit qualified clock inputs"); + return _serving; + } + + private MembershipView servingView() { + ServingMembership current = _servingMembership; + checkState(!_closed && current.available() && current.view() != null + && current.view().revision().getIndex() >= current.minimumRevision(), + "Coherent native broker serving membership is unavailable"); + return current.view(); } private String routingDescription() { @@ -273,9 +594,15 @@ final class NativeBrokerParticipant implements AutoCloseable, ServiceStatus.Serv } private void validateResponse(MetadataRpc.MembershipResponse response) throws IOException { - NativeOfflineSegmentWorker.validateMessage(response); + try { + NativeOfflineSegmentWorker.validateMessage(response); + } catch (IOException e) { + throw new IllegalArgumentException("Malformed broker membership response", e); + } checkArgument(response.hasRevision() && response.getRevision().getHistoryId().equals(_config.historyId().toString()) - && response.getRevision().getIndex() >= _appliedIndex, "Broker response history or revision differs"); + && response.getRevision().getIndex() >= _appliedIndex + && response.getRevision().getIndex() >= _servingMembership.minimumRevision(), + "Broker response history or revision differs"); } private void validateToken(MetadataRpc.ParticipantToken token) { @@ -287,20 +614,102 @@ final class NativeBrokerParticipant implements AutoCloseable, ServiceStatus.Serv "Invalid broker incarnation"); } - private void validateOwnProjection(MetadataRpc.MembershipResponse response) { + private MembershipView validateOwnProjection(MetadataRpc.MembershipResponse response) { checkArgument(response.hasControl() && response.hasParticipant(), "Broker membership projection is incomplete"); MetadataRpc.MembershipControl control = response.getControl(); MetadataRpc.ParticipantState state = response.getParticipant(); - checkArgument(state.getParticipant().equals(_participant) && state.getCapabilitiesCount() == 0 - && state.getEndpointsCount() == 0 && !state.getReady() && !state.getDraining() && !state.getRetired(), + boolean queryCapability = state.getCapabilitiesList().equals(List.of(MetadataRpc.ServingCapability.BROKER_QUERY)); + checkArgument(state.getParticipant().equals(_participant) + && (state.getCapabilitiesCount() == 0 || _servingSettings != null && queryCapability) + && state.getEndpointsCount() == 0 && !state.getRetired() && !(state.getReady() && state.getDraining()) + && (_servingSettings != null || !state.getReady() && !state.getDraining()), "Broker membership identity, capabilities or lifecycle changed"); checkArgument(control.hasControlEpoch() && control.hasPolicy() && control.getRound() >= 0 && control.getControlEpoch().getHistoryId().equals(_config.historyId().toString()) - && control.getControlEpoch().getRaftTerm() > 0 && control.getClockBarrierIndex() > 0 + && control.getControlEpoch().getRaftTerm() > 0 && control.getControlEpoch().getBarrierIndex() > 0 + && !control.getControlEpoch().getLeaderId().isBlank() + && control.getClockBarrierIndex() >= control.getControlEpoch().getBarrierIndex() && control.getClockBarrierIndex() <= response.getRevision().getIndex() && state.getExpiryRound() > control.getRound() && state.getLastModifiedIndex() > 0 - && state.getLastModifiedIndex() <= response.getRevision().getIndex(), "Broker membership lease is not live"); + && state.getLastModifiedIndex() <= response.getRevision().getIndex() + && state.getLastRenewalSequence() >= 0 && state.hasAppliedRevision() + && state.getAppliedRevision().getHistoryId().equals(_config.historyId().toString()) + && state.getAppliedRevision().getIndex() >= 0 + && state.getAppliedRevision().getIndex() < state.getLastModifiedIndex(), + "Broker membership lease is not live"); + String activation = control.getControlEpoch().getActivationId(); + checkArgument(UUID.fromString(activation).toString().equals(activation), "Invalid broker control activation"); + var policy = control.getPolicy(); + checkArgument(policy.getVersion() > 0 && policy.getMinTickIntervalMillis() > 0 && policy.getLeaseRounds() >= 2 + && policy.getPermitLifetimeMillis() > 0 && policy.getMaxQueryLifetimeMillis() > 0 + && policy.getMaxRevisionLag() > 0 + && (policy.getPermitLifetimeMillis() - 1) / policy.getMinTickIntervalMillis() < policy.getLeaseRounds() - 1 + && control.getMinimumTickDelayMillis() >= policy.getMinTickIntervalMillis() + && control.getMaxConfiguredPermitLifetimeMillis() >= policy.getPermitLifetimeMillis() + && (control.getMaxConfiguredPermitLifetimeMillis() == policy.getPermitLifetimeMillis() + || control.getMinimumTickDelayMillis() >= control.getMaxConfiguredPermitLifetimeMillis()), + "Invalid broker serving policy or timing barrier"); _appliedIndex = response.getRevision().getIndex(); + return new MembershipView(response.getRevision(), control.getControlEpoch(), state.getParticipant(), true, + state.getReady(), state.getDraining(), queryCapability ? Set.of(ReadOnlyCapability.BROKER_QUERY) : Set.of(), + policy.getVersion()); + } + + private void observeServingResponse(MetadataRpc.MembershipResponse response) { + if (_servingSettings == null) { + return; + } + _admission.writeLock().lock(); + try { + ServingMembership current = _servingMembership; + _servingMembership = new ServingMembership(current.view(), false, + Math.max(current.minimumRevision(), response.getRevision().getIndex())); + } finally { + _admission.writeLock().unlock(); + } + } + + private void publishMembership(MembershipView view, boolean available, @Nullable MetadataRpc.ServingPermit permit) { + _admission.writeLock().lock(); + try { + if (_closed) { + return; + } + ServingMembership current = _servingMembership; + MembershipView previous = current.view(); + checkArgument(view.revision().getIndex() >= current.minimumRevision() + && (previous == null || view.revision().getIndex() >= previous.revision().getIndex() + && (!view.revision().equals(previous.revision()) || view.equals(previous))), + "Broker membership image regressed or conflicts at the same revision"); + if (previous != null) { + boolean beganDrain = !previous.draining() && view.draining(); + boolean sameAuthority = previous.participant().equals(view.participant()) + && previous.control().equals(view.control()) && previous.live() == view.live() + && previous.capabilities().equals(view.capabilities()) && previous.policyVersion() == view.policyVersion() + && (previous.ready() == view.ready() && previous.draining() == view.draining() + || beganDrain && previous.ready()); + if (!sameAuthority) { + invalidateRouting(); + } else if (beganDrain) { + // Discard a pre-drain hydration cycle without revoking finite, already admitted work. + _routingGeneration++; + } + } + _servingMembership = new ServingMembership(view, available, current.minimumRevision()); + if (permit != null) { + ServingPermitGuard guard = requireNonNull(_serving); + try { + guard.accept(permit); + } catch (RuntimeException e) { + // Do not expose the new view with an old grant between this lock release and lifecycle fencing. + _servingMembership = new ServingMembership(view, false, current.minimumRevision()); + guard.invalidate(); + throw e; + } + } + } finally { + _admission.writeLock().unlock(); + } } private MetadataRpc.ParticipantSession session() { @@ -316,7 +725,15 @@ final class NativeBrokerParticipant implements AutoCloseable, ServiceStatus.Serv private void fence(String detail) { _fenced = true; _membershipHealthy = false; - invalidateRouting(); + _admission.writeLock().lock(); + try { + membershipUnavailable(); + if (_serving != null) { + _serving.invalidate(); + } + } finally { + _admission.writeLock().unlock(); + } _description = detail + "; query listeners disabled"; _heartbeats.shutdown(); _routingPoller.shutdownNow(); @@ -334,26 +751,41 @@ final class NativeBrokerParticipant implements AutoCloseable, ServiceStatus.Serv } @Override - public synchronized void close() { - if (_closed) { - return; + public void close() { + // Fence locally before waiting for the membership monitor, whose heartbeat may be blocked in an RPC. + _admission.writeLock().lock(); + try { + _closed = true; + ServingMembership current = _servingMembership; + _servingMembership = new ServingMembership(current.view(), false, current.minimumRevision()); + _queryRoutes = Map.of(); + if (_serving != null) { + _serving.invalidate(); + } + _description = "Native broker stopped; query listeners disabled"; + } finally { + _admission.writeLock().unlock(); } - _closed = true; - invalidateRouting(); - _description = "Native broker stopped; query listeners disabled"; _heartbeats.shutdownNow(); _routingPoller.shutdownNow(); - _routingConnection.close(); - try { - if (_participant != null) { - try { - _connection.membershipExecute(command().setUnregister(session()).build(), _config.rpcTimeoutMillis()); - } catch (RuntimeException e) { - LOGGER.warn("Could not confirm native broker unregister; its lease must expire", e); + synchronized (this) { + if (_cleanedUp) { + return; + } + _cleanedUp = true; + invalidateRouting(); + _routingConnection.close(); + try { + if (_participant != null) { + try { + _connection.membershipExecute(command().setUnregister(session()).build(), _config.rpcTimeoutMillis()); + } catch (RuntimeException e) { + LOGGER.warn("Could not confirm native broker unregister; its lease must expire", e); + } } + } finally { + _connection.close(); } - } finally { - _connection.close(); } } } 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 da258c9ffd7..7ddff4b2250 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 @@ -18,6 +18,7 @@ */ package org.apache.pinot.broker.broker.helix; +import com.google.protobuf.ByteString; import io.grpc.Status; import java.io.IOException; import java.nio.file.Files; @@ -25,18 +26,30 @@ import java.nio.file.Path; import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Deque; +import java.util.HashMap; 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 java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import org.apache.pinot.common.coordination.MetadataRpcConnection; import org.apache.pinot.common.coordination.NativeMetadataConfig; +import org.apache.pinot.common.coordination.QueryDeadlineGuard; +import org.apache.pinot.common.coordination.ServingPermitGuard.ClockReading; +import org.apache.pinot.common.coordination.ServingPermitGuard.QualifiedLifetimePolicy; +import org.apache.pinot.common.request.context.NativeQueryDeadline; import org.apache.pinot.common.utils.ServiceStatus; import org.apache.pinot.coordination.ratis.rpc.MetadataRpc; import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.ResultStatus; +import org.apache.pinot.spi.config.table.TableType; +import org.apache.pinot.spi.data.FieldSpec.DataType; +import org.apache.pinot.spi.data.Schema; import org.apache.pinot.spi.env.PinotConfiguration; +import org.apache.pinot.spi.utils.JsonUtils; +import org.apache.pinot.spi.utils.builder.TableConfigBuilder; import org.testng.annotations.Test; import static org.mockito.ArgumentMatchers.any; @@ -174,6 +187,7 @@ public class NativeBrokerParticipantTest { MetadataRpc.ParticipantToken original = fixture._token; fixture._token = original.toBuilder().setIncarnation(UUID.randomUUID().toString()) .setRegistrationEpoch(original.getRegistrationEpoch() + 1).build(); + fixture._participantModifiedIndex = ++fixture._index; fixture._participant.heartbeat(); assertFalse(fixture._requests.stream().anyMatch(MetadataRpc.MembershipCommandRequest::hasRenew)); fixture._participant.close(); @@ -199,10 +213,15 @@ public class NativeBrokerParticipantTest { fixture._participant.start(); fixture._participant.refreshRouting(); long routingIndex = fixture._participant.routingImage("orders_OFFLINE").revision().getIndex(); + // A newer routing read completes while the heartbeat's earlier own-membership read is still in flight. + fixture._afterReadAction = fixture._participant::refreshRouting; fixture._participant.heartbeat(); + fixture._afterReadAction = () -> { }; var renewal = fixture._requests.getLast().getRenew(); - assertTrue(renewal.getAppliedRevision().getIndex() < routingIndex, - "Per-table routing captures must not advance the membership-only applied revision"); + assertEquals(renewal.getAppliedRevision().getIndex(), fixture._lastOwnReadIndex); + assertTrue(renewal.getAppliedRevision().getIndex() + < fixture._participant.routingImage("orders_OFFLINE").revision().getIndex(), + "A later routing read must not rewrite the already captured own-membership applied revision"); fixture._routingFailure = true; fixture._participant.refreshRouting(); assertThrows(IllegalStateException.class, () -> fixture._participant.routingImage("orders_OFFLINE")); @@ -249,6 +268,265 @@ public class NativeBrokerParticipantTest { } } + @Test(timeOut = 10_000) + public void testExplicitServingCapturesPlanningAndRenewalCannotExtendOldAdmission() throws Exception { + try (Fixture fixture = new Fixture(true)) { + fixture._participant.start(); + fixture._participant.refreshRouting(); + assertThrows(IllegalStateException.class, fixture::admit); + fixture._participant.heartbeat(); + fixture._participant.refreshRouting(); + var original = fixture.admit(); + assertEquals(original.requests().size(), 1); + assertEquals(original.requests().getFirst().context().requiredNames(), List.of("segment")); + assertEquals(original.planningMetadata(false).tableCache().getSchema("orders").getSchemaName(), "orders"); + assertTrue(fixture._requests.stream().anyMatch(MetadataRpc.MembershipCommandRequest::hasReady)); + assertEquals(fixture._requests.getFirst().getRegister().getEndpointsCount(), 0); + AtomicInteger delivered = new AtomicInteger(); + fixture._participant.deliverQuery(original, delivered::incrementAndGet); + assertEquals(delivered.get(), 1); + fixture._millis.set(50); + fixture._participant.heartbeat(); + fixture._participant.refreshRouting(); + fixture._participant.checkQueryDelivery(original); + var renewed = fixture.admit(); + fixture._millis.set(201); + assertThrows(IllegalStateException.class, + () -> fixture._participant.deliverQuery(original, delivered::incrementAndGet)); + assertEquals(delivered.get(), 1, "Rejected delivery must never invoke the callback"); + fixture._participant.deliverQuery(renewed, delivered::incrementAndGet); + assertEquals(delivered.get(), 2); + assertEquals(fixture._participant.getServiceStatus(), ServiceStatus.Status.BAD, + "Explicit authority does not start or report a query listener"); + } + try (Fixture fixture = new Fixture(true)) { + fixture.startServing(); + NativeQueryDeadline beforePlanning = fixture._participant.originateDeadline(100); + fixture._millis.set(101); + fixture._participant.heartbeat(); + fixture._participant.refreshRouting(); + assertThrows(IllegalStateException.class, () -> fixture._participant.admitQuery("orders_OFFLINE", + UUID.randomUUID(), UUID.randomUUID(), beforePlanning)); + fixture._participant.checkQueryDelivery(fixture.admit()); + } + } + + @Test(timeOut = 10_000) + public void testPostRenewalReadClosesAuthorityWithoutBlockingLocalChecks() throws Exception { + try (Fixture fixture = new Fixture(true); var executor = Executors.newFixedThreadPool(2)) { + fixture.startServing(); + var original = fixture.admit(); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + fixture._postRenewalReadAction = () -> awaitRelease(entered, release); + var heartbeat = executor.submit(fixture._participant::heartbeat); + try { + assertTrue(entered.await(1, TimeUnit.SECONDS)); + executor.submit(() -> assertThrows(IllegalStateException.class, + () -> fixture._participant.checkQueryDelivery(original))).get(1, TimeUnit.SECONDS); + executor.submit(() -> assertThrows(IllegalStateException.class, fixture::admit)).get(1, TimeUnit.SECONDS); + } finally { + release.countDown(); + } + heartbeat.get(1, TimeUnit.SECONDS); + fixture._participant.refreshRouting(); + fixture._participant.checkQueryDelivery(fixture.admit()); + List<Long> sequences = fixture._requests.stream().filter(MetadataRpc.MembershipCommandRequest::hasRenew) + .map(request -> request.getRenew().getSequence()).toList(); + assertEquals(sequences, List.of(1L, 2L)); + } + try (Fixture fixture = new Fixture(true)) { + fixture.startServing(); + var original = fixture.admit(); + fixture._postRenewalReadAction = () -> { + throw Status.UNAVAILABLE.asRuntimeException(); + }; + fixture._participant.heartbeat(); + assertThrows(IllegalStateException.class, () -> fixture._participant.checkQueryDelivery(original)); + fixture._postRenewalReadAction = () -> { }; + fixture._participant.heartbeat(); + fixture._participant.refreshRouting(); + fixture._participant.checkQueryDelivery(fixture.admit()); + assertThrows(IllegalStateException.class, () -> fixture._participant.checkQueryDelivery(original)); + assertEquals(fixture._requests.stream().filter(MetadataRpc.MembershipCommandRequest::hasRenew) + .map(request -> request.getRenew().getSequence()).toList(), List.of(1L, 2L, 3L), + "A failed follow-up read must not replay an already consumed renewal"); + } + } + + @Test(timeOut = 10_000) + public void testBlockedMembershipRpcDoesNotBlockDeliveryAndCloseFencesBeforeUnregister() throws Exception { + try (Fixture fixture = new Fixture(true); var executor = Executors.newFixedThreadPool(3)) { + fixture.startServing(); + var admission = fixture.admit(); + fixture._unregisterAction = () -> assertThrows(IllegalStateException.class, + () -> fixture._participant.checkQueryDelivery(admission)); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + fixture._readAction = () -> awaitRelease(entered, release); + var heartbeat = executor.submit(fixture._participant::heartbeat); + try { + assertTrue(entered.await(1, TimeUnit.SECONDS)); + AtomicInteger delivered = new AtomicInteger(); + executor.submit(() -> fixture._participant.deliverQuery(admission, delivered::incrementAndGet)) + .get(1, TimeUnit.SECONDS); + assertEquals(delivered.get(), 1, "Delivery must not acquire the RPC-holding lifecycle monitor"); + var close = executor.submit(fixture._participant::close); + // The status projection is local and must become closed even while heartbeat owns the lifecycle monitor. + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(1); + while (fixture._participant.getServiceStatus() != ServiceStatus.Status.SHUTTING_DOWN + && System.nanoTime() < deadline) { + Thread.onSpinWait(); + } + assertEquals(fixture._participant.getServiceStatus(), ServiceStatus.Status.SHUTTING_DOWN); + executor.submit(() -> assertThrows(IllegalStateException.class, + () -> fixture._participant.deliverQuery(admission, delivered::incrementAndGet))) + .get(1, TimeUnit.SECONDS); + release.countDown(); + close.get(1, TimeUnit.SECONDS); + assertEquals(delivered.get(), 1); + } finally { + release.countDown(); + } + heartbeat.get(1, TimeUnit.SECONDS); + fixture.assertOwnUnregister(); + } + } + + @Test(timeOut = 10_000) + public void testPeerLeaseStartsBeforeRoutingRpcAndCannotBeExtendedByRefresh() throws Exception { + try (Fixture fixture = new Fixture(true)) { + fixture._participant.start(); + fixture._participant.heartbeat(); + fixture._serverExpiryRound = 2; + fixture._routingAdvanceMillis = 101; + fixture._participant.refreshRouting(); + // A delayed response cannot restart the server's one-full-round lease budget. + assertThrows(IllegalStateException.class, fixture::admit); + fixture._routingAdvanceMillis = 0; + fixture._serverExpiryRound = 3; + fixture._participant.refreshRouting(); + fixture._participant.checkQueryDelivery(fixture.admit()); + } + try (Fixture fixture = new Fixture(true)) { + fixture._serverExpiryRound = 2; + fixture.startServing(); + var original = fixture.admit(); + fixture._millis.set(50); + fixture._participant.heartbeat(); + fixture._serverExpiryRound = 3; + fixture._participant.refreshRouting(); + var refreshed = fixture.admit(); + fixture._millis.set(101); + assertThrows(IllegalStateException.class, () -> fixture._participant.checkQueryDelivery(original)); + fixture._participant.checkQueryDelivery(refreshed); + } + } + + @Test(timeOut = 10_000) + public void testRoutingRemovalAndFailureCannotResurrectOldAdmissions() throws Exception { + try (Fixture fixture = new Fixture(true)) { + fixture.startServing(); + var original = fixture.admit(); + fixture._routingTables = List.of(); + fixture._participant.refreshRouting(); + assertThrows(IllegalStateException.class, () -> fixture._participant.checkQueryDelivery(original)); + fixture._routingTables = List.of("orders_OFFLINE"); + fixture._participant.refreshRouting(); + var replacement = fixture.admit(); + assertThrows(IllegalStateException.class, () -> fixture._participant.checkQueryDelivery(original)); + fixture._routingFailure = true; + fixture._participant.refreshRouting(); + fixture._routingFailure = false; + fixture._participant.refreshRouting(); + assertThrows(IllegalStateException.class, () -> fixture._participant.checkQueryDelivery(replacement)); + var beforeEndpointChange = fixture.admit(); + String endpoint = "grpcs://server-replaced:8090"; + var changed = fixture._table.toBuilder(); + changed.getSegmentsBuilder(0).getReplicasBuilder(0).setEndpoint(endpoint); + changed.getServerAuthoritiesBuilder(0).putEndpoints("queryGrpc", endpoint); + fixture._table = changed.build(); + fixture._participant.refreshRouting(); + assertThrows(IllegalStateException.class, () -> fixture._participant.checkQueryDelivery(beforeEndpointChange)); + fixture._participant.checkQueryDelivery(fixture.admit()); + } + } + + @Test(timeOut = 10_000) + public void testLostGrantKeepsFirstSendAndOldControlReplayCannotRestoreAuthority() throws Exception { + try (Fixture fixture = new Fixture(true)) { + fixture.startServing(); + fixture._millis.set(50); + fixture._renewals.add("lost-grant"); + fixture._participant.heartbeat(); + fixture._millis.set(240); + fixture._participant.heartbeat(); + fixture._participant.refreshRouting(); + var admitted = fixture.admit(); + var requests = fixture._requests.stream().filter(MetadataRpc.MembershipCommandRequest::hasRenew).toList(); + assertEquals(requests.get(1), requests.get(2), "Retry must keep the original sequence and applied revision"); + fixture._millis.set(251); + assertThrows(IllegalStateException.class, () -> fixture._participant.checkQueryDelivery(admitted)); + assertThrows(IllegalStateException.class, fixture::admit); + } + try (Fixture fixture = new Fixture(true)) { + fixture.startServing(); + var original = fixture.admit(); + fixture._renewals.add("lost-grant"); + fixture._participant.heartbeat(); + long barrier = fixture._index + 1; + fixture._control = fixture._control.toBuilder().setClockBarrierIndex(barrier) + .setControlEpoch(fixture._control.getControlEpoch().toBuilder().setRaftTerm(2).setBarrierIndex(barrier) + .setActivationId(UUID.randomUUID().toString())).build(); + fixture._participant.heartbeat(); + fixture._participant.refreshRouting(); + assertThrows(IllegalStateException.class, fixture::admit); + assertThrows(IllegalStateException.class, () -> fixture._participant.checkQueryDelivery(original)); + fixture._participant.heartbeat(); + fixture._participant.refreshRouting(); + fixture._participant.checkQueryDelivery(fixture.admit()); + assertThrows(IllegalStateException.class, () -> fixture._participant.checkQueryDelivery(original)); + } + } + + @Test(timeOut = 10_000) + public void testDrainPreservesFiniteDeliveryButConsumedNotReadyRevokesOldAdmissions() throws Exception { + try (Fixture fixture = new Fixture(true)) { + fixture.startServing(); + var original = fixture.admit(); + fixture._ready = false; + fixture._draining = true; + fixture._participantModifiedIndex = ++fixture._index; + fixture._participant.heartbeat(); + fixture._participant.refreshRouting(); + assertThrows(IllegalStateException.class, fixture::admit); + fixture._participant.checkQueryDelivery(original); + fixture._millis.set(201); + assertThrows(IllegalStateException.class, () -> fixture._participant.checkQueryDelivery(original)); + } + try (Fixture fixture = new Fixture(true)) { + fixture.startServing(); + var original = fixture.admit(); + fixture._renewals.add(ResultStatus.NOT_READY); + fixture._participant.heartbeat(); + assertThrows(IllegalStateException.class, fixture::admit); + fixture._participant.heartbeat(); + fixture._participant.refreshRouting(); + fixture._participant.checkQueryDelivery(fixture.admit()); + assertThrows(IllegalStateException.class, () -> fixture._participant.checkQueryDelivery(original)); + } + } + + private static void awaitRelease(CountDownLatch entered, CountDownLatch release) { + entered.countDown(); + try { + assertTrue(release.await(3, TimeUnit.SECONDS)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e); + } + } + private static final class Fixture implements AutoCloseable { private final String _history = UUID.randomUUID().toString(); private final String _id = "Broker_native_" + UUID.randomUUID(); @@ -259,15 +537,37 @@ public class NativeBrokerParticipantTest { private final NativeBrokerParticipant _participant; private final List<MetadataRpc.MembershipCommandRequest> _requests = new ArrayList<>(); private final Deque<Object> _renewals = new ArrayDeque<>(); - private final MetadataRpc.MembershipControl _control; + private final Map<Long, MetadataRpc.ServingPermit> _grants = new HashMap<>(); + private final AtomicLong _millis = new AtomicLong(); + private final boolean _serving; + private MetadataRpc.TableRouting _table; + private MetadataRpc.MembershipControl _control; private MetadataRpc.ParticipantToken _token; private long _index; - private long _routingIndex = 100; + private long _appliedIndex; + private long _participantModifiedIndex; + private long _lastOwnReadIndex; + private long _renewalSequence; private List<String> _routingTables = List.of("orders_OFFLINE"); private boolean _routingFailure; private boolean _grantCapabilities; + private boolean _ready; + private boolean _draining; + private long _serverExpiryRound = 3; + private long _routingAdvanceMillis; + private boolean _followupRead; + private Runnable _readAction = () -> { }; + private Runnable _afterReadAction = () -> { }; + private Runnable _postRenewalReadAction = () -> { }; + private Runnable _unregisterAction = () -> { }; private Fixture() throws Exception { + this(false); + } + + private Fixture(boolean serving) throws Exception { + _serving = serving; + _grantCapabilities = serving; _configuration = new PinotConfiguration(Map.of("pinot.coordination.backend", "RATIS", NativeMetadataConfig.PREFIX + "historyId", _history, NativeMetadataConfig.PREFIX + "participantId", _id, @@ -283,12 +583,20 @@ public class NativeBrokerParticipantTest { .setPolicy(MetadataRpc.MembershipPolicy.newBuilder().setVersion(1).setMinTickIntervalMillis(100) .setLeaseRounds(3).setPermitLifetimeMillis(200).setMaxQueryLifetimeMillis(300).setMaxRevisionLag(1000)) .build(); + _table = planningTable(); when(_connection.membershipRead(any(), anyLong())).thenAnswer(invocation -> { + _readAction.run(); + if (_followupRead) { + _followupRead = false; + _postRenewalReadAction.run(); + } MetadataRpc.MembershipResponse.Builder response = response(_token == null ? ResultStatus.NOT_FOUND : ResultStatus.OK); if (_token != null) { response.setParticipant(state()); } + _lastOwnReadIndex = response.getRevision().getIndex(); + _afterReadAction.run(); return response.build(); }); when(_connection.membershipExecute(any(), anyLong())).thenAnswer(invocation -> { @@ -298,7 +606,9 @@ public class NativeBrokerParticipantTest { _token = MetadataRpc.ParticipantToken.newBuilder().setHistoryId(_history).setParticipantId(_id) .setRole(MetadataRpc.ParticipantRole.BROKER).setIncarnation(request.getRegister().getIncarnation()) .setRegistrationEpoch(request.getRegister().getExpectedEpoch() + 1).build(); - return response(ResultStatus.OK).setParticipant(state()).build(); + var response = response(ResultStatus.OK); + _participantModifiedIndex = _index; + return response.setParticipant(state()).build(); } if (request.hasRenew()) { Object next = _renewals.isEmpty() ? ResultStatus.OK : _renewals.removeFirst(); @@ -310,14 +620,51 @@ public class NativeBrokerParticipantTest { } MetadataRpc.RenewalResult.Builder renewal = MetadataRpc.RenewalResult.newBuilder() .setParticipant(_token).setSequence(request.getRenew().getSequence()); + var response = response(ResultStatus.OK); if (next.equals("unexpected-permit")) { renewal.setStatus(ResultStatus.OK).setPermit(MetadataRpc.ServingPermit.getDefaultInstance()); } else { - renewal.setStatus((ResultStatus) next); + renewal.setStatus(next.equals("lost-grant") ? ResultStatus.OK : (ResultStatus) next); + if (_serving && renewal.getStatus() == ResultStatus.OK) { + long sequence = request.getRenew().getSequence(); + var grant = _grants.get(sequence); + if (grant == null && _ready && !_draining) { + grant = MetadataRpc.ServingPermit.newBuilder().setParticipant(_token).setRenewalSequence(sequence) + .setControlEpoch(_control.getControlEpoch()).setPolicyVersion(_control.getPolicy().getVersion()) + .setMinAppliedRevision(response.getRevision()).setPermitLifetimeMillis(200) + .setMaxQueryLifetimeMillis(300).addCapabilities(MetadataRpc.ServingCapability.BROKER_QUERY) + .build(); + _grants.put(sequence, grant); + } + if (grant != null) { + renewal.setPermit(grant); + } + } + } + if (request.getRenew().getSequence() > _renewalSequence) { + if (renewal.getStatus() == ResultStatus.OK) { + _appliedIndex = request.getRenew().getAppliedRevision().getIndex(); + } + _renewalSequence = request.getRenew().getSequence(); + _participantModifiedIndex = _index; } - return response(ResultStatus.OK).addRenewals(renewal).build(); + _followupRead = _serving; + if (next.equals("lost-grant")) { + throw Status.DEADLINE_EXCEEDED.asRuntimeException(); + } + return response.addRenewals(renewal).build(); + } + if (request.hasReady()) { + assertTrue(_serving, "Default brokers must not set ready"); + assertEquals(request.getReady().getSession().getIncarnation(), _token.getIncarnation()); + _ready = request.getReady().getReady(); + _appliedIndex = request.getReady().getAppliedRevision().getIndex(); + var response = response(ResultStatus.OK); + _participantModifiedIndex = _index; + return response.setParticipant(state()).build(); } assertTrue(request.hasUnregister(), "Membership-only brokers must never set ready or advertise endpoints"); + _unregisterAction.run(); return response(ResultStatus.OK).build(); }); when(_routingConnection.routingSnapshot(any(), anyLong())).thenAnswer(invocation -> { @@ -326,8 +673,14 @@ public class NativeBrokerParticipantTest { } return routingResponse(invocation.getArgument(0)); }); - _participant = new NativeBrokerParticipant(new NativeMetadataConfig(_configuration), _connection, - _routingConnection); + var config = new NativeMetadataConfig(_configuration); + _participant = serving ? new NativeBrokerParticipant(config, _connection, _routingConnection, + new NativeBrokerParticipant.ServingSettings( + () -> new ClockReading(TimeUnit.MILLISECONDS.toNanos(_millis.get()), 0, true), + new QualifiedLifetimePolicy(200, 300, 1_000_000, 0, 0), + new QueryDeadlineGuard(() -> new QueryDeadlineGuard.TimeInterval( + 1000 + _millis.get(), 1000 + _millis.get(), true), "test-clock", 300, 0, 1_000_000))) + : new NativeBrokerParticipant(config, _connection, _routingConnection); } private MetadataRpc.RoutingSnapshot routingResponse(MetadataRpc.RoutingSnapshotRequest request) { @@ -335,9 +688,14 @@ public class NativeBrokerParticipantTest { 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)); + .setIndex(++_index)); if (request.getTableName().isEmpty()) { response.addAllTables(_routingTables); + } else if (_serving) { + _millis.addAndGet(_routingAdvanceMillis); + response.setTable(_table.toBuilder().setServerAuthorities(0, + _table.getServerAuthorities(0).toBuilder().setExpiryRound(_serverExpiryRound) + .setLastRenewalSequence(_renewalSequence).setLastModifiedIndex(_index))); } else { response.setTable(MetadataRpc.TableRouting.newBuilder().setTableName(request.getTableName()) .setSchemaName("orders") @@ -356,14 +714,65 @@ public class NativeBrokerParticipantTest { private MetadataRpc.ParticipantState state() { MetadataRpc.ParticipantState.Builder state = MetadataRpc.ParticipantState.newBuilder().setParticipant(_token) - .setExpiryRound(3).setLastModifiedIndex(1) - .setAppliedRevision(MetadataRpc.Revision.newBuilder().setHistoryId(_history).setIndex(0)); + .setExpiryRound(3).setLastModifiedIndex(_participantModifiedIndex).setReady(_ready).setDraining(_draining) + .setLastRenewalSequence(_renewalSequence) + .setAppliedRevision(MetadataRpc.Revision.newBuilder().setHistoryId(_history).setIndex(_appliedIndex)); if (_grantCapabilities) { state.addCapabilities(MetadataRpc.ServingCapability.BROKER_QUERY); } return state.build(); } + private MetadataRpc.TableRouting planningTable() throws IOException { + var server = MetadataRpc.ParticipantToken.newBuilder().setHistoryId(_history).setParticipantId("Server_native") + .setRole(MetadataRpc.ParticipantRole.SERVER).setIncarnation(UUID.randomUUID().toString()) + .setRegistrationEpoch(1).build(); + var table = object(); + var schema = object(); + var replica = MetadataRpc.ReplicaRouting.newBuilder().setServer(server).setAssignment(object()) + .setAssignmentEpoch(1).setMaterializationId(UUID.randomUUID().toString()) + .setInstallationId(UUID.randomUUID().toString()).setEndpoint("grpcs://server-native:8090"); + return MetadataRpc.TableRouting.newBuilder().setTableName("orders_OFFLINE").setSchemaName("orders") + .setTable(table).setSchema(schema) + .setTableConfig(record(table, "orders_OFFLINE", MetadataRpc.RecordKind.TABLE_CONFIG, + MetadataRpc.CatalogValue.newBuilder().setTableConfigJson(ByteString.copyFrom(JsonUtils.objectToBytes( + new TableConfigBuilder(TableType.OFFLINE).setTableName("orders").build()))).build())) + .setSchemaRecord(record(schema, "orders", MetadataRpc.RecordKind.SCHEMA, + MetadataRpc.CatalogValue.newBuilder().setSchemaJson(ByteString.copyFrom(JsonUtils.objectToBytes( + new Schema.SchemaBuilder().setSchemaName("orders") + .addSingleValueDimension("id", DataType.INT).build()))).build())) + .addSegments(MetadataRpc.SegmentRouting.newBuilder().setName("segment").setContent(object()) + .setContentSha256("a".repeat(64)).addReplicas(replica)) + .addServerAuthorities(MetadataRpc.ParticipantState.newBuilder().setParticipant(server).setReady(true) + .setExpiryRound(3).setLastModifiedIndex(1) + .setAppliedRevision(MetadataRpc.Revision.newBuilder().setHistoryId(_history).setIndex(1)) + .addCapabilities(MetadataRpc.ServingCapability.SERVER_QUERY) + .putEndpoints("queryGrpc", "grpcs://server-native:8090").putEndpoints("nativeQueryVersion", "2")) + .build(); + } + + private MetadataRpc.ObjectVersion object() { + return MetadataRpc.ObjectVersion.newBuilder().setHistoryId(_history) + .setObjectId(UUID.randomUUID().toString()).setVersion(1).build(); + } + + private MetadataRpc.CatalogRecord record(MetadataRpc.ObjectVersion identity, String name, + MetadataRpc.RecordKind kind, MetadataRpc.CatalogValue value) { + return MetadataRpc.CatalogRecord.newBuilder().setKind(kind).setName(name).setObjectId(identity.getObjectId()) + .setVersion(1).setSchemaVersion(1).setLastModifiedIndex(1).setValue(value).build(); + } + + private void startServing() throws IOException { + _participant.start(); + _participant.heartbeat(); + _participant.refreshRouting(); + } + + private NativeBrokerParticipant.QueryAdmission admit() { + return _participant.admitQuery("orders_OFFLINE", UUID.randomUUID(), UUID.randomUUID(), + _participant.originateDeadline(300)); + } + private void assertOwnUnregister() { MetadataRpc.MembershipCommandRequest last = _requests.getLast(); assertTrue(last.hasUnregister()); diff --git a/pinot-coordination-ratis/CONTROLLER-API-PARITY.md b/pinot-coordination-ratis/CONTROLLER-API-PARITY.md index aa803bd39b9..b0ab8dd3e46 100644 --- a/pinot-coordination-ratis/CONTROLLER-API-PARITY.md +++ b/pinot-coordination-ratis/CONTROLLER-API-PARITY.md @@ -39,8 +39,9 @@ The standard resource package has **264 annotated HTTP methods across 43 resourc - Native broker routing: authenticated, tenant-scoped table discovery and complete OFFLINE routing observations, immutable broker images and query-identity construction. Typed broker certificate/session/tenant grants now support local server admission and guarded gRPC delivery. Coherent table/schema planning inputs and a captured environment - in the shared single-stage compiler are available. Normal broker admission, dispatch and serving activation remain - outstanding; installed replicas without query authority are not selectable. + in the shared single-stage compiler are available. Optional broker admission now binds finite own/server leases, + deadlines and invalidation generations. Authenticated request-entry wiring, dispatch, final HTTP delivery fencing + 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 4f33741bfb2..88214c053d0 100644 --- a/pinot-coordination-ratis/README.md +++ b/pinot-coordination-ratis/README.md @@ -433,9 +433,34 @@ passes its table cache and routing manager through optimization, route construct environment uses the existing providers. Native callers can supply one captured image without replacing global cache state or copying the SQL compiler. -This is planning infrastructure, not a native query listener. Normal broker admission/deadline origination, -authenticated dispatch, receiving and final-response fences, quotas and qualified clock integration remain -required before activation. Unsupported native query forms must be rejected explicitly by that integration. +This is planning infrastructure, not a native query listener. Connecting authenticated request entry, dispatch, +receiving and final-response fences, quotas and qualified clock providers remains required before activation. +Unsupported native query forms must be rejected explicitly by that integration. + +## Broker admission lifecycle + +The broker participant accepts optional explicit serving settings with qualified local and shared clocks. +Its existing constructor and normal starters remain membership-only, without readiness, query endpoints or +query listeners. Supplying settings commits readiness but does not start a listener or manufacture a capability: +the authenticated participant policy must grant `BROKER_QUERY`. + +Each known renewal is followed by a coherent own-membership read before its grant can authorize work. The renewal +reply raises a minimum serving revision immediately; the earlier view cannot authorize delivery while that read is +outstanding. An unknown outcome retains the exact request and original first-send time. A consumed renewal keeps +its sequence even if the follow-up read fails. + +An admission retains its immutable table image, exact per-server identity envelopes, original query deadline, +finite broker ticket and selected-server lease bounds. Server bounds start before the table snapshot RPC, so network +delay consumes their budget. Renewals cannot extend an admitted attempt. Routing changes, removal, failed refreshes +and authority revocation invalidate older attempts permanently, including after equal-looking metadata returns. +Draining rejects new work while preserving already admitted work within its original bounds. + +Admission and delivery use a separate local lock and perform no metadata RPCs. A bounded delivery callback runs +under that lock, serialized with local invalidation; it must not call lifecycle operations or acquire their locks. +Shutdown fences locally before waiting for an in-flight membership RPC or unregistering. + +These APIs are not yet connected to authenticated query entry, normal network dispatch or final HTTP serialization. +Synthetic clocks in the focused tests establish lifecycle behavior, not production clock qualification. ## Authenticated native query admission --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
