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 00:23:08 UTC
(pinot) 01/01: Wire Ratis into normal starters with query listeners disabled
This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch xiangfu0/codex2/ratis-normal-starters
in repository https://gitbox.apache.org/repos/asf/pinot.git
commit c86ba95b6f4ac128dd33dbe597c0aedcbc22149c
Author: Xiang Fu <xi...@apache.org>
AuthorDate: Mon Oct 19 17:21:23 2026 -0700
Wire Ratis into normal starters with query listeners disabled
---
.../broker/broker/helix/BaseBrokerStarter.java | 67 ++++
.../helix/MultiClusterHelixBrokerStarter.java | 4 +
.../broker/helix/NativeBrokerParticipant.java | 241 +++++++++++++++
.../broker/helix/NativeBrokerParticipantTest.java | 290 ++++++++++++++++++
.../common/coordination/CoordinationBackend.java | 39 +++
.../common/coordination/MetadataRpcConnection.java | 27 +-
.../common/coordination/NativeMetadataConfig.java | 119 ++++++++
.../apache/pinot/common/utils/ServiceStatus.java | 5 +
.../coordination/NativeMetadataConfigTest.java | 69 +++++
pinot-controller/pom.xml | 4 +
.../pinot/controller/BaseControllerStarter.java | 40 +++
.../coordination/RatisControllerProvider.java | 52 ++++
.../coordination/RatisControllerRuntime.java | 251 +++++++++++++++
pinot-coordination-ratis/README.md | 87 +++++-
.../server/starter/NativeMetadataConnection.java | 101 +-----
.../starter/NativeServerStarterLifecycle.java | 217 +++++++++++++
.../pinot/server/starter/ServerInstance.java | 9 +
.../server/starter/helix/BaseServerStarter.java | 42 ++-
.../starter/helix/NativeServerStarterTest.java | 270 ++++++++++++++++
.../admin/command/StartServiceManagerCommand.java | 33 +-
.../pinot/tools/service/PinotServiceManager.java | 13 +-
.../apache/pinot/tools/utils/PinotConfigUtils.java | 7 +
.../command/StartServiceManagerCommandTest.java | 81 +++++
.../pinot/tools/service/RatisStartersTest.java | 339 +++++++++++++++++++++
.../pinot/tools/utils/PinotConfigUtilsTest.java | 20 ++
25 files changed, 2292 insertions(+), 135 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 878d08f2db8..13fb043f8f4 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
@@ -72,6 +72,8 @@ import org.apache.pinot.common.config.NettyConfig;
import org.apache.pinot.common.config.TlsConfig;
import org.apache.pinot.common.config.provider.TableCache;
import org.apache.pinot.common.config.provider.ZkTableCache;
+import org.apache.pinot.common.coordination.CoordinationBackend;
+import org.apache.pinot.common.coordination.NativeMetadataConfig;
import org.apache.pinot.common.cursors.AbstractResponseStore;
import org.apache.pinot.common.evaluator.GroovyFunctionEvaluator;
import org.apache.pinot.common.failuredetector.FailureDetector;
@@ -157,6 +159,14 @@ public abstract class BaseBrokerStarter implements ServiceStartable {
@VisibleForTesting
static final String WARMUP_GATE_STARTING_DESCRIPTION = "Warming up broker data plane";
+ private final Object _nativeLifecycleLock = new Object();
+ private boolean _nativeSelected;
+ private boolean _nativeStatusRegistered;
+ @Nullable
+ private NativeMetadataConfig _nativeConfig;
+ @Nullable
+ private NativeBrokerParticipant _nativeParticipant;
+
protected PinotConfiguration _brokerConf;
protected List<ListenerConfig> _listenerConfigs;
protected String _clusterName;
@@ -228,6 +238,22 @@ public abstract class BaseBrokerStarter implements ServiceStartable {
public void init(PinotConfiguration brokerConf)
throws Exception {
_brokerConf = brokerConf;
+ _nativeSelected = CoordinationBackend.from(brokerConf) == CoordinationBackend.RATIS;
+ if (_nativeSelected) {
+ NativeMetadataConfig nativeConfig = new NativeMetadataConfig(brokerConf);
+ _instanceId = nativeConfig.participantId();
+ Preconditions.checkArgument(InstanceTypeUtils.isBroker(_instanceId),
+ "Native broker participant ID must have prefix '%s'", Helix.PREFIX_OF_BROKER_INSTANCE);
+ // Preserve validation of the legacy instanceId alias accepted by the normal broker starter.
+ @SuppressWarnings("deprecation")
+ String configuredId = brokerConf.getProperty(Broker.CONFIG_OF_BROKER_ID,
+ brokerConf.getProperty(Helix.Instance.INSTANCE_ID_KEY));
+ Preconditions.checkArgument(configuredId == null || configuredId.equals(_instanceId),
+ "Broker ID differs from native participant ID");
+ brokerConf.setProperty(Broker.CONFIG_OF_BROKER_ID, _instanceId);
+ _nativeConfig = nativeConfig;
+ return;
+ }
// Remove all white-spaces from the list of zkServers (if any).
_zkServers = brokerConf.getProperty(Helix.CONFIG_OF_ZOOKEEPER_SERVER).replaceAll("\\s+", "");
_clusterName = brokerConf.getProperty(Helix.CONFIG_OF_CLUSTER_NAME);
@@ -277,6 +303,11 @@ public abstract class BaseBrokerStarter implements ServiceStartable {
NettyInspector.logAllChecks();
}
+ @VisibleForTesting
+ NativeBrokerParticipant createNativeParticipant(NativeMetadataConfig config) throws IOException {
+ return new NativeBrokerParticipant(config, config.openConnection());
+ }
+
/// Can be overridden to apply custom configs to the broker conf.
protected void applyCustomConfigs(PinotConfiguration brokerConf) {
}
@@ -409,6 +440,29 @@ public abstract class BaseBrokerStarter implements ServiceStartable {
@Override
public void start()
throws Exception {
+ if (_nativeSelected) {
+ synchronized (_nativeLifecycleLock) {
+ Preconditions.checkState(_nativeConfig != null, "Native broker initialization did not complete");
+ Preconditions.checkState(_nativeParticipant == null && !_isShuttingDown,
+ "Native broker lifecycle cannot be restarted");
+ _isStarting = true;
+ try {
+ _nativeParticipant = createNativeParticipant(_nativeConfig);
+ _nativeParticipant.start();
+ ServiceStatus.setServiceStatusCallback(_instanceId, _nativeParticipant);
+ _nativeStatusRegistered = true;
+ LOGGER.info("Native broker {} registered for membership only; query listeners remain disabled", _instanceId);
+ } catch (Exception | Error e) {
+ if (_nativeParticipant != null) {
+ _nativeParticipant.close();
+ }
+ throw e;
+ } finally {
+ _isStarting = false;
+ }
+ }
+ return;
+ }
LOGGER.info("Starting Pinot broker (Version: {})", PinotVersion.VERSION);
LOGGER.info("Broker configs: {}", new PinotAppConfigs(getConfig()).toJSONString());
long startTimeMs = System.currentTimeMillis();
@@ -1154,6 +1208,19 @@ public abstract class BaseBrokerStarter implements ServiceStartable {
@Override
public void stop() {
LOGGER.info("Shutting down Pinot broker");
+ if (_nativeSelected) {
+ synchronized (_nativeLifecycleLock) {
+ _isShuttingDown = true;
+ if (_nativeParticipant != null) {
+ _nativeParticipant.close();
+ }
+ if (_nativeStatusRegistered) {
+ ServiceStatus.removeServiceStatusCallback(_instanceId, _nativeParticipant);
+ _nativeStatusRegistered = false;
+ }
+ }
+ return;
+ }
_isShuttingDown = true;
stopWarmup();
stopPreConnect();
diff --git a/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/MultiClusterHelixBrokerStarter.java b/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/MultiClusterHelixBrokerStarter.java
index 36ce21bc594..1f7964b8139 100644
--- a/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/MultiClusterHelixBrokerStarter.java
+++ b/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/MultiClusterHelixBrokerStarter.java
@@ -31,6 +31,7 @@ import org.apache.pinot.broker.routing.manager.MultiClusterRoutingManager;
import org.apache.pinot.broker.routing.manager.RemoteClusterBrokerRoutingManager;
import org.apache.pinot.common.config.provider.TableCache;
import org.apache.pinot.common.config.provider.ZkTableCache;
+import org.apache.pinot.common.coordination.CoordinationBackend;
import org.apache.pinot.common.metrics.BrokerMeter;
import org.apache.pinot.core.routing.MultiClusterRoutingContext;
import org.apache.pinot.spi.env.PinotConfiguration;
@@ -72,6 +73,9 @@ public class MultiClusterHelixBrokerStarter extends BaseBrokerStarter {
@Override
public void init(PinotConfiguration brokerConf)
throws Exception {
+ if (CoordinationBackend.from(brokerConf) != CoordinationBackend.HELIX) {
+ throw new IllegalArgumentException("Multi-cluster broker requires the Helix coordination backend");
+ }
super.init(brokerConf);
_remoteInstanceId = _instanceId + "_remote";
initRemoteClusterNamesAndZk(brokerConf);
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
new file mode 100644
index 00000000000..ddc9a0b71cb
--- /dev/null
+++ b/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/NativeBrokerParticipant.java
@@ -0,0 +1,241 @@
+/**
+ * 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.broker.helix;
+
+import com.google.common.annotations.VisibleForTesting;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import java.io.IOException;
+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.common.coordination.MetadataRpcConnection;
+import org.apache.pinot.common.coordination.NativeMetadataConfig;
+import org.apache.pinot.common.coordination.NativeOfflineSegmentWorker;
+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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+
+
+/// Owns a broker membership session without advertising query endpoints, readiness, or serving capabilities.
+/// 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 {
+ private static final Logger LOGGER = LoggerFactory.getLogger(NativeBrokerParticipant.class);
+ private final NativeMetadataConfig _config;
+ private final MetadataRpcConnection _connection;
+ private final UUID _incarnation = UUID.randomUUID();
+ private final ScheduledExecutorService _heartbeats = Executors.newSingleThreadScheduledExecutor(
+ runnable -> new Thread(runnable, "native-broker-membership"));
+ @Nullable
+ private MetadataRpc.ParticipantToken _participant;
+ @Nullable
+ private MetadataRpc.MembershipCommandRequest _pendingRenewal;
+ private long _nextRenewal = 1;
+ private long _appliedIndex;
+ private boolean _started;
+ private boolean _fenced;
+ private volatile boolean _closed;
+ private volatile String _description = "Native broker membership starting; query listeners disabled";
+
+ NativeBrokerParticipant(NativeMetadataConfig config, MetadataRpcConnection connection) {
+ _config = config;
+ _connection = connection;
+ }
+
+ synchronized void start() throws IOException {
+ checkState(!_started && !_closed, "Native broker membership cannot be restarted");
+ _started = true;
+ try {
+ MetadataRpc.MembershipResponse current = read();
+ checkArgument(current.getStatus() == ResultStatus.OK || current.getStatus() == ResultStatus.NOT_FOUND,
+ "Cannot inspect native broker registration: %s", current.getStatus());
+ long previousEpoch = 0;
+ if (current.getStatus() == ResultStatus.OK) {
+ checkArgument(current.hasParticipant(), "Current broker registration is missing");
+ validateToken(current.getParticipant().getParticipant());
+ previousEpoch = current.getParticipant().getParticipant().getRegistrationEpoch();
+ }
+ checkArgument(previousEpoch < Long.MAX_VALUE, "Broker registration epochs exhausted");
+ MetadataRpc.MembershipCommandRequest registration = command().setRegister(
+ MetadataRpc.ParticipantRegistration.newBuilder().setIncarnation(_incarnation.toString())
+ .setExpectedEpoch(previousEpoch)).build();
+ MetadataRpc.MembershipResponse registered = _connection.membershipExecute(registration,
+ _config.rpcTimeoutMillis());
+ validateResponse(registered);
+ checkArgument(registered.getStatus() == ResultStatus.OK && registered.hasParticipant(),
+ "Cannot acquire native broker registration: %s", registered.getStatus());
+ MetadataRpc.ParticipantToken token = registered.getParticipant().getParticipant();
+ validateToken(token);
+ checkArgument(token.getIncarnation().equals(_incarnation.toString())
+ && 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);
+ _description = "Native broker membership active; routing unavailable and query listeners disabled";
+ _heartbeats.scheduleWithFixedDelay(this::heartbeat, _config.heartbeatMillis(), _config.heartbeatMillis(),
+ TimeUnit.MILLISECONDS);
+ } catch (IOException | RuntimeException e) {
+ close();
+ throw e;
+ }
+ }
+
+ @VisibleForTesting
+ synchronized void heartbeat() {
+ if (_closed || _fenced) {
+ return;
+ }
+ try {
+ if (_pendingRenewal == null) {
+ MetadataRpc.MembershipResponse image = read();
+ checkArgument(image.getStatus() == ResultStatus.OK, "Broker membership is unavailable: %s", image.getStatus());
+ validateOwnProjection(image);
+ _pendingRenewal = command().setRenew(MetadataRpc.ParticipantRenewal.newBuilder().setSession(session())
+ .setSequence(_nextRenewal).setAppliedRevision(image.getRevision())).build();
+ }
+ // 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);
+ if (response.getStatus() == ResultStatus.NOT_READY) {
+ _description = "Native broker control epoch unavailable; query listeners disabled";
+ return;
+ }
+ checkArgument(response.getStatus() == ResultStatus.OK && response.getRenewalsCount() == 1,
+ "Broker renewal rejected: %s", response.getStatus());
+ MetadataRpc.RenewalResult renewal = response.getRenewals(0);
+ 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");
+ checkArgument(_nextRenewal < Long.MAX_VALUE, "Broker renewal sequences exhausted");
+ _nextRenewal++;
+ _pendingRenewal = null;
+ _description = renewal.getStatus() == ResultStatus.OK
+ ? "Native broker membership active; routing unavailable and query listeners disabled"
+ : "Native broker membership projection must catch up; query listeners disabled";
+ } catch (StatusRuntimeException e) {
+ Status.Code code = e.getStatus().getCode();
+ if (code == Status.Code.PERMISSION_DENIED || code == Status.Code.UNAUTHENTICATED
+ || code == Status.Code.INVALID_ARGUMENT) {
+ fence("Broker membership authorization or request rejected: " + code);
+ } else {
+ _description = "Native broker membership transport unavailable; query listeners disabled";
+ }
+ } catch (IOException | IllegalArgumentException e) {
+ fence("Invalid broker membership response: " + e.getMessage());
+ } catch (RuntimeException e) {
+ fence("Broker membership failed: " + e.getClass().getSimpleName());
+ }
+ }
+
+ private MetadataRpc.MembershipResponse read() throws IOException {
+ MetadataRpc.MembershipResponse response = _connection.membershipRead(MetadataRpc.MembershipReadRequest.newBuilder()
+ .setHistoryId(_config.historyId().toString()).build(), _config.rpcTimeoutMillis());
+ validateResponse(response);
+ return response;
+ }
+
+ private void validateResponse(MetadataRpc.MembershipResponse response) throws IOException {
+ NativeOfflineSegmentWorker.validateMessage(response);
+ checkArgument(response.hasRevision() && response.getRevision().getHistoryId().equals(_config.historyId().toString())
+ && response.getRevision().getIndex() >= _appliedIndex, "Broker response history or revision differs");
+ }
+
+ private void validateToken(MetadataRpc.ParticipantToken token) {
+ checkArgument(token.getHistoryId().equals(_config.historyId().toString())
+ && token.getParticipantId().equals(_config.participantId())
+ && token.getRole() == MetadataRpc.ParticipantRole.BROKER && token.getRegistrationEpoch() > 0,
+ "Broker participant identity differs");
+ checkArgument(UUID.fromString(token.getIncarnation()).toString().equals(token.getIncarnation()),
+ "Invalid broker incarnation");
+ }
+
+ private void 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(),
+ "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.getClockBarrierIndex() <= response.getRevision().getIndex()
+ && state.getExpiryRound() > control.getRound() && state.getLastModifiedIndex() > 0
+ && state.getLastModifiedIndex() <= response.getRevision().getIndex(), "Broker membership lease is not live");
+ _appliedIndex = response.getRevision().getIndex();
+ }
+
+ private MetadataRpc.ParticipantSession session() {
+ checkState(_participant != null, "Broker registration is absent");
+ return MetadataRpc.ParticipantSession.newBuilder().setIncarnation(_participant.getIncarnation())
+ .setRegistrationEpoch(_participant.getRegistrationEpoch()).build();
+ }
+
+ private MetadataRpc.MembershipCommandRequest.Builder command() {
+ return MetadataRpc.MembershipCommandRequest.newBuilder().setHistoryId(_config.historyId().toString());
+ }
+
+ private void fence(String detail) {
+ _fenced = true;
+ _description = detail + "; query listeners disabled";
+ _heartbeats.shutdown();
+ LOGGER.warn("{}", _description);
+ }
+
+ @Override
+ public ServiceStatus.Status getServiceStatus() {
+ return _closed ? ServiceStatus.Status.SHUTTING_DOWN : ServiceStatus.Status.BAD;
+ }
+
+ @Override
+ public String getStatusDescription() {
+ return _description;
+ }
+
+ @Override
+ public synchronized void close() {
+ if (_closed) {
+ return;
+ }
+ _closed = true;
+ _description = "Native broker stopped; query listeners disabled";
+ _heartbeats.shutdownNow();
+ 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();
+ }
+ }
+}
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
new file mode 100644
index 00000000000..74e67f5d4db
--- /dev/null
+++ b/pinot-broker/src/test/java/org/apache/pinot/broker/broker/helix/NativeBrokerParticipantTest.java
@@ -0,0 +1,290 @@
+/**
+ * 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.broker.helix;
+
+import io.grpc.Status;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Deque;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import org.apache.pinot.common.coordination.MetadataRpcConnection;
+import org.apache.pinot.common.coordination.NativeMetadataConfig;
+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.env.PinotConfiguration;
+import org.testng.annotations.Test;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNull;
+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.
+public class NativeBrokerParticipantTest {
+ @Test(timeOut = 10_000)
+ public void testNormalStarterBypassesZookeeperAndLeavesQueriesUnavailable() throws Exception {
+ try (Fixture fixture = new Fixture()) {
+ HelixBrokerStarter starter = new HelixBrokerStarter() {
+ @Override
+ NativeBrokerParticipant createNativeParticipant(NativeMetadataConfig config) {
+ return fixture._participant;
+ }
+ };
+ // No ZK configuration and no query ports are supplied. The mock avoids opening TLS connections in this unit test.
+ starter.init(fixture._configuration);
+ assertEquals(starter.getInstanceId(), fixture._id);
+ assertNull(starter._listenerConfigs);
+ starter.start();
+ try {
+ assertNull(starter._spectatorHelixManager);
+ assertNull(starter._participantHelixManager);
+ assertNull(starter._routingManager);
+ assertNull(starter._brokerRequestHandler);
+ assertNull(starter._brokerAdminApplication);
+ assertNull(starter._brokerGrpcServer);
+ assertEquals(ServiceStatus.getServiceStatus(fixture._id), ServiceStatus.Status.BAD);
+ assertTrue(ServiceStatus.getStatusDescription(fixture._id).contains("query listeners disabled"));
+ assertTrue(fixture._requests.stream().noneMatch(MetadataRpc.MembershipCommandRequest::hasReady));
+ assertEquals(fixture._requests.getFirst().getRegister().getEndpointsCount(), 0);
+ assertThrows(IllegalArgumentException.class,
+ () -> new MultiClusterHelixBrokerStarter().init(fixture._configuration));
+ HelixBrokerStarter initializedOnly = new HelixBrokerStarter();
+ initializedOnly.init(fixture._configuration);
+ initializedOnly.stop();
+ assertEquals(ServiceStatus.getServiceStatus(fixture._id), ServiceStatus.Status.BAD);
+ NativeBrokerParticipant rejected = mock(NativeBrokerParticipant.class);
+ doThrow(new IOException("Registration rejected")).when(rejected).start();
+ HelixBrokerStarter duplicate = new HelixBrokerStarter() {
+ @Override
+ NativeBrokerParticipant createNativeParticipant(NativeMetadataConfig config) {
+ return rejected;
+ }
+ };
+ duplicate.init(fixture._configuration);
+ assertThrows(IOException.class, duplicate::start);
+ duplicate.stop();
+ assertTrue(ServiceStatus.getStatusDescription(fixture._id).contains("query listeners disabled"));
+ } finally {
+ starter.stop();
+ }
+ assertEquals(ServiceStatus.getServiceStatus(fixture._id), ServiceStatus.Status.NOT_STARTED);
+ fixture.assertOwnUnregister();
+ }
+ try (Fixture fixture = new Fixture()) {
+ HelixBrokerStarter starter = new HelixBrokerStarter() {
+ @Override
+ NativeBrokerParticipant createNativeParticipant(NativeMetadataConfig config) {
+ return fixture._participant;
+ }
+ };
+ starter.init(fixture._configuration);
+ starter.start();
+ ServiceStatus.setServiceStatusCallback(fixture._id,
+ new ServiceStatus.LifecycleServiceStatusCallback(() -> false, () -> false));
+ try {
+ starter.stop();
+ assertEquals(ServiceStatus.getServiceStatus(fixture._id), ServiceStatus.Status.GOOD,
+ "An old process must not remove a replacement's status callback");
+ } finally {
+ starter.stop();
+ ServiceStatus.removeServiceStatusCallback(fixture._id);
+ }
+ }
+ HelixBrokerStarter invalid = new HelixBrokerStarter();
+ assertThrows(IllegalArgumentException.class,
+ () -> invalid.init(new PinotConfiguration(Map.of("pinot.coordination.backend", "RATIS"))));
+ assertThrows(IllegalStateException.class, invalid::start);
+ invalid.stop();
+ assertNull(invalid._participantHelixManager);
+ }
+
+ @Test(timeOut = 10_000)
+ public void testUnknownRenewalRetriesAndConsumedNotReadySequence() throws Exception {
+ try (Fixture fixture = new Fixture()) {
+ fixture._participant.start();
+ fixture._renewals.add(Status.DEADLINE_EXCEEDED.asRuntimeException());
+ fixture._renewals.add("outer-not-ready");
+ fixture._renewals.add(ResultStatus.NOT_READY);
+ fixture._renewals.add(ResultStatus.OK);
+ fixture._renewals.add(ResultStatus.FENCED);
+ for (int i = 0; i < 5; i++) {
+ fixture._participant.heartbeat();
+ }
+ List<MetadataRpc.MembershipCommandRequest> renewals = fixture._requests.stream()
+ .filter(MetadataRpc.MembershipCommandRequest::hasRenew).toList();
+ assertEquals(renewals.size(), 5);
+ assertEquals(renewals.get(0), renewals.get(1), "An outer rejection must preserve the exact pending request");
+ assertEquals(renewals.get(1), renewals.get(2), "A lost reply must not change the applied revision or sequence");
+ assertEquals(renewals.get(2).getRenew().getSequence(), 1L);
+ assertEquals(renewals.get(3).getRenew().getSequence(), 2L, "A durable inner NOT_READY consumes its sequence");
+ assertEquals(renewals.get(4).getRenew().getSequence(), 3L);
+ assertTrue(renewals.get(3).getRenew().getAppliedRevision().getIndex()
+ > renewals.getFirst().getRenew().getAppliedRevision().getIndex());
+ int requests = fixture._requests.size();
+ fixture._participant.heartbeat();
+ assertEquals(fixture._requests.size(), requests, "Fenced sessions must stop renewing");
+ assertEquals(fixture._participant.getServiceStatus(), ServiceStatus.Status.BAD);
+ }
+ }
+
+ @Test(timeOut = 10_000)
+ public void testUnexpectedAuthorityAndReplacementFenceTheConfirmedSession() throws Exception {
+ try (Fixture fixture = new Fixture()) {
+ fixture._grantCapabilities = true;
+ assertThrows(IllegalArgumentException.class, fixture._participant::start);
+ fixture.assertOwnUnregister();
+ verify(fixture._connection).close();
+ }
+ try (Fixture fixture = new Fixture()) {
+ fixture._participant.start();
+ MetadataRpc.ParticipantToken original = fixture._token;
+ fixture._token = original.toBuilder().setIncarnation(UUID.randomUUID().toString())
+ .setRegistrationEpoch(original.getRegistrationEpoch() + 1).build();
+ fixture._participant.heartbeat();
+ assertFalse(fixture._requests.stream().anyMatch(MetadataRpc.MembershipCommandRequest::hasRenew));
+ fixture._participant.close();
+ MetadataRpc.MembershipCommandRequest last = fixture._requests.getLast();
+ assertTrue(last.hasUnregister());
+ assertEquals(last.getUnregister().getIncarnation(), original.getIncarnation());
+ assertEquals(last.getUnregister().getRegistrationEpoch(), original.getRegistrationEpoch());
+ }
+ try (Fixture fixture = new Fixture()) {
+ fixture._participant.start();
+ fixture._renewals.add("unexpected-permit");
+ fixture._participant.heartbeat();
+ int requests = fixture._requests.size();
+ fixture._participant.heartbeat();
+ assertEquals(fixture._requests.size(), requests);
+ assertTrue(fixture._participant.getStatusDescription().contains("unexpected authority"));
+ }
+ }
+
+ 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 PinotConfiguration _configuration;
+ 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 MetadataRpc.ParticipantToken _token;
+ private long _index;
+ private boolean _grantCapabilities;
+
+ private Fixture() throws Exception {
+ _configuration = new PinotConfiguration(Map.of("pinot.coordination.backend", "RATIS",
+ NativeMetadataConfig.PREFIX + "historyId", _history,
+ NativeMetadataConfig.PREFIX + "participantId", _id,
+ NativeMetadataConfig.PREFIX + "metadata.endpoints", "localhost:10000",
+ NativeMetadataConfig.PREFIX + "tls.certChain", _tls.toString(),
+ NativeMetadataConfig.PREFIX + "tls.privateKey", _tls.toString(),
+ NativeMetadataConfig.PREFIX + "tls.trustCerts", _tls.toString(),
+ NativeMetadataConfig.PREFIX + "heartbeatMillis", 30_000));
+ _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(1000))
+ .build();
+ when(_connection.membershipRead(any(), anyLong())).thenAnswer(invocation -> {
+ MetadataRpc.MembershipResponse.Builder response =
+ response(_token == null ? ResultStatus.NOT_FOUND : ResultStatus.OK);
+ if (_token != null) {
+ response.setParticipant(state());
+ }
+ return response.build();
+ });
+ when(_connection.membershipExecute(any(), anyLong())).thenAnswer(invocation -> {
+ MetadataRpc.MembershipCommandRequest request = invocation.getArgument(0);
+ _requests.add(request);
+ if (request.hasRegister()) {
+ _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();
+ }
+ if (request.hasRenew()) {
+ Object next = _renewals.isEmpty() ? ResultStatus.OK : _renewals.removeFirst();
+ if (next instanceof RuntimeException failure) {
+ throw failure;
+ }
+ if (next.equals("outer-not-ready")) {
+ return response(ResultStatus.NOT_READY).build();
+ }
+ MetadataRpc.RenewalResult.Builder renewal = MetadataRpc.RenewalResult.newBuilder()
+ .setParticipant(_token).setSequence(request.getRenew().getSequence());
+ if (next.equals("unexpected-permit")) {
+ renewal.setStatus(ResultStatus.OK).setPermit(MetadataRpc.ServingPermit.getDefaultInstance());
+ } else {
+ renewal.setStatus((ResultStatus) next);
+ }
+ return response(ResultStatus.OK).addRenewals(renewal).build();
+ }
+ 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);
+ }
+
+ private MetadataRpc.MembershipResponse.Builder response(ResultStatus status) {
+ return MetadataRpc.MembershipResponse.newBuilder().setStatus(status).setControl(_control)
+ .setRevision(MetadataRpc.Revision.newBuilder().setHistoryId(_history).setIndex(++_index));
+ }
+
+ 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));
+ if (_grantCapabilities) {
+ state.addCapabilities(MetadataRpc.ServingCapability.BROKER_QUERY);
+ }
+ return state.build();
+ }
+
+ private void assertOwnUnregister() {
+ MetadataRpc.MembershipCommandRequest last = _requests.getLast();
+ assertTrue(last.hasUnregister());
+ assertEquals(last.getUnregister().getIncarnation(), _token.getIncarnation());
+ assertEquals(last.getUnregister().getRegistrationEpoch(), _token.getRegistrationEpoch());
+ }
+
+ @Override
+ public void close() throws Exception {
+ _participant.close();
+ Files.deleteIfExists(_tls);
+ }
+ }
+}
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/coordination/CoordinationBackend.java b/pinot-common/src/main/java/org/apache/pinot/common/coordination/CoordinationBackend.java
new file mode 100644
index 00000000000..e3d085c47f0
--- /dev/null
+++ b/pinot-common/src/main/java/org/apache/pinot/common/coordination/CoordinationBackend.java
@@ -0,0 +1,39 @@
+/**
+ * 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.common.coordination;
+
+import java.util.Locale;
+import org.apache.pinot.spi.env.PinotConfiguration;
+
+
+/// Bootstrap-only coordination selection. Absence preserves Helix; misspellings never silently choose a backend.
+public enum CoordinationBackend {
+ HELIX, RATIS;
+
+ public static final String CONFIG_KEY = "pinot.coordination.backend";
+
+ public static CoordinationBackend from(PinotConfiguration config) {
+ String value = config.getProperty(CONFIG_KEY, HELIX.name());
+ try {
+ return valueOf(value.trim().toUpperCase(Locale.ROOT));
+ } catch (IllegalArgumentException | NullPointerException e) {
+ throw new IllegalArgumentException(CONFIG_KEY + " must be HELIX or RATIS", e);
+ }
+ }
+}
diff --git a/pinot-server/src/main/java/org/apache/pinot/server/starter/NativeMetadataConnection.java b/pinot-common/src/main/java/org/apache/pinot/common/coordination/MetadataRpcConnection.java
similarity index 89%
copy from pinot-server/src/main/java/org/apache/pinot/server/starter/NativeMetadataConnection.java
copy to pinot-common/src/main/java/org/apache/pinot/common/coordination/MetadataRpcConnection.java
index 7c027d67013..52527d847b1 100644
--- a/pinot-server/src/main/java/org/apache/pinot/server/starter/NativeMetadataConnection.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/coordination/MetadataRpcConnection.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.pinot.server.starter;
+package org.apache.pinot.common.coordination;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
@@ -27,7 +27,6 @@ import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiFunction;
-import org.apache.pinot.common.coordination.MetadataRpcClient;
import org.apache.pinot.coordination.ratis.rpc.MetadataRpc;
import static com.google.common.base.Preconditions.checkArgument;
@@ -35,17 +34,31 @@ import static com.google.common.base.Preconditions.checkArgument;
/// Bounded endpoint failover that resends the identical application request after an unknown transport outcome.
/// Instances own their channels; membership uses a separate instance from metadata hydration and installation.
-public final class NativeMetadataConnection implements NativeServerParticipant.Rpc {
+public class MetadataRpcConnection implements AutoCloseable {
private final List<MetadataRpcClient> _clients;
private final AtomicInteger _preferred = new AtomicInteger();
- public record Endpoint(String host, int port) {
- public Endpoint {
+ /// A TLS endpoint whose hostname is verified against the peer certificate.
+ public static final class Endpoint {
+ private final String _host;
+ private final int _port;
+
+ public Endpoint(String host, int port) {
checkArgument(host != null && !host.isBlank() && port > 0 && port <= 65535, "Invalid metadata endpoint");
+ _host = host;
+ _port = port;
+ }
+
+ public String host() {
+ return _host;
+ }
+
+ public int port() {
+ return _port;
}
}
- public NativeMetadataConnection(List<Endpoint> endpoints, Path certificate, Path key, Path trust,
+ public MetadataRpcConnection(List<Endpoint> endpoints, Path certificate, Path key, Path trust,
long timeoutMillis) throws IOException {
checkArgument(endpoints != null && !endpoints.isEmpty() && endpoints.size() <= 16,
"Between one and sixteen metadata endpoints are required");
@@ -79,12 +92,10 @@ public final class NativeMetadataConnection implements NativeServerParticipant.R
return call(timeoutMillis, (client, timeout) -> client.installationLookup(request, timeout));
}
- @Override
public MetadataRpc.CommandResult execute(MetadataRpc.InstallationCommandRequest request, long timeoutMillis) {
return call(timeoutMillis, (client, timeout) -> client.installationExecute(request, timeout));
}
- @Override
public MetadataRpc.CommandResult outcome(MetadataRpc.InstallationOutcomeRequest request, long timeoutMillis) {
return call(timeoutMillis, (client, timeout) -> client.installationOutcome(request, timeout));
}
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/coordination/NativeMetadataConfig.java b/pinot-common/src/main/java/org/apache/pinot/common/coordination/NativeMetadataConfig.java
new file mode 100644
index 00000000000..c1ddc7232b9
--- /dev/null
+++ b/pinot-common/src/main/java/org/apache/pinot/common/coordination/NativeMetadataConfig.java
@@ -0,0 +1,119 @@
+/**
+ * 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.common.coordination;
+
+import com.google.common.net.HostAndPort;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import org.apache.pinot.common.coordination.MetadataRpcConnection.Endpoint;
+import org.apache.pinot.spi.env.PinotConfiguration;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+
+/// Immutable bootstrap identity, TLS and bounded connection settings shared by native participant starters.
+/// Scheduling intervals are transport controls, never qualifications of authority clocks or serving lifetimes.
+public final class NativeMetadataConfig {
+ public static final String PREFIX = "pinot.coordination.ratis.";
+ private final UUID _historyId;
+ private final String _participantId;
+ private final List<Endpoint> _endpoints;
+ private final Path _certificate;
+ private final Path _privateKey;
+ private final Path _trustCertificates;
+ private final long _rpcTimeoutMillis;
+ private final long _heartbeatMillis;
+
+ public NativeMetadataConfig(PinotConfiguration config) {
+ _historyId = UUID.fromString(required(config, "historyId"));
+ _participantId = required(config, "participantId");
+ checkArgument(_participantId.length() <= 128, "Participant ID must contain at most 128 characters");
+ List<Endpoint> endpoints = new ArrayList<>();
+ Set<String> unique = new HashSet<>();
+ for (String value : required(config, "metadata.endpoints").split(",", -1)) {
+ HostAndPort address = HostAndPort.fromString(value.trim());
+ checkArgument(address.hasPort() && unique.add(address.toString()), "Missing port or duplicate metadata endpoint");
+ endpoints.add(new Endpoint(address.getHost(), address.getPort()));
+ }
+ checkArgument(!endpoints.isEmpty() && endpoints.size() <= 16, "Between one and sixteen endpoints are required");
+ _endpoints = List.copyOf(endpoints);
+ _certificate = readableFile(config, "tls.certChain");
+ _privateKey = readableFile(config, "tls.privateKey");
+ _trustCertificates = readableFile(config, "tls.trustCerts");
+ _rpcTimeoutMillis = config.getProperty(PREFIX + "rpc.timeoutMs", 5000L);
+ _heartbeatMillis = config.getProperty(PREFIX + "heartbeatMillis", 1000L);
+ checkArgument(_rpcTimeoutMillis > 0 && _rpcTimeoutMillis <= 30_000, "RPC timeout must be between 1 and 30000 ms");
+ checkArgument(_heartbeatMillis > 0 && _heartbeatMillis <= 30_000,
+ "Heartbeat interval must be between 1 and 30000 ms");
+ }
+
+ public static String required(PinotConfiguration config, String suffix) {
+ String value = config.getProperty(PREFIX + suffix);
+ checkArgument(value != null && !value.isBlank(), "Missing required configuration: %s%s", PREFIX, suffix);
+ return value.trim();
+ }
+
+ public static Path readableFile(PinotConfiguration config, String suffix) {
+ Path path = Path.of(required(config, suffix)).toAbsolutePath().normalize();
+ checkArgument(Files.isRegularFile(path) && Files.isReadable(path), "Unreadable file for %s%s", PREFIX, suffix);
+ return path;
+ }
+
+ public UUID historyId() {
+ return _historyId;
+ }
+
+ public String participantId() {
+ return _participantId;
+ }
+
+ public List<Endpoint> endpoints() {
+ return _endpoints;
+ }
+
+ public Path certificate() {
+ return _certificate;
+ }
+
+ public Path privateKey() {
+ return _privateKey;
+ }
+
+ public Path trustCertificates() {
+ return _trustCertificates;
+ }
+
+ public long rpcTimeoutMillis() {
+ return _rpcTimeoutMillis;
+ }
+
+ public long heartbeatMillis() {
+ return _heartbeatMillis;
+ }
+
+ public MetadataRpcConnection openConnection() throws IOException {
+ return new MetadataRpcConnection(_endpoints, _certificate, _privateKey, _trustCertificates, _rpcTimeoutMillis);
+ }
+}
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/utils/ServiceStatus.java b/pinot-common/src/main/java/org/apache/pinot/common/utils/ServiceStatus.java
index 947e8a9c02e..d5090622904 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/utils/ServiceStatus.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/utils/ServiceStatus.java
@@ -65,6 +65,11 @@ public class ServiceStatus {
ServiceStatus.SERVICE_STATUS_CALLBACK_MAP.remove(name);
}
+ /// Removes only the caller's registration, preserving a replacement process's callback for the same stable ID.
+ public static void removeServiceStatusCallback(String name, ServiceStatusCallback callback) {
+ SERVICE_STATUS_CALLBACK_MAP.remove(name, callback);
+ }
+
public static Status getServiceStatus() {
return getServiceStatus(SERVICE_STATUS_CALLBACK);
}
diff --git a/pinot-common/src/test/java/org/apache/pinot/common/coordination/NativeMetadataConfigTest.java b/pinot-common/src/test/java/org/apache/pinot/common/coordination/NativeMetadataConfigTest.java
new file mode 100644
index 00000000000..12b1a303f72
--- /dev/null
+++ b/pinot-common/src/test/java/org/apache/pinot/common/coordination/NativeMetadataConfigTest.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.common.coordination;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Map;
+import java.util.UUID;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertThrows;
+
+
+/// Bootstrap validation must reject incomplete native settings rather than attempt legacy coordination.
+public class NativeMetadataConfigTest {
+ @Test
+ public void testBackendSelectionAndNativeConnectionValidation() throws Exception {
+ PinotConfiguration config = new PinotConfiguration();
+ assertEquals(CoordinationBackend.from(config), CoordinationBackend.HELIX);
+ config.setProperty(CoordinationBackend.CONFIG_KEY, " RATIS ");
+ assertEquals(CoordinationBackend.from(config), CoordinationBackend.RATIS);
+ assertThrows(IllegalArgumentException.class, () -> new NativeMetadataConfig(config));
+ config.setProperty(CoordinationBackend.CONFIG_KEY, "RATSI");
+ assertThrows(IllegalArgumentException.class, () -> CoordinationBackend.from(config));
+
+ Path pem = Files.createTempFile("native-config", ".pem");
+ try {
+ // Config parsing checks readable paths; the connection validates PEM contents before opening channels.
+ PinotConfiguration nativeConfig = new PinotConfiguration(Map.of(
+ NativeMetadataConfig.PREFIX + "historyId", UUID.randomUUID().toString(),
+ NativeMetadataConfig.PREFIX + "participantId", "Broker_test",
+ NativeMetadataConfig.PREFIX + "metadata.endpoints", "localhost:8000,[::1]:8001",
+ NativeMetadataConfig.PREFIX + "tls.certChain", pem.toString(),
+ NativeMetadataConfig.PREFIX + "tls.privateKey", pem.toString(),
+ NativeMetadataConfig.PREFIX + "tls.trustCerts", pem.toString()));
+ NativeMetadataConfig parsed = new NativeMetadataConfig(nativeConfig);
+ assertEquals(parsed.endpoints().size(), 2);
+ assertEquals(parsed.endpoints().get(1).host(), "::1");
+ assertEquals(parsed.endpoints().get(1).port(), 8001);
+ assertThrows(UnsupportedOperationException.class, () -> parsed.endpoints().clear());
+ PinotConfiguration invalid = nativeConfig;
+ invalid.setProperty(NativeMetadataConfig.PREFIX + "metadata.endpoints", "localhost:8000,localhost:8000");
+ assertThrows(IllegalArgumentException.class, () -> new NativeMetadataConfig(invalid));
+ invalid.setProperty(NativeMetadataConfig.PREFIX + "metadata.endpoints", "localhost:8000");
+ invalid.setProperty(NativeMetadataConfig.PREFIX + "rpc.timeoutMs", 30001);
+ assertThrows(IllegalArgumentException.class, () -> new NativeMetadataConfig(invalid));
+ } finally {
+ Files.deleteIfExists(pem);
+ }
+ }
+}
diff --git a/pinot-controller/pom.xml b/pinot-controller/pom.xml
index 8e2d7de1755..99c9aa4dfed 100644
--- a/pinot-controller/pom.xml
+++ b/pinot-controller/pom.xml
@@ -36,6 +36,10 @@
<npmRepositoryUrl>https://registry.npmjs.org/npm/-/</npmRepositoryUrl>
</properties>
<dependencies>
+ <dependency>
+ <groupId>org.apache.pinot</groupId>
+ <artifactId>pinot-coordination-ratis</artifactId>
+ </dependency>
<dependency>
<groupId>org.apache.pinot</groupId>
<artifactId>pinot-materialized-view</artifactId>
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/BaseControllerStarter.java b/pinot-controller/src/main/java/org/apache/pinot/controller/BaseControllerStarter.java
index c8a92a9c1e3..0a773058137 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/BaseControllerStarter.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/BaseControllerStarter.java
@@ -37,6 +37,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
+import javax.annotation.Nullable;
import javax.inject.Singleton;
import javax.net.ssl.SSLContext;
import nl.altindag.ssl.SSLFactory;
@@ -68,6 +69,7 @@ import org.apache.pinot.common.Utils;
import org.apache.pinot.common.audit.AuditServiceBinder;
import org.apache.pinot.common.config.DefaultClusterConfigChangeHandler;
import org.apache.pinot.common.config.TlsConfig;
+import org.apache.pinot.common.coordination.CoordinationBackend;
import org.apache.pinot.common.evaluator.GroovyFunctionEvaluator;
import org.apache.pinot.common.function.FunctionRegistry;
import org.apache.pinot.common.http.PoolingHttpClientConnectionManagerHelper;
@@ -101,6 +103,7 @@ import org.apache.pinot.controller.api.access.AccessControlFactory;
import org.apache.pinot.controller.api.events.MetadataEventNotifierFactory;
import org.apache.pinot.controller.api.resources.ControllerFilePathProvider;
import org.apache.pinot.controller.api.resources.InvalidControllerConfigException;
+import org.apache.pinot.controller.coordination.RatisControllerRuntime;
import org.apache.pinot.controller.helix.RealtimeConsumerMonitor;
import org.apache.pinot.controller.helix.SegmentStatusChecker;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
@@ -242,11 +245,23 @@ public abstract class BaseControllerStarter implements ServiceStartable {
protected TableRebalanceManager _tableRebalanceManager;
protected DefaultClusterConfigChangeHandler _clusterConfigChangeHandler;
protected MaterializedViewConsistencyManager _materializedViewConsistencyManager;
+ @Nullable
+ private RatisControllerRuntime _ratisRuntime;
+ private boolean _ratisSelected;
+ private boolean _ratisStatusRegistered;
+ private final Object _ratisLifecycleLock = new Object();
@Override
public void init(PinotConfiguration pinotConfiguration)
throws Exception {
_config = new ControllerConf(pinotConfiguration.toMap());
+ _ratisSelected = CoordinationBackend.from(_config) == CoordinationBackend.RATIS;
+ if (_ratisSelected) {
+ applyCustomConfigs(_config);
+ _ratisRuntime = new RatisControllerRuntime(_config);
+ _helixParticipantInstanceId = _ratisRuntime.instanceId();
+ return;
+ }
_helixZkURL = HelixConfig.getAbsoluteZkPathForHelix(_config.getZkStr());
_helixClusterName = _config.getHelixClusterName();
_controllerMode = _config.getControllerMode();
@@ -481,6 +496,15 @@ public abstract class BaseControllerStarter implements ServiceStartable {
@Override
public void start()
throws Exception {
+ if (_ratisSelected) {
+ synchronized (_ratisLifecycleLock) {
+ Preconditions.checkState(_ratisRuntime != null, "Native controller initialization did not complete");
+ _ratisRuntime.start();
+ ServiceStatus.setServiceStatusCallback(_helixParticipantInstanceId, _ratisRuntime);
+ _ratisStatusRegistered = true;
+ }
+ return;
+ }
LOGGER.info("Starting Pinot controller in mode: {}. (Version: {})", _controllerMode.name(), PinotVersion.VERSION);
LOGGER.info("Controller configs: {}", new PinotAppConfigs(getConfig()).toJSONString());
long startTimeMs = System.currentTimeMillis();
@@ -1186,6 +1210,22 @@ public abstract class BaseControllerStarter implements ServiceStartable {
@Override
public void stop() {
+ if (_ratisSelected) {
+ synchronized (_ratisLifecycleLock) {
+ if (_ratisStatusRegistered) {
+ ServiceStatus.removeServiceStatusCallback(_helixParticipantInstanceId, _ratisRuntime);
+ _ratisStatusRegistered = false;
+ }
+ try {
+ if (_ratisRuntime != null) {
+ _ratisRuntime.close();
+ }
+ } catch (Exception e) {
+ LOGGER.error("Failed to close native controller", e);
+ }
+ }
+ return;
+ }
switch (_controllerMode) {
case DUAL:
stopPinotController();
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/RatisControllerProvider.java b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/RatisControllerProvider.java
new file mode 100644
index 00000000000..cedef5ff253
--- /dev/null
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/RatisControllerProvider.java
@@ -0,0 +1,52 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.controller.coordination;
+
+import org.apache.pinot.coordination.ratis.RatisMetadataStore.MembershipClockSettings;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Policy;
+import org.apache.pinot.coordination.ratis.transport.ControllerAccessPolicy;
+import org.apache.pinot.coordination.ratis.transport.MetadataAudit;
+import org.apache.pinot.coordination.ratis.transport.ParticipantAccessPolicy;
+import org.apache.pinot.spi.env.PinotConfiguration;
+
+
+/// Supplies explicitly qualified clock and certificate authorization inputs to the native controller lifecycle.
+/// There is no default implementation: selecting a class or configuring numeric bounds does not qualify a clock.
+/// Authorization policies must be current, nonblocking and thread-safe. This starter denies participant bindings
+/// with serving capabilities; query listeners and controller HTTP resources are outside this lifecycle.
+public interface RatisControllerProvider extends AutoCloseable {
+ void init(PinotConfiguration configuration) throws Exception;
+
+ MembershipClockSettings membershipClockSettings();
+
+ ControllerAccessPolicy controllerAccessPolicy();
+
+ ParticipantAccessPolicy participantAccessPolicy();
+
+ /// Version-one policy for a previously unconfigured history. Existing committed policies are never replaced.
+ Policy bootstrapPolicy();
+
+ default MetadataAudit audit() {
+ return (client, operation, outcome) -> { };
+ }
+
+ @Override
+ default void close() throws Exception {
+ }
+}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/RatisControllerRuntime.java b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/RatisControllerRuntime.java
new file mode 100644
index 00000000000..ba7cc2c69ef
--- /dev/null
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/RatisControllerRuntime.java
@@ -0,0 +1,251 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.controller.coordination;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.file.Path;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.coordination.NativeMetadataConfig;
+import org.apache.pinot.common.utils.ServiceStatus;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.coordination.ratis.RatisMetadataAccess;
+import org.apache.pinot.coordination.ratis.RatisMetadataStore;
+import org.apache.pinot.coordination.ratis.RatisMetadataStore.MembershipClockSettings;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ConfigurePolicy;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.GetControl;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Policy;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Response;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.MembershipCommand;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.MembershipQuery;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Status;
+import org.apache.pinot.coordination.ratis.transport.ControllerAccessPolicy;
+import org.apache.pinot.coordination.ratis.transport.MetadataAudit;
+import org.apache.pinot.coordination.ratis.transport.MetadataGrpcServer;
+import org.apache.pinot.coordination.ratis.transport.MetadataGrpcService.Limits;
+import org.apache.pinot.coordination.ratis.transport.ParticipantAccessPolicy;
+import org.apache.pinot.coordination.ratis.transport.ParticipantAccessPolicy.ParticipantIdentity;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.plugin.PluginManager;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+import static java.util.Objects.requireNonNull;
+
+
+/// Owns one embedded voter, its qualified membership clock and authenticated metadata endpoint. Lifecycle methods
+/// are serialized; health checks use a bounded consensus read and do not claim HTTP, scheduling or query readiness.
+/// The current voter transport remains restricted to three loopback peers. No Helix resources are constructed.
+public final class RatisControllerRuntime implements AutoCloseable, ServiceStatus.ServiceStatusCallback {
+ private final PinotConfiguration _config;
+ private final UUID _history;
+ private final String _instanceId;
+ private final String _providerClass;
+ private final InetSocketAddress _address;
+ private final Path _certificate;
+ private final Path _privateKey;
+ private final Path _trust;
+ private final long _startupMillis;
+ @Nullable
+ private RatisControllerProvider _provider;
+ @Nullable
+ private MetadataGrpcServer _endpoint;
+ @Nullable
+ private volatile RatisMetadataStore _store;
+ @Nullable
+ private volatile RatisMetadataAccess _access;
+ private volatile boolean _started;
+ private volatile boolean _closed;
+ private volatile String _status = "Native controller has not started";
+
+ public RatisControllerRuntime(ControllerConf config) {
+ _config = new PinotConfiguration(config.toMap());
+ _history = UUID.fromString(NativeMetadataConfig.required(config, "historyId"));
+ String voter = NativeMetadataConfig.required(config, "id");
+ _instanceId = config.getInstanceId() == null ? "Controller_ratis_" + voter : config.getInstanceId();
+ checkArgument(!_instanceId.isBlank(), "Native controller instance ID must not be blank");
+ _providerClass = NativeMetadataConfig.required(config, "controller.provider.class");
+ String host = NativeMetadataConfig.required(config, "metadata.host");
+ int port = Integer.parseInt(NativeMetadataConfig.required(config, "metadata.port"));
+ checkArgument(port > 0 && port <= 65535, "Metadata port must be between 1 and 65535");
+ _address = new InetSocketAddress(host, port);
+ _certificate = NativeMetadataConfig.readableFile(config, "tls.certChain");
+ _privateKey = NativeMetadataConfig.readableFile(config, "tls.privateKey");
+ _trust = NativeMetadataConfig.readableFile(config, "tls.trustCerts");
+ _startupMillis = config.getProperty(NativeMetadataConfig.PREFIX + "controller.startup.timeoutMs", 30_000L);
+ checkArgument(_startupMillis > 0 && _startupMillis <= 300_000,
+ "Native controller startup timeout must be between 1 and 300000 milliseconds");
+ }
+
+ public String instanceId() {
+ return _instanceId;
+ }
+
+ public synchronized void start() throws Exception {
+ checkState(!_started && !_closed && _provider == null, "Native controller is already started or closed");
+ try {
+ _provider = PluginManager.get().createInstance(_providerClass);
+ _provider.init(_config);
+ MembershipClockSettings clock = requireNonNull(_provider.membershipClockSettings(), "membershipClockSettings");
+ ControllerAccessPolicy controllers = requireNonNull(_provider.controllerAccessPolicy(), "controllerAccessPolicy");
+ ParticipantAccessPolicy participants = requireNonNull(_provider.participantAccessPolicy(),
+ "participantAccessPolicy");
+ MetadataAudit audit = requireNonNull(_provider.audit(), "audit");
+ Policy policy = requireNonNull(_provider.bootstrapPolicy(), "bootstrapPolicy");
+ validateBootstrapPolicy(policy);
+ _store = new RatisMetadataStore(_config);
+ _access = _store.openMetadataAccess();
+ _store.startMembershipClock(clock);
+ initializePolicy(policy);
+ _endpoint = new MetadataGrpcServer(_address, _certificate, _privateKey, _trust, _access, controllers, _access,
+ fingerprint -> {
+ ParticipantIdentity identity = participants.authorize(fingerprint);
+ return identity != null && identity.capabilities().isEmpty() ? identity : null;
+ }, audit, Limits.defaults());
+ checkClock();
+ _started = true;
+ _status = "Native metadata endpoint and membership initialized; controller HTTP and query serving are disabled";
+ } catch (Exception | Error e) {
+ _status = "Native controller startup failed: " + e.getMessage();
+ try {
+ close();
+ } catch (Exception cleanup) {
+ e.addSuppressed(cleanup);
+ }
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ throw e;
+ }
+ }
+
+ private static void validateBootstrapPolicy(Policy policy) {
+ long maxMillis = Long.MAX_VALUE / 1_000_000;
+ checkArgument(policy.version() == 1 && policy.minTickIntervalMillis() > 0
+ && policy.minTickIntervalMillis() <= maxMillis && policy.leaseRounds() >= 2
+ && policy.permitLifetimeMillis() > 0 && policy.permitLifetimeMillis() <= maxMillis
+ && policy.maxQueryLifetimeMillis() > 0 && policy.maxQueryLifetimeMillis() <= maxMillis
+ && policy.maxRevisionLag() > 0
+ && (policy.permitLifetimeMillis() - 1) / policy.minTickIntervalMillis() < policy.leaseRounds() - 1,
+ "Bootstrap membership policy requires explicit safe version-one timing bounds");
+ }
+
+ private void initializePolicy(Policy policy) throws IOException, InterruptedException {
+ long started = System.nanoTime();
+ long budget = TimeUnit.MILLISECONDS.toNanos(_startupMillis);
+ IOException lastFailure = null;
+ while (System.nanoTime() - started < budget) {
+ checkClock();
+ try {
+ Response current = requireNonNull(_access).query(new MembershipQuery(_history, new GetControl()),
+ remainingMillis(started, budget));
+ checkState(current.status() == Status.OK && current.control() != null,
+ "Cannot read membership bootstrap state: %s", current.status());
+ if (current.control().policy() != null) {
+ return;
+ }
+ if (current.control().controlEpoch() != null) {
+ // A timeout is an unknown result. Every retry still expects version zero, so an existing policy cannot
+ // be reset even after leadership changes or competing controllers bootstrap concurrently.
+ Response configured = requireNonNull(_access).execute(new MembershipCommand(_history, _instanceId,
+ new ConfigurePolicy(current.control().controlEpoch(), 0, policy)), remainingMillis(started, budget));
+ if (configured.status() == Status.OK) {
+ return;
+ }
+ checkState(configured.status() == Status.CONFLICT || configured.status() == Status.FENCED
+ || configured.status() == Status.NOT_READY, "Membership bootstrap rejected: %s", configured.status());
+ }
+ } catch (IOException e) {
+ lastFailure = e;
+ }
+ long remaining = budget - (System.nanoTime() - started);
+ if (remaining > 0) {
+ TimeUnit.NANOSECONDS.sleep(Math.min(remaining, TimeUnit.MILLISECONDS.toNanos(50)));
+ }
+ }
+ throw new IOException("Timed out initializing native membership through quorum", lastFailure);
+ }
+
+ private static long remainingMillis(long started, long budget) throws IOException {
+ long remaining = budget - (System.nanoTime() - started);
+ if (remaining <= 0) {
+ throw new IOException("Native controller startup deadline expired");
+ }
+ return Math.max(1, TimeUnit.NANOSECONDS.toMillis(remaining));
+ }
+
+ private void checkClock() throws IOException {
+ String failure = requireNonNull(_store).membershipClockFailure();
+ if (failure != null) {
+ throw new IOException("Membership clock qualification failed: " + failure);
+ }
+ }
+
+ @Override
+ public ServiceStatus.Status getServiceStatus() {
+ if (!_started || _closed) {
+ return ServiceStatus.Status.NOT_STARTED;
+ }
+ try {
+ checkClock();
+ Response response = requireNonNull(_access).query(new MembershipQuery(_history, new GetControl()), 1000);
+ if (response.status() != Status.OK || response.control() == null || response.control().policy() == null) {
+ throw new IOException("Committed membership policy is unavailable");
+ }
+ _status = "Native metadata endpoint and membership initialized; controller HTTP and query serving are disabled";
+ return ServiceStatus.Status.GOOD;
+ } catch (Exception e) {
+ _status = "Native metadata is unavailable: " + e.getMessage();
+ return ServiceStatus.Status.BAD;
+ }
+ }
+
+ @Override
+ public String getStatusDescription() {
+ return _status;
+ }
+
+ @Override
+ public synchronized void close() throws Exception {
+ if (_closed) {
+ return;
+ }
+ _closed = true;
+ _started = false;
+ Exception failure = null;
+ for (AutoCloseable resource : new AutoCloseable[]{_endpoint, _access, _store, _provider}) {
+ if (resource != null) {
+ try {
+ resource.close();
+ } catch (Exception e) {
+ if (failure == null) {
+ failure = e;
+ } else {
+ failure.addSuppressed(e);
+ }
+ }
+ }
+ }
+ if (failure != null) {
+ throw failure;
+ }
+ }
+}
diff --git a/pinot-coordination-ratis/README.md b/pinot-coordination-ratis/README.md
index 031dffad06f..3c821f56878 100644
--- a/pinot-coordination-ratis/README.md
+++ b/pinot-coordination-ratis/README.md
@@ -24,7 +24,8 @@ This internal, experimental module develops metadata, recovery and native offlin
Ratis. It targets Java 25, keeps Ratis dependencies outside the Java 11 common/SPI artifacts, and uses Apache Ratis 3.3.1.
The module is not a complete ZooKeeper or Helix replacement. Existing Pinot starters and quickstarts continue to use
-Helix/ZooKeeper by default. No normal service starts a Ratis voter merely because this module is built.
+Helix/ZooKeeper by default. Explicit `pinot.coordination.backend=RATIS` selects the bounded native starter lifecycle
+described below; building this module alone does not select it.
The initial catalog stores cluster configuration, schemas and table configurations. Conditional transactions check
history, object identity and version before applying any mutation. Deleted object identities remain reserved so a
@@ -252,8 +253,8 @@ acknowledgement. Recovery runs independently of operation discovery, so a comple
snapshot cannot strand the report stream. Old-session files are retained as evidence, never replayed under a new
session. Evidence is bounded to 128 files of at most 256 KiB each; saturation requires explicit archival. Startup and
shutdown serialize; shutdown drains local actions and outbox leases before releasing directory ownership, then
-retires the confirmed registration. Normal starter configuration, network query integration and general table lifecycle remain
-unimplemented; this runtime does not make the replacement complete or alter Helix defaults.
+retires the confirmed registration. Normal starter configuration is available for the installation-only lifecycle below. Network query integration and
+general table lifecycle remain unimplemented; this runtime does not make the replacement complete or alter Helix defaults.
`RatisOfflineSegmentExecutor` connects fresh own-target work to the native `LOAD`, `RELOAD` or `RECONCILE` action and receipt publisher.
The installer verifies the downloaded SHA-256 and segment CRC, runs Pinot's existing preprocessing, and opens the
@@ -273,7 +274,7 @@ its installation ID. Forced download bypasses local-byte reuse. Ordinary `LOAD`
Publication refreshes work and checks local authority and the expected prior installation under the segment lock;
consensus revalidates the receipt against later global changes. Neither local installation nor a committed receipt
grants query-serving rights. This path currently supports ordinary OFFLINE tables with the default local segment
-directory loader, without tiers, peer download, upsert or dimension tables. Normal-role starters, controller REST
+directory loader, without tiers, peer download, upsert or dimension tables. Controller REST
actions, controller work scheduling, routing publication and complete query-serving fences remain absent. This is an internal execution
path, not a complete controller/server reload integration or a supported Helix/ZooKeeper replacement.
@@ -314,6 +315,82 @@ production clock qualification, broker deadline propagation, transport response-
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.
+## Normal starters (experimental, query listeners disabled)
+
+`StartController`, `StartBroker`, and `StartServer` use their existing starter classes. Set
+`pinot.coordination.backend=RATIS` in each component config to select the native lifecycle **before** any Helix/ZK
+bootstrap or cluster-config read. Missing backend configuration retains `HELIX`; unknown values fail startup. Backend
+selection, process identity, peers, TLS and local storage are bootstrap settings, not runtime cluster toggles.
+
+- Controller starts one embedded voter, its explicitly qualified membership clock and the mutually authenticated
+ metadata/membership/installation endpoint. It initializes an absent membership policy with expected version zero;
+ restart or a competing controller cannot overwrite an existing committed policy.
+- Server starts the existing native offline participant with independent membership/work connections, a locked durable
+ 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.
+
+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 controller HTTP/UI resources and server admin resources are not started on the native
+path. Minions and multi-cluster brokers are unsupported. These entry points exercise coordination and offline
+installation; they do not yet provide a queryable replacement cluster or the normal controller UI.
+
+All native components need `pinot.coordination.ratis.historyId`. Controllers retain the voter settings in the next
+section and additionally require:
+
+```properties
+pinot.coordination.backend=RATIS
+pinot.coordination.ratis.metadata.host=127.0.0.1
+pinot.coordination.ratis.metadata.port=24001
+pinot.coordination.ratis.tls.certChain=/absolute/path/controller.pem
+pinot.coordination.ratis.tls.privateKey=/absolute/path/controller-key.pem
+pinot.coordination.ratis.tls.trustCerts=/absolute/path/trust.pem
+pinot.coordination.ratis.controller.provider.class=your.deployment.QualifiedRatisControllerProvider
+```
+
+The provider implements `RatisControllerProvider`: initialization, qualified `MembershipClockSettings`, current
+certificate-to-controller and certificate-to-participant policies, and an explicit safe version-one bootstrap
+membership policy. There is **no bundled production clock provider or timing default**. Supplying a class name or
+positive bounds does not qualify a platform. Test providers are only test fixtures. Participant policies with serving
+capabilities are denied by this starter; certificate identities never come from RPC request fields. Controller
+startup waits for quorum and policy initialization, bounded by `controller.startup.timeoutMs` (default 30,000;
+maximum 300,000), so launch the three controller processes concurrently. Current Raft peers remain loopback-only.
+`StartServiceManager -bootstrapConfigPaths ...` starts configured Ratis controllers concurrently and waits for them
+before starting broker/server roles; its existing sequential Helix controller bootstrap is preserved.
+
+Broker and server configs use:
+
+```properties
+pinot.coordination.backend=RATIS
+pinot.coordination.ratis.historyId=<the-controller-history-uuid>
+pinot.coordination.ratis.participantId=Server_native1
+pinot.coordination.ratis.metadata.endpoints=127.0.0.1:24001,127.0.0.1:24002,127.0.0.1:24003
+pinot.coordination.ratis.tls.certChain=/absolute/path/participant.pem
+pinot.coordination.ratis.tls.privateKey=/absolute/path/participant-key.pem
+pinot.coordination.ratis.tls.trustCerts=/absolute/path/trust.pem
+```
+
+Use a `Broker_` participant ID for a broker and authorize each certificate for that exact role and stable ID.
+Servers additionally require `pinot.server.instance.id` equal to the participant ID and explicit absolute
+`pinot.server.instance.dataDir` and `pinot.server.instance.segmentTarDir` paths. Existing filesystem, fetcher and crypter
+plugin settings are retained. `rpc.timeoutMs` defaults to 5,000 and `heartbeatMillis` to 1,000 under the Ratis prefix;
+both are bounded to 1–30,000 milliseconds. Server-specific `server.snapshotMillis`, `server.actionThreads`, and
+`server.queuedActions` default to 1,000, 1, and 128. Scheduling intervals must fit the committed membership policy;
+these transport controls do not qualify clocks or grant serving authority.
+
+After supplying deployment-qualified providers, TLS identities and all voter settings, use the existing commands:
+
+```bash
+bin/pinot-admin.sh StartController -configFileName controller-1.conf
+bin/pinot-admin.sh StartBroker -configFileName broker.conf
+bin/pinot-admin.sh StartServer -configFileName server.conf
+```
+
+Startup failure closes owned channels, voters and directory ownership. Shutdown retires only the process's confirmed
+registration; unknown remote outcomes rely on lease expiry. The existing Helix/ZK quickstarts are unchanged.
+
## Voter configuration and validation
The voter configuration uses the `pinot.coordination.ratis.` prefix with `clusterId`, `historyId`, `id`, `voters`,
@@ -348,7 +425,7 @@ Validate the shared query wire context and deadline/permit composition:
-Dsurefire.failIfNoSpecifiedTests=false test
```
-Metadata durability, distribution and membership contracts are parts of the replacement. Normal role integration,
+Metadata durability, distribution, membership and starter wiring are parts of the replacement. Full role integration,
qualified liveness and query fencing, controller ownership integration, assignment and state transitions, realtime
ingestion, minion tasks, migration and a standalone Ratis quickstart require their own integration and qualification
before a supported replacement is claimed.
diff --git a/pinot-server/src/main/java/org/apache/pinot/server/starter/NativeMetadataConnection.java b/pinot-server/src/main/java/org/apache/pinot/server/starter/NativeMetadataConnection.java
index 7c027d67013..19f007b112c 100644
--- a/pinot-server/src/main/java/org/apache/pinot/server/starter/NativeMetadataConnection.java
+++ b/pinot-server/src/main/java/org/apache/pinot/server/starter/NativeMetadataConnection.java
@@ -18,109 +18,16 @@
*/
package org.apache.pinot.server.starter;
-import io.grpc.Status;
-import io.grpc.StatusRuntimeException;
import java.io.IOException;
import java.nio.file.Path;
-import java.util.ArrayList;
import java.util.List;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.function.BiFunction;
-import org.apache.pinot.common.coordination.MetadataRpcClient;
-import org.apache.pinot.coordination.ratis.rpc.MetadataRpc;
+import org.apache.pinot.common.coordination.MetadataRpcConnection;
-import static com.google.common.base.Preconditions.checkArgument;
-
-
-/// Bounded endpoint failover that resends the identical application request after an unknown transport outcome.
-/// Instances own their channels; membership uses a separate instance from metadata hydration and installation.
-public final class NativeMetadataConnection implements NativeServerParticipant.Rpc {
- private final List<MetadataRpcClient> _clients;
- private final AtomicInteger _preferred = new AtomicInteger();
-
- public record Endpoint(String host, int port) {
- public Endpoint {
- checkArgument(host != null && !host.isBlank() && port > 0 && port <= 65535, "Invalid metadata endpoint");
- }
- }
+/// Server installation adapter for the shared bounded metadata transport. Each instance owns its TLS channels.
+public final class NativeMetadataConnection extends MetadataRpcConnection implements NativeServerParticipant.Rpc {
public NativeMetadataConnection(List<Endpoint> endpoints, Path certificate, Path key, Path trust,
long timeoutMillis) throws IOException {
- checkArgument(endpoints != null && !endpoints.isEmpty() && endpoints.size() <= 16,
- "Between one and sixteen metadata endpoints are required");
- List<MetadataRpcClient> clients = new ArrayList<>();
- try {
- for (Endpoint endpoint : List.copyOf(endpoints)) {
- clients.add(new MetadataRpcClient(endpoint.host(), endpoint.port(), certificate, key, trust, timeoutMillis));
- }
- } catch (IOException | RuntimeException e) {
- clients.forEach(MetadataRpcClient::close);
- throw e;
- }
- _clients = List.copyOf(clients);
- }
-
- public MetadataRpc.MembershipResponse membershipRead(MetadataRpc.MembershipReadRequest request, long timeoutMillis) {
- return call(timeoutMillis, (client, timeout) -> client.membershipRead(request, timeout));
- }
-
- public MetadataRpc.MembershipResponse membershipExecute(MetadataRpc.MembershipCommandRequest request,
- long timeoutMillis) {
- return call(timeoutMillis, (client, timeout) -> client.membershipExecute(request, timeout));
- }
-
- public MetadataRpc.InstallationSnapshot snapshot(MetadataRpc.InstallationSnapshotRequest request,
- long timeoutMillis) {
- return call(timeoutMillis, (client, timeout) -> client.installationSnapshot(request, timeout));
- }
-
- public MetadataRpc.InstallationWorkResult lookup(MetadataRpc.InstallationLookupRequest request, long timeoutMillis) {
- return call(timeoutMillis, (client, timeout) -> client.installationLookup(request, timeout));
- }
-
- @Override
- public MetadataRpc.CommandResult execute(MetadataRpc.InstallationCommandRequest request, long timeoutMillis) {
- return call(timeoutMillis, (client, timeout) -> client.installationExecute(request, timeout));
- }
-
- @Override
- public MetadataRpc.CommandResult outcome(MetadataRpc.InstallationOutcomeRequest request, long timeoutMillis) {
- return call(timeoutMillis, (client, timeout) -> client.installationOutcome(request, timeout));
- }
-
- private <T> T call(long timeoutMillis, BiFunction<MetadataRpcClient, Long, T> operation) {
- checkArgument(timeoutMillis > 0 && timeoutMillis <= 30_000, "Metadata deadline must be between 1 and 30000 ms");
- long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(timeoutMillis);
- int first = _preferred.get();
- StatusRuntimeException failure = null;
- for (int attempt = 0; attempt < _clients.size(); attempt++) {
- long remaining = TimeUnit.NANOSECONDS.toMillis(deadline - System.nanoTime());
- if (remaining <= 0) {
- break;
- }
- int selected = (first + attempt) % _clients.size();
- try {
- // Reserve time for the other configured endpoints instead of spending the whole deadline on a dead peer.
- T result = operation.apply(_clients.get(selected), Math.max(1, remaining / (_clients.size() - attempt)));
- _preferred.set(selected);
- return result;
- } catch (StatusRuntimeException e) {
- if (e.getStatus().getCode() != Status.Code.UNAVAILABLE
- && e.getStatus().getCode() != Status.Code.DEADLINE_EXCEEDED) {
- throw e;
- }
- failure = e;
- }
- }
- if (failure != null) {
- throw failure;
- }
- throw Status.DEADLINE_EXCEEDED.withDescription("Metadata endpoint deadline expired").asRuntimeException();
- }
-
- @Override
- public void close() {
- _clients.forEach(MetadataRpcClient::close);
+ super(endpoints, certificate, key, trust, timeoutMillis);
}
}
diff --git a/pinot-server/src/main/java/org/apache/pinot/server/starter/NativeServerStarterLifecycle.java b/pinot-server/src/main/java/org/apache/pinot/server/starter/NativeServerStarterLifecycle.java
new file mode 100644
index 00000000000..d64b6ebbff1
--- /dev/null
+++ b/pinot-server/src/main/java/org/apache/pinot/server/starter/NativeServerStarterLifecycle.java
@@ -0,0 +1,217 @@
+/**
+ * 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.server.starter;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.coordination.NativeMetadataConfig;
+import org.apache.pinot.common.metrics.ServerGauge;
+import org.apache.pinot.common.metrics.ServerMetrics;
+import org.apache.pinot.common.utils.ServiceStatus;
+import org.apache.pinot.common.utils.TarCompressionUtils;
+import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
+import org.apache.pinot.common.utils.tls.PinotInsecureMode;
+import org.apache.pinot.common.version.PinotVersion;
+import org.apache.pinot.segment.local.segment.store.TextIndexUtils;
+import org.apache.pinot.segment.local.utils.ClusterConfigForTable;
+import org.apache.pinot.segment.local.utils.SegmentOperationsThrottlerSet;
+import org.apache.pinot.segment.local.utils.ServerReloadJobStatusCache;
+import org.apache.pinot.segment.spi.index.IndexService;
+import org.apache.pinot.segment.spi.memory.PinotDataBuffer;
+import org.apache.pinot.segment.spi.memory.unsafe.MmapMemoryConfig;
+import org.apache.pinot.server.conf.ServerConf;
+import org.apache.pinot.spi.crypt.PinotCrypterFactory;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.CommonConstants.Server;
+import org.apache.pinot.spi.utils.PinotMd5Mode;
+import org.slf4j.Logger;
+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 the installation-only native branch of the normal server starter. Initialization validates local bootstrap
+/// inputs; start and close serialize ownership of metadata channels, participant, storage factories, and metrics.
+/// No Helix client, server admin endpoint, query listener, or query-serving capability is created.
+public final class NativeServerStarterLifecycle implements AutoCloseable {
+ private static final Logger LOGGER = LoggerFactory.getLogger(NativeServerStarterLifecycle.class);
+ private static final String PREFIX = "pinot.coordination.ratis.server.";
+ private final PinotConfiguration _config;
+ private final NativeMetadataConfig _metadataConfig;
+ private final NativeServerParticipant.Settings _settings;
+ @Nullable
+ private NativeServerParticipant _participant;
+ @Nullable
+ private ServerMetrics _metrics;
+ @Nullable
+ private ServerInstance _instance;
+ private boolean _started;
+ private boolean _closed;
+ private boolean _storageInitialized;
+ @Nullable
+ private ServiceStatus.ServiceStatusCallback _statusCallback;
+
+ public NativeServerStarterLifecycle(PinotConfiguration config) {
+ _config = config.clone();
+ _metadataConfig = new NativeMetadataConfig(_config);
+ String instanceId = _config.getProperty(Server.CONFIG_OF_INSTANCE_ID);
+ checkArgument(_metadataConfig.participantId().equals(instanceId),
+ "Native server requires an explicit instance ID matching pinot.coordination.ratis.participantId");
+ requireDirectory(_config, Server.CONFIG_OF_INSTANCE_DATA_DIR);
+ requireDirectory(_config, Server.CONFIG_OF_INSTANCE_SEGMENT_TAR_DIR);
+ checkArgument(new ServerConf(_config).getInstanceDataManagerClassName()
+ .equals(Server.DEFAULT_INSTANCE_DATA_MANAGER_CLASS),
+ "Custom instance data manager classes are not supported by the native starter");
+ _settings = new NativeServerParticipant.Settings(_metadataConfig.historyId(), instanceId, Map.of(),
+ _metadataConfig.rpcTimeoutMillis(), _metadataConfig.heartbeatMillis(),
+ _config.getProperty(PREFIX + "snapshotMillis", 1000L),
+ _config.getProperty(PREFIX + "actionThreads", 1), _config.getProperty(PREFIX + "queuedActions", 128));
+ }
+
+ private static void requireDirectory(PinotConfiguration config, String key) {
+ String value = config.getProperty(key);
+ checkArgument(value != null && !value.isBlank(), "Native server requires an explicit durable directory: %s", key);
+ checkArgument(Path.of(value).isAbsolute(), "Native server directory must be absolute: %s", key);
+ }
+
+ public String instanceId() {
+ return _settings.participantId();
+ }
+
+ public synchronized void start(SegmentOperationsThrottlerSet throttlers) throws Exception {
+ checkState(!_started && !_closed, "Native starter is already started or closed");
+ _started = true;
+ try {
+ LOGGER.info("Starting installation-only Ratis server: {}; admin and query listeners are disabled", instanceId());
+ ServerConf serverConf = new ServerConf(_config);
+ _metrics = ServerMetricsInitUtils.initServerMetrics(serverConf);
+ _metrics.setValueOfGlobalGauge(ServerGauge.VERSION, PinotVersion.VERSION_METRIC_NAME, 1);
+ throttlers.initializeMetrics();
+ PinotInsecureMode.setPinotInInsecureMode(
+ _config.getProperty(CommonConstants.CONFIG_OF_PINOT_INSECURE_MODE, false));
+ PinotMd5Mode.setPinotMd5Disabled(_config.getProperty(CommonConstants.CONFIG_OF_PINOT_MD5_DISABLED,
+ PinotMd5Mode.isPinotMd5Disabled()));
+ String codec = _config.getProperty(CommonConstants.CONFIG_OF_PINOT_TAR_COMPRESSION_CODEC_NAME);
+ if (codec != null) {
+ TarCompressionUtils.setDefaultCompressor(codec);
+ }
+ String mmapAdvice = _config.getProperty(Server.CONFIG_OF_MMAP_DEFAULT_ADVICE);
+ if (mmapAdvice != null) {
+ MmapMemoryConfig.setDefaultAdvice(mmapAdvice);
+ }
+ PinotDataBuffer.loadDefaultFactory(_config);
+ LOGGER.info("Loaded {} index types", IndexService.getInstance().getAllIndexes().size());
+ _storageInitialized = true;
+ initializeSegmentFetching(_config);
+ ServerReloadJobStatusCache jobs = new ServerReloadJobStatusCache(instanceId());
+ NativeMetadataConnection membership = connection();
+ NativeMetadataConnection work;
+ try {
+ work = connection();
+ } catch (Exception e) {
+ membership.close();
+ throw e;
+ }
+ try {
+ _participant = new NativeServerParticipant(_settings, membership, work);
+ } catch (RuntimeException e) {
+ work.close();
+ membership.close();
+ throw e;
+ }
+ _participant.clusterConfig().registerClusterConfigChangeListener(
+ new TextIndexUtils.LuceneMaxClauseCountConfigChangeListener());
+ _participant.clusterConfig().registerClusterConfigChangeListener(
+ new ClusterConfigForTable.ConfigChangeListener());
+ _participant.clusterConfig().registerClusterConfigChangeListener(jobs);
+ _participant.clusterConfig().registerClusterConfigChangeListener(throttlers);
+ _participant.start(serverConf.getInstanceDataManagerConfig(), _metrics, throttlers, jobs);
+ _instance = new ServerInstance(serverConf, instanceId(), _participant.dataManager(), jobs);
+ _statusCallback = new ServiceStatus.ServiceStatusCallback() {
+ @Override
+ public ServiceStatus.Status getServiceStatus() {
+ return _participant.isReadyForInstallations() ? ServiceStatus.Status.GOOD : ServiceStatus.Status.BAD;
+ }
+
+ @Override
+ public String getStatusDescription() {
+ return "Native installations only; query listeners disabled"
+ + (_participant.failure() == null ? "" : ": " + _participant.failure());
+ }
+ };
+ ServiceStatus.setServiceStatusCallback(instanceId(), _statusCallback);
+ } catch (Exception e) {
+ try {
+ close();
+ } catch (RuntimeException cleanup) {
+ e.addSuppressed(cleanup);
+ }
+ throw e;
+ }
+ }
+
+ private NativeMetadataConnection connection() throws IOException {
+ return new NativeMetadataConnection(_metadataConfig.endpoints(), _metadataConfig.certificate(),
+ _metadataConfig.privateKey(), _metadataConfig.trustCertificates(), _metadataConfig.rpcTimeoutMillis());
+ }
+
+ public synchronized ServerInstance serverInstance() {
+ return requireNonNull(_instance, "Native server is not started");
+ }
+
+ /// Shared with the Helix starter so both paths initialize the existing filesystem, fetcher and crypter plugins.
+ public static void initializeSegmentFetching(PinotConfiguration config) throws Exception {
+ PinotFSFactory.init(config.subset(Server.PREFIX_OF_CONFIG_OF_PINOT_FS_FACTORY));
+ SegmentFetcherFactory.init(config.subset(Server.PREFIX_OF_CONFIG_OF_SEGMENT_FETCHER_FACTORY));
+ PinotCrypterFactory.init(config.subset(Server.PREFIX_OF_CONFIG_OF_PINOT_CRYPTER));
+ }
+
+ @Override
+ public synchronized void close() {
+ if (_closed) {
+ return;
+ }
+ _closed = true;
+ if (_statusCallback != null) {
+ ServiceStatus.removeServiceStatusCallback(instanceId(), _statusCallback);
+ _statusCallback = null;
+ }
+ if (_participant != null) {
+ // Drain actions before releasing shared storage resources. A failed drain must retain those resources.
+ _participant.close();
+ }
+ try {
+ if (_storageInitialized) {
+ PinotFSFactory.shutdown();
+ }
+ } catch (IOException e) {
+ throw new IllegalStateException("Failed to close native storage factories", e);
+ } finally {
+ if (_metrics != null) {
+ _metrics.getMetricsRegistry().shutdown();
+ }
+ }
+ }
+}
diff --git a/pinot-server/src/main/java/org/apache/pinot/server/starter/ServerInstance.java b/pinot-server/src/main/java/org/apache/pinot/server/starter/ServerInstance.java
index bb562626419..465a3ba0aec 100644
--- a/pinot-server/src/main/java/org/apache/pinot/server/starter/ServerInstance.java
+++ b/pinot-server/src/main/java/org/apache/pinot/server/starter/ServerInstance.java
@@ -115,6 +115,15 @@ public class ServerInstance {
sendStatsPredicate, keepPipelineBreakerStatsPredicate, reloadJobStatusCache);
}
+ /// Wraps an initialized native installation manager without allocating query or access-control resources.
+ /// The participant that initialized the manager owns its start, drain, and close lifecycle.
+ public ServerInstance(ServerConf serverConf, String instanceId, NativeInstanceDataManager initializedDataManager,
+ ServerReloadJobStatusCache reloadJobStatusCache)
+ throws Exception {
+ this(serverConf, instanceId, null, requireNonNull(initializedDataManager, "initializedDataManager"), null, null,
+ null, null, null, null, reloadJobStatusCache);
+ }
+
private ServerInstance(ServerConf serverConf, String instanceId, @Nullable HelixManager helixManager,
@Nullable InstanceDataManager initializedDataManager, @Nullable AccessControlFactory accessControlFactory,
@Nullable AccessControl initializedAccessControl,
diff --git a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BaseServerStarter.java b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BaseServerStarter.java
index 7d90df3ac06..b4e6f82bc8f 100644
--- a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BaseServerStarter.java
+++ b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BaseServerStarter.java
@@ -60,6 +60,7 @@ import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.pinot.common.Utils;
import org.apache.pinot.common.config.DefaultClusterConfigChangeHandler;
import org.apache.pinot.common.config.TlsConfig;
+import org.apache.pinot.common.coordination.CoordinationBackend;
import org.apache.pinot.common.metadata.ZKMetadataProvider;
import org.apache.pinot.common.metrics.MseMetrics;
import org.apache.pinot.common.metrics.ServerGauge;
@@ -75,7 +76,6 @@ import org.apache.pinot.common.utils.ServiceStatus.Status;
import org.apache.pinot.common.utils.TarCompressionUtils;
import org.apache.pinot.common.utils.config.QueryWorkloadConfigUtils;
import org.apache.pinot.common.utils.config.TagNameUtils;
-import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
import org.apache.pinot.common.utils.helix.HelixHelper;
import org.apache.pinot.common.utils.regex.PatternFactory;
import org.apache.pinot.common.utils.tls.PinotInsecureMode;
@@ -115,6 +115,7 @@ import org.apache.pinot.server.api.resources.ReingestionConsumptionTimeout;
import org.apache.pinot.server.conf.ServerConf;
import org.apache.pinot.server.realtime.ControllerLeaderLocator;
import org.apache.pinot.server.realtime.ServerSegmentCompletionProtocolHandler;
+import org.apache.pinot.server.starter.NativeServerStarterLifecycle;
import org.apache.pinot.server.starter.ServerInstance;
import org.apache.pinot.server.starter.ServerMetricsInitUtils;
import org.apache.pinot.server.starter.ServerQueriesDisabledTracker;
@@ -124,7 +125,6 @@ import org.apache.pinot.spi.accounting.ThreadAccountantUtils;
import org.apache.pinot.spi.accounting.ThreadResourceUsageProvider;
import org.apache.pinot.spi.accounting.WorkloadBudgetManagerFactory;
import org.apache.pinot.spi.config.provider.PinotClusterConfigChangeListener;
-import org.apache.pinot.spi.crypt.PinotCrypterFactory;
import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.environmentprovider.PinotEnvironmentProvider;
import org.apache.pinot.spi.environmentprovider.PinotEnvironmentProviderFactory;
@@ -188,6 +188,9 @@ public abstract class BaseServerStarter implements ServiceStartable {
protected ScheduledExecutorService _helixMessageCountScheduler;
protected ServerReloadJobStatusCache _reloadJobStatusCache;
protected ReingestionConsumptionTimeout _reingestionConsumptionTimeout;
+ @Nullable
+ private NativeServerStarterLifecycle _nativeLifecycle;
+ private boolean _nativeBackend;
// Override this to provide custom thread pool for Helix state transitions. Null means using Helix's default
@Nullable
protected StateTransitionThreadPoolManager _transitionThreadPoolManager;
@@ -197,6 +200,13 @@ public abstract class BaseServerStarter implements ServiceStartable {
throws Exception {
// Make a clone so that changes to the config won't propagate to the caller
_serverConf = serverConf.clone();
+ _nativeBackend = CoordinationBackend.from(_serverConf) == CoordinationBackend.RATIS;
+ if (_nativeBackend) {
+ applyCustomConfigs(_serverConf);
+ _nativeLifecycle = new NativeServerStarterLifecycle(_serverConf);
+ _instanceId = _nativeLifecycle.instanceId();
+ return;
+ }
_zkAddress = _serverConf.getProperty(CommonConstants.Helix.CONFIG_OF_ZOOKEEPER_SERVER);
_helixClusterName = _serverConf.getProperty(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME);
ServiceStartableUtils.applyClusterConfig(_serverConf, _zkAddress, _helixClusterName, ServiceRole.SERVER);
@@ -659,6 +669,18 @@ public abstract class BaseServerStarter implements ServiceStartable {
@Override
public void start()
throws Exception {
+ if (_nativeBackend) {
+ Preconditions.checkState(_nativeLifecycle != null, "Native server initialization did not complete");
+ if (_segmentOperationsThrottlerSet == null) {
+ _segmentOperationsThrottlerSet = new SegmentOperationsThrottlerSet(createSegmentAllIndexPreprocessThrottler(),
+ createSegmentStarTreePreprocessThrottler(), createSegmentDownloadThrottler(),
+ createMultiColumnIndexPreprocessThrottler());
+ }
+ _nativeLifecycle.start(_segmentOperationsThrottlerSet);
+ _serverInstance = _nativeLifecycle.serverInstance();
+ _serverMetrics = _serverInstance.getServerMetrics();
+ return;
+ }
LOGGER.info("Starting Pinot server (Version: {})", PinotVersion.VERSION);
LOGGER.info("Server configs: {}", new PinotAppConfigs(getConfig()).toJSONString());
long startTimeMs = System.currentTimeMillis();
@@ -1056,6 +1078,12 @@ public abstract class BaseServerStarter implements ServiceStartable {
@Override
public void stop() {
+ if (_nativeBackend) {
+ if (_nativeLifecycle != null) {
+ _nativeLifecycle.close();
+ }
+ return;
+ }
LOGGER.info("Shutting down Pinot server");
long startTimeMs = System.currentTimeMillis();
@@ -1274,15 +1302,7 @@ public abstract class BaseServerStarter implements ServiceStartable {
/// components are initialized where SegmentFetcherAndLoader was initialized.
private void initSegmentFetcher(PinotConfiguration config)
throws Exception {
- PinotConfiguration pinotFSConfig = config.subset(CommonConstants.Server.PREFIX_OF_CONFIG_OF_PINOT_FS_FACTORY);
- PinotFSFactory.init(pinotFSConfig);
-
- PinotConfiguration segmentFetcherFactoryConfig =
- config.subset(CommonConstants.Server.PREFIX_OF_CONFIG_OF_SEGMENT_FETCHER_FACTORY);
- SegmentFetcherFactory.init(segmentFetcherFactoryConfig);
-
- PinotConfiguration pinotCrypterConfig = config.subset(CommonConstants.Server.PREFIX_OF_CONFIG_OF_PINOT_CRYPTER);
- PinotCrypterFactory.init(pinotCrypterConfig);
+ NativeServerStarterLifecycle.initializeSegmentFetching(config);
}
protected AdminApiApplication createServerAdminApp() {
diff --git a/pinot-server/src/test/java/org/apache/pinot/server/starter/helix/NativeServerStarterTest.java b/pinot-server/src/test/java/org/apache/pinot/server/starter/helix/NativeServerStarterTest.java
new file mode 100644
index 00000000000..969097e7837
--- /dev/null
+++ b/pinot-server/src/test/java/org/apache/pinot/server/starter/helix/NativeServerStarterTest.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.server.starter.helix;
+
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.commons.io.FileUtils;
+import org.apache.helix.HelixManagerFactory;
+import org.apache.pinot.common.utils.ServiceStatus;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.MembershipCommandRequest;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.ResultStatus;
+import org.apache.pinot.server.starter.NativeInstanceDataManager;
+import org.apache.pinot.server.starter.NativeMetadataConnection;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Helix;
+import org.apache.pinot.spi.utils.CommonConstants.Server;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
+import org.testng.annotations.Test;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
+
+
+/// Exercises the normal starter's native dispatch and real participant lifecycle with controlled metadata replies.
+/// Transport authentication and actual quorum composition are covered by the metadata transport integration tests.
+public class NativeServerStarterTest {
+ @Test
+ public void testNativeStarterOwnsInstallationLifecycleWithoutHelixOrListeners() throws Exception {
+ try (Fixture fixture = new Fixture(); ServerSocket occupied = new ServerSocket(0);
+ MockedStatic<HelixManagerFactory> helix = mockStatic(HelixManagerFactory.class);
+ MockedConstruction<NativeMetadataConnection> connections = fixture.connections()) {
+ PinotConfiguration config = fixture.config();
+ config.setProperty(Helix.KEY_OF_SERVER_NETTY_PORT, occupied.getLocalPort());
+ config.setProperty(Server.CONFIG_OF_ADMIN_API_PORT, occupied.getLocalPort());
+ HelixServerStarter starter = new HelixServerStarter();
+ try {
+ starter.init(config);
+ assertFalse(Files.exists(fixture._directory.resolve("data")));
+ assertTrue(connections.constructed().isEmpty(), "Initialization must not open metadata channels");
+ starter.start();
+ assertEquals(connections.constructed().size(), 2, "Membership and work need independent admission");
+ assertEquals(starter.getInstanceId(), Fixture.INSTANCE_ID);
+ assertEquals(ServiceStatus.getServiceStatus(Fixture.INSTANCE_ID), ServiceStatus.Status.GOOD);
+ assertTrue(ServiceStatus.getStatusDescription(Fixture.INSTANCE_ID).contains("query listeners disabled"));
+ HelixServerStarter unstarted = new HelixServerStarter();
+ unstarted.init(fixture.config());
+ unstarted.stop();
+ assertEquals(ServiceStatus.getServiceStatus(Fixture.INSTANCE_ID), ServiceStatus.Status.GOOD,
+ "Closing an unstarted duplicate must not remove the running participant's status");
+ assertTrue(starter.getServerInstance().getInstanceDataManager() instanceof NativeInstanceDataManager);
+ assertNull(starter._helixManager);
+ assertNull(starter._helixAdmin);
+ assertNull(starter._adminApiApplication);
+ assertThrows(IllegalStateException.class, starter.getServerInstance()::getQueryScheduler);
+ assertFalse(starter.isServerReadyToServeQueries());
+ assertThrows(IllegalStateException.class, starter.getServerInstance()::startQueryServer);
+ assertThrows(IllegalStateException.class, starter.getServerInstance()::getHelixManager);
+ assertTrue(fixture._token != null && fixture._registeredEndpoints == 0);
+ } finally {
+ starter.stop();
+ }
+ starter.stop();
+ assertNotNull(fixture._drained);
+ assertNotNull(fixture._unregistered);
+ assertEquals(ServiceStatus.getServiceStatus(Fixture.INSTANCE_ID), ServiceStatus.Status.NOT_STARTED);
+ connections.constructed().forEach(connection -> verify(connection).close());
+ helix.verifyNoInteractions();
+ // A second process lifecycle can take the same durable directory after the first participant drains.
+ HelixServerStarter restarted = new HelixServerStarter();
+ try {
+ restarted.init(fixture.config());
+ restarted.start();
+ assertEquals(ServiceStatus.getServiceStatus(Fixture.INSTANCE_ID), ServiceStatus.Status.GOOD);
+ } finally {
+ restarted.stop();
+ }
+ }
+ }
+
+ @Test
+ public void testFailedRegistrationClosesChannelsAndReleasesDirectory() throws Exception {
+ try (Fixture fixture = new Fixture();
+ MockedStatic<HelixManagerFactory> helix = mockStatic(HelixManagerFactory.class);
+ MockedConstruction<NativeMetadataConnection> connections = fixture.connections()) {
+ fixture._rejectRegistration = true;
+ HelixServerStarter starter = new HelixServerStarter();
+ starter.init(fixture.config());
+ assertThrows(IOException.class, starter::start);
+ starter.stop();
+ connections.constructed().forEach(connection -> verify(connection).close());
+ assertEquals(ServiceStatus.getServiceStatus(Fixture.INSTANCE_ID), ServiceStatus.Status.NOT_STARTED);
+ fixture._rejectRegistration = false;
+ HelixServerStarter recovered = new HelixServerStarter();
+ try {
+ recovered.init(fixture.config());
+ recovered.start();
+ assertEquals(ServiceStatus.getServiceStatus(Fixture.INSTANCE_ID), ServiceStatus.Status.GOOD);
+ } finally {
+ recovered.stop();
+ }
+ helix.verifyNoInteractions();
+ }
+ }
+
+ @Test
+ public void testNativeBootstrapRejectsMissingStorageOrMismatchedIdentityBeforeResources() throws Exception {
+ try (Fixture fixture = new Fixture();
+ MockedStatic<HelixManagerFactory> helix = mockStatic(HelixManagerFactory.class);
+ MockedConstruction<NativeMetadataConnection> connections = fixture.connections()) {
+ PinotConfiguration missingStorage = fixture.config();
+ missingStorage.setProperty(Server.CONFIG_OF_INSTANCE_DATA_DIR, "");
+ HelixServerStarter failed = new HelixServerStarter();
+ assertThrows(IllegalArgumentException.class, () -> failed.init(missingStorage));
+ failed.stop();
+ assertThrows(IllegalStateException.class, failed::start);
+ PinotConfiguration wrongIdentity = fixture.config();
+ wrongIdentity.setProperty(Server.CONFIG_OF_INSTANCE_ID, "other-server");
+ assertThrows(IllegalArgumentException.class, () -> new HelixServerStarter().init(wrongIdentity));
+ PinotConfiguration unknownBackend = fixture.config();
+ unknownBackend.setProperty("pinot.coordination.backend", "rtais");
+ assertThrows(IllegalArgumentException.class, () -> new HelixServerStarter().init(unknownBackend));
+ HelixServerStarter initialized = new HelixServerStarter();
+ initialized.init(fixture.config());
+ initialized.stop();
+ assertFalse(Files.exists(fixture._directory.resolve("data")));
+ assertTrue(connections.constructed().isEmpty());
+ helix.verifyNoInteractions();
+ }
+ }
+
+ private static final class Fixture implements AutoCloseable {
+ private static final String INSTANCE_ID = "Server_native_starter";
+ private final Path _directory = Files.createTempDirectory("native-starter-");
+ private final UUID _history = UUID.randomUUID();
+ private final UUID _activation = UUID.randomUUID();
+ private final AtomicLong _index = new AtomicLong(1);
+ private volatile MetadataRpc.ParticipantToken _token;
+ private volatile MetadataRpc.ParticipantSession _drained;
+ private volatile MetadataRpc.ParticipantSession _unregistered;
+ private volatile boolean _ready;
+ private volatile boolean _rejectRegistration;
+ private volatile int _registeredEndpoints;
+
+ private Fixture() throws IOException {
+ // These readable placeholders are never parsed: this fixture controls the metadata channel boundary.
+ Files.writeString(_directory.resolve("tls.pem"), "fixture");
+ }
+
+ private PinotConfiguration config() {
+ return new PinotConfiguration(Map.ofEntries(
+ Map.entry("pinot.coordination.backend", "RATIS"),
+ Map.entry("pinot.coordination.ratis.historyId", _history.toString()),
+ Map.entry("pinot.coordination.ratis.participantId", INSTANCE_ID),
+ Map.entry("pinot.coordination.ratis.metadata.endpoints", "localhost:1"),
+ Map.entry("pinot.coordination.ratis.tls.certChain", _directory.resolve("tls.pem").toString()),
+ Map.entry("pinot.coordination.ratis.tls.privateKey", _directory.resolve("tls.pem").toString()),
+ Map.entry("pinot.coordination.ratis.tls.trustCerts", _directory.resolve("tls.pem").toString()),
+ Map.entry("pinot.coordination.ratis.heartbeatMillis", 10),
+ Map.entry("pinot.coordination.ratis.server.snapshotMillis", 10),
+ Map.entry(Server.CONFIG_OF_INSTANCE_ID, INSTANCE_ID),
+ Map.entry(Server.CONFIG_OF_INSTANCE_DATA_DIR, _directory.resolve("data").toString()),
+ Map.entry(Server.CONFIG_OF_INSTANCE_SEGMENT_TAR_DIR, _directory.resolve("tar").toString())));
+ }
+
+ private MockedConstruction<NativeMetadataConnection> connections() {
+ return mockConstruction(NativeMetadataConnection.class, (connection, context) -> {
+ when(connection.membershipRead(any(), anyLong())).thenAnswer(ignored -> response(ResultStatus.NOT_FOUND));
+ when(connection.membershipExecute(any(), anyLong())).thenAnswer(invocation ->
+ membership(invocation.getArgument(0)));
+ when(connection.snapshot(any(), anyLong())).thenAnswer(ignored -> snapshot());
+ when(connection.execute(any(), anyLong())).thenAnswer(ignored -> MetadataRpc.CommandResult.newBuilder()
+ .setStatus(ResultStatus.OK).setClientEpoch(1).setRevision(revision()).build());
+ });
+ }
+
+ private MetadataRpc.MembershipResponse membership(MembershipCommandRequest request) {
+ switch (request.getCommandCase()) {
+ case REGISTER:
+ if (_rejectRegistration) {
+ return response(ResultStatus.CONFLICT);
+ }
+ _token = MetadataRpc.ParticipantToken.newBuilder().setHistoryId(_history.toString())
+ .setParticipantId(INSTANCE_ID).setRole(MetadataRpc.ParticipantRole.SERVER)
+ .setIncarnation(request.getRegister().getIncarnation())
+ .setRegistrationEpoch(request.getRegister().getExpectedEpoch() + 1).build();
+ _registeredEndpoints = request.getRegister().getEndpointsCount();
+ _ready = false;
+ return response(ResultStatus.OK).toBuilder().setParticipant(participant()).build();
+ case READY:
+ _ready = request.getReady().getReady();
+ return response(ResultStatus.OK).toBuilder().setParticipant(participant()).build();
+ case RENEW:
+ return response(ResultStatus.OK).toBuilder().addRenewals(MetadataRpc.RenewalResult.newBuilder()
+ .setParticipant(_token).setSequence(request.getRenew().getSequence()).setStatus(ResultStatus.OK)).build();
+ case DRAIN:
+ _drained = request.getDrain();
+ return response(ResultStatus.OK);
+ case UNREGISTER:
+ _unregistered = request.getUnregister();
+ return response(ResultStatus.OK);
+ default:
+ throw new AssertionError("Unexpected membership operation: " + request.getCommandCase());
+ }
+ }
+
+ private MetadataRpc.ParticipantState participant() {
+ return MetadataRpc.ParticipantState.newBuilder().setParticipant(_token).setReady(_ready).setExpiryRound(100)
+ .setLastModifiedIndex(1).setAppliedRevision(MetadataRpc.Revision.newBuilder()
+ .setHistoryId(_history.toString()).setIndex(1)).build();
+ }
+
+ private MetadataRpc.InstallationSnapshot snapshot() {
+ return MetadataRpc.InstallationSnapshot.newBuilder().setStatus(ResultStatus.OK).setRevision(revision())
+ .setParticipant(participant()).setControl(MetadataRpc.MembershipControl.newBuilder()
+ .setClockBarrierIndex(1).setMinimumTickDelayMillis(100).setMaxConfiguredPermitLifetimeMillis(100)
+ .setControlEpoch(MetadataRpc.ControlToken.newBuilder().setHistoryId(_history.toString()).setRaftTerm(1)
+ .setBarrierIndex(1).setLeaderId("controller").setActivationId(_activation.toString()))
+ .setPolicy(MetadataRpc.MembershipPolicy.newBuilder().setVersion(1).setMinTickIntervalMillis(100)
+ .setLeaseRounds(100).setPermitLifetimeMillis(100).setMaxQueryLifetimeMillis(100)
+ .setMaxRevisionLag(1000))).build();
+ }
+
+ private MetadataRpc.MembershipResponse response(ResultStatus status) {
+ return MetadataRpc.MembershipResponse.newBuilder().setStatus(status).setRevision(revision()).build();
+ }
+
+ private MetadataRpc.Revision revision() {
+ return MetadataRpc.Revision.newBuilder().setHistoryId(_history.toString()).setIndex(_index.incrementAndGet())
+ .build();
+ }
+
+ @Override
+ public void close() throws IOException {
+ FileUtils.deleteDirectory(_directory.toFile());
+ }
+ }
+}
diff --git a/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/StartServiceManagerCommand.java b/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/StartServiceManagerCommand.java
index e4d1674715a..1140c67060b 100644
--- a/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/StartServiceManagerCommand.java
+++ b/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/StartServiceManagerCommand.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.tools.admin.command;
+import com.google.common.annotations.VisibleForTesting;
import java.io.File;
import java.net.SocketException;
import java.net.UnknownHostException;
@@ -30,7 +31,9 @@ import java.util.Map.Entry;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.pinot.common.coordination.CoordinationBackend;
import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.services.ServiceRole;
import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.tools.Command;
@@ -46,7 +49,7 @@ import static org.apache.pinot.spi.utils.CommonConstants.Helix.PINOT_SERVICE_ROL
/// Starts services in the following order, and returns false on any startup failure:
///
/// 1. [PinotServiceManager]
-/// 2. Bootstrap services in role [ServiceRole#CONTROLLER]
+/// 2. Helix controllers sequentially, then native controllers concurrently so their voters can form a quorum
/// 3. All remaining bootstrap services in parallel
@CommandLine.Command(name = "StartServiceManager", mixinStandardHelpOptions = true)
public class StartServiceManagerCommand extends AbstractBaseAdminCommand implements Command {
@@ -218,22 +221,29 @@ public class StartServiceManagerCommand extends AbstractBaseAdminCommand impleme
}
}
- /// Starts a controller synchronously unless the cluster already exists. Other services start in parallel.
+ /// Completes controller bootstrap before starting other roles.
private boolean startBootstrapServices() {
- if (_bootstrapConfigurations.isEmpty()) {
+ return startBootstrapServices(_pinotServiceManager, _bootstrapConfigurations);
+ }
+
+ @VisibleForTesting
+ static boolean startBootstrapServices(PinotServiceManager pinotServiceManager,
+ List<Entry<ServiceRole, Map<String, Object>>> bootstrapConfigurations) {
+ if (bootstrapConfigurations.isEmpty()) {
return true;
}
+ List<Entry<ServiceRole, Map<String, Object>>> nativeControllers = new ArrayList<>();
List<Entry<ServiceRole, Map<String, Object>>> parallelConfigs = new ArrayList<>();
- // Start controller(s) synchronously so that other services don't fail
- //
- // Note: Technically, we don't need to do this if the cluster already exists, but checking the
- // cluster takes time and clutters logs with errors when it doesn't exist.
- for (Entry<ServiceRole, Map<String, Object>> roleToConfig : _bootstrapConfigurations) {
+ // Preserve sequential Helix bootstrap. Native controllers wait for quorum during start(), so their voters
+ // must start together and complete before dependent broker/server roles are admitted.
+ for (Entry<ServiceRole, Map<String, Object>> roleToConfig : bootstrapConfigurations) {
if (roleToConfig.getKey() == ServiceRole.CONTROLLER) {
- if (!startPinotService(ServiceRole.CONTROLLER,
- () -> _pinotServiceManager.startRole(ServiceRole.CONTROLLER, roleToConfig.getValue()))) {
+ if (CoordinationBackend.from(new PinotConfiguration(roleToConfig.getValue())) == CoordinationBackend.RATIS) {
+ nativeControllers.add(roleToConfig);
+ } else if (!startPinotService(ServiceRole.CONTROLLER,
+ () -> pinotServiceManager.startRole(ServiceRole.CONTROLLER, roleToConfig.getValue()))) {
return false;
}
} else {
@@ -241,7 +251,8 @@ public class StartServiceManagerCommand extends AbstractBaseAdminCommand impleme
}
}
- return startBootstrapServicesInParallel(_pinotServiceManager, parallelConfigs);
+ return startBootstrapServicesInParallel(pinotServiceManager, nativeControllers)
+ && startBootstrapServicesInParallel(pinotServiceManager, parallelConfigs);
}
static boolean startBootstrapServicesInParallel(PinotServiceManager pinotServiceManager,
diff --git a/pinot-tools/src/main/java/org/apache/pinot/tools/service/PinotServiceManager.java b/pinot-tools/src/main/java/org/apache/pinot/tools/service/PinotServiceManager.java
index f1f2449d576..1bb4af364a9 100644
--- a/pinot-tools/src/main/java/org/apache/pinot/tools/service/PinotServiceManager.java
+++ b/pinot-tools/src/main/java/org/apache/pinot/tools/service/PinotServiceManager.java
@@ -22,6 +22,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.pinot.broker.broker.helix.HelixBrokerStarter;
+import org.apache.pinot.common.coordination.CoordinationBackend;
import org.apache.pinot.common.utils.ServiceStatus;
import org.apache.pinot.controller.ControllerStarter;
import org.apache.pinot.minion.MinionStarter;
@@ -106,7 +107,8 @@ public class PinotServiceManager {
if (!controllerConf.containsKey(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME)) {
controllerConf.setProperty(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME, _clusterName);
}
- if (!controllerConf.containsKey(CommonConstants.Helix.CONFIG_OF_ZOOKEEPER_SERVER)) {
+ if (CoordinationBackend.from(controllerConf) == CoordinationBackend.HELIX
+ && !controllerConf.containsKey(CommonConstants.Helix.CONFIG_OF_ZOOKEEPER_SERVER)) {
controllerConf.setProperty(CommonConstants.Helix.CONFIG_OF_ZOOKEEPER_SERVER, _zkAddress);
}
ServiceStartable controllerStarter = getServiceStartable(controllerStarterClassName);
@@ -124,7 +126,8 @@ public class PinotServiceManager {
if (!brokerConf.containsKey(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME)) {
brokerConf.setProperty(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME, _clusterName);
}
- if (!brokerConf.containsKey(CommonConstants.Helix.CONFIG_OF_ZOOKEEPER_SERVER)) {
+ if (CoordinationBackend.from(brokerConf) == CoordinationBackend.HELIX
+ && !brokerConf.containsKey(CommonConstants.Helix.CONFIG_OF_ZOOKEEPER_SERVER)) {
brokerConf.setProperty(CommonConstants.Helix.CONFIG_OF_ZOOKEEPER_SERVER, _zkAddress);
}
ServiceStartable brokerStarter;
@@ -155,7 +158,8 @@ public class PinotServiceManager {
serverConf.setProperty(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME, _clusterName);
}
- if (!serverConf.containsKey(CommonConstants.Helix.CONFIG_OF_ZOOKEEPER_SERVER)) {
+ if (CoordinationBackend.from(serverConf) == CoordinationBackend.HELIX
+ && !serverConf.containsKey(CommonConstants.Helix.CONFIG_OF_ZOOKEEPER_SERVER)) {
serverConf.setProperty(CommonConstants.Helix.CONFIG_OF_ZOOKEEPER_SERVER, _zkAddress);
}
ServiceStartable serverStarter = getServiceStartable(serverStarterClassName);
@@ -170,6 +174,9 @@ public class PinotServiceManager {
public String startMinion(String minionStarterClassName, PinotConfiguration minionConf)
throws Exception {
+ if (CoordinationBackend.from(minionConf) == CoordinationBackend.RATIS) {
+ throw new IllegalArgumentException("Ratis minion startup is not supported");
+ }
LOGGER.info("Trying to start Pinot Minion...");
if (!minionConf.containsKey(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME)) {
minionConf.setProperty(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME, _clusterName);
diff --git a/pinot-tools/src/main/java/org/apache/pinot/tools/utils/PinotConfigUtils.java b/pinot-tools/src/main/java/org/apache/pinot/tools/utils/PinotConfigUtils.java
index 4cef85fc900..be41b23ee3d 100644
--- a/pinot-tools/src/main/java/org/apache/pinot/tools/utils/PinotConfigUtils.java
+++ b/pinot-tools/src/main/java/org/apache/pinot/tools/utils/PinotConfigUtils.java
@@ -31,6 +31,7 @@ import javax.annotation.Nullable;
import org.apache.commons.configuration2.builder.fluent.Configurations;
import org.apache.commons.configuration2.ex.ConfigurationException;
import org.apache.commons.lang3.StringUtils;
+import org.apache.pinot.common.coordination.CoordinationBackend;
import org.apache.pinot.controller.ControllerConf;
import org.apache.pinot.controller.ControllerConf.ControllerPeriodicTasksConf;
import org.apache.pinot.core.accounting.ResourceUsageAccountantFactory;
@@ -132,6 +133,12 @@ public class PinotConfigUtils {
String.format(CONTROLLER_CONFIG_VALIDATION_ERROR_MESSAGE_FORMAT, "null conf object."));
}
+ if (CoordinationBackend.from(conf) == CoordinationBackend.RATIS) {
+ // The native starter validates voter identity, qualified providers and TLS before opening any resources.
+ // Its metadata endpoint does not use the legacy controller HTTP listener or Helix cluster settings.
+ return true;
+ }
+
List<String> protocols = validateControllerAccessProtocols(conf);
if (conf.getControllerPort() == null && protocols.isEmpty()) {
diff --git a/pinot-tools/src/test/java/org/apache/pinot/tools/admin/command/StartServiceManagerCommandTest.java b/pinot-tools/src/test/java/org/apache/pinot/tools/admin/command/StartServiceManagerCommandTest.java
new file mode 100644
index 00000000000..cf9b6d6b0e4
--- /dev/null
+++ b/pinot-tools/src/test/java/org/apache/pinot/tools/admin/command/StartServiceManagerCommandTest.java
@@ -0,0 +1,81 @@
+/**
+ * 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.tools.admin.command;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.pinot.common.coordination.CoordinationBackend;
+import org.apache.pinot.spi.services.ServiceRole;
+import org.apache.pinot.tools.service.PinotServiceManager;
+import org.testng.annotations.Test;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertTrue;
+
+
+/// Ensures normal bootstrap can form a native voter quorum while preserving controller-before-participant ordering.
+public class StartServiceManagerCommandTest {
+ @Test(timeOut = 10_000)
+ public void testNativeVotersStartTogetherAfterSequentialHelixControllers() throws Exception {
+ PinotServiceManager manager = mock(PinotServiceManager.class);
+ CountDownLatch nativeVoters = new CountDownLatch(3);
+ AtomicInteger helixControllers = new AtomicInteger();
+ AtomicInteger completedVoters = new AtomicInteger();
+ AtomicInteger participants = new AtomicInteger();
+ Thread bootstrapThread = Thread.currentThread();
+ when(manager.startRole(any(), any())).thenAnswer(invocation -> {
+ ServiceRole role = invocation.getArgument(0);
+ Map<String, Object> config = invocation.getArgument(1);
+ if (role == ServiceRole.CONTROLLER) {
+ if ("RATIS".equals(config.get(CoordinationBackend.CONFIG_KEY))) {
+ assertEquals(helixControllers.get(), 2);
+ nativeVoters.countDown();
+ assertTrue(nativeVoters.await(5, TimeUnit.SECONDS), "Voters must start together to form a quorum");
+ completedVoters.incrementAndGet();
+ } else {
+ assertSame(Thread.currentThread(), bootstrapThread, "Default Helix controllers remain synchronous");
+ assertEquals(helixControllers.incrementAndGet(), ((Number) config.get("ordinal")).intValue());
+ }
+ } else {
+ assertEquals(helixControllers.get(), 2);
+ assertEquals(completedVoters.get(), 3, "Participants must wait for every native voter startup");
+ participants.incrementAndGet();
+ }
+ return role.toString();
+ });
+ List<Entry<ServiceRole, Map<String, Object>>> configurations = List.of(
+ Map.entry(ServiceRole.BROKER, Map.of()),
+ Map.entry(ServiceRole.CONTROLLER, Map.of(CoordinationBackend.CONFIG_KEY, "RATIS")),
+ Map.entry(ServiceRole.CONTROLLER, Map.of("ordinal", 1)),
+ Map.entry(ServiceRole.SERVER, Map.of()),
+ Map.entry(ServiceRole.CONTROLLER, Map.of(CoordinationBackend.CONFIG_KEY, "RATIS")),
+ Map.entry(ServiceRole.CONTROLLER, Map.of("ordinal", 2)),
+ Map.entry(ServiceRole.CONTROLLER, Map.of(CoordinationBackend.CONFIG_KEY, "RATIS")));
+ assertTrue(StartServiceManagerCommand.startBootstrapServices(manager, configurations));
+ assertEquals(participants.get(), 2);
+ }
+}
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
new file mode 100644
index 00000000000..09bbd583919
--- /dev/null
+++ b/pinot-tools/src/test/java/org/apache/pinot/tools/service/RatisStartersTest.java
@@ -0,0 +1,339 @@
+/**
+ * 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.tools.service;
+
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.security.KeyStore;
+import java.security.cert.X509Certificate;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.coordination.MetadataRpcClient;
+import org.apache.pinot.common.coordination.NativeMetadataConfig;
+import org.apache.pinot.common.coordination.ServingPermitGuard.ClockReading;
+import org.apache.pinot.common.utils.ServiceStatus;
+import org.apache.pinot.controller.ControllerStarter;
+import org.apache.pinot.controller.coordination.RatisControllerProvider;
+import org.apache.pinot.coordination.ratis.RatisMetadataStore.MembershipClockSettings;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Policy;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Role;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.ResultStatus;
+import org.apache.pinot.coordination.ratis.transport.ControllerAccessPolicy;
+import org.apache.pinot.coordination.ratis.transport.ControllerAccessPolicy.ControllerIdentity;
+import org.apache.pinot.coordination.ratis.transport.ParticipantAccessPolicy;
+import org.apache.pinot.coordination.ratis.transport.ParticipantAccessPolicy.ParticipantIdentity;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.services.ServiceRole;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotEquals;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
+
+
+/// Exercises normal service-manager dispatch with real voters and mTLS. No ZooKeeper is started; configured query
+/// ports remain available because these native starters expose metadata and installation membership only.
+public class RatisStartersTest {
+ private static final String PREFIX = NativeMetadataConfig.PREFIX;
+ private static final String BROKER = "Broker_native";
+ private static final String SERVER = "Server_native";
+
+ @Test(timeOut = 45_000)
+ public void testFailedControllerStartReleasesOwnership() throws Exception {
+ ControllerStarter invalid = new ControllerStarter();
+ assertThrows(IllegalArgumentException.class,
+ () -> invalid.init(new PinotConfiguration(Map.of("pinot.coordination.backend", "RATIS"))));
+ invalid.stop();
+ assertThrows(IllegalStateException.class, invalid::start);
+
+ Path directory = Files.createTempDirectory("pinot-ratis-failed-starter-");
+ String instance = "Controller_ratis_failure-0";
+ ServiceStatus.setServiceStatusCallback(instance,
+ new ServiceStatus.LifecycleServiceStatusCallback(() -> false, () -> false));
+ try {
+ CertificateFiles certificate = certificate(directory, "controller");
+ int[] ports = ports(4);
+ Map<String, Object> config = tls(UUID.randomUUID().toString(), certificate, certificate.certificate());
+ config.put(PREFIX + "clusterId", UUID.randomUUID().toString());
+ config.put(PREFIX + "id", "failure-0");
+ config.put(PREFIX + "voters", "failure-0@127.0.0.1:" + ports[0] + ",failure-1@127.0.0.1:" + ports[1]
+ + ",failure-2@127.0.0.1:" + ports[2]);
+ config.put(PREFIX + "dataDir", directory.resolve("voter").toString());
+ config.put(PREFIX + "bootstrap", true);
+ config.put(PREFIX + "metadata.host", "127.0.0.1");
+ config.put(PREFIX + "metadata.port", ports[3]);
+ config.put(PREFIX + "controller.provider.class", TestProvider.class.getName());
+ config.put(PREFIX + "controller.startup.timeoutMs", 250L);
+ for (int attempt = 0; attempt < 2; attempt++) {
+ ControllerStarter starter = new ControllerStarter();
+ try {
+ starter.init(new PinotConfiguration(config));
+ IOException failure = expectThrows(IOException.class, starter::start);
+ assertTrue(failure.getMessage().contains("Timed out initializing"), failure.getMessage());
+ } finally {
+ starter.stop();
+ }
+ assertEquals(ServiceStatus.getServiceStatus(instance), ServiceStatus.Status.GOOD,
+ "A failed startup must not remove another runtime's callback");
+ config.put(PREFIX + "bootstrap", false);
+ }
+ try (ServerSocket released = new ServerSocket(ports[0])) {
+ assertTrue(released.isBound(), "Failed startup must release its voter listener");
+ }
+ } finally {
+ ServiceStatus.removeServiceStatusCallback(instance);
+ FileUtils.deleteDirectory(directory.toFile());
+ }
+ }
+
+ @Test(timeOut = 120_000)
+ public void testNormalStartersWithoutZooKeeper() throws Exception {
+ Path directory = Files.createTempDirectory("pinot-ratis-starters-");
+ PinotServiceManager manager = new PinotServiceManager("unreachable-zookeeper.invalid:1", "unused", -1);
+ ExecutorService starters = Executors.newFixedThreadPool(3);
+ try {
+ CertificateFiles controller = certificate(directory, "controller");
+ CertificateFiles broker = certificate(directory, "broker");
+ CertificateFiles server = certificate(directory, "server");
+ Path trust = directory.resolve("clients.pem");
+ Files.writeString(trust, Files.readString(controller.certificate()) + Files.readString(broker.certificate())
+ + Files.readString(server.certificate()));
+ int[] ports = ports(9);
+ String history = UUID.randomUUID().toString();
+ String cluster = UUID.randomUUID().toString();
+ String voters = "voter-0@127.0.0.1:" + ports[0] + ",voter-1@127.0.0.1:" + ports[1]
+ + ",voter-2@127.0.0.1:" + ports[2];
+ List<Map<String, Object>> controllers = new ArrayList<>();
+ List<Future<String>> starts = new ArrayList<>();
+ manager.start();
+ for (int i = 0; i < 3; i++) {
+ Map<String, Object> config = tls(history, controller, trust);
+ config.put(PREFIX + "clusterId", cluster);
+ config.put(PREFIX + "id", "voter-" + i);
+ config.put(PREFIX + "voters", voters);
+ config.put(PREFIX + "dataDir", directory.resolve("voter-" + i).toString());
+ config.put(PREFIX + "bootstrap", true);
+ config.put(PREFIX + "metadata.host", "127.0.0.1");
+ config.put(PREFIX + "metadata.port", ports[3 + i]);
+ config.put(PREFIX + "controller.provider.class", TestProvider.class.getName());
+ config.put("test.controllerFingerprint", controller.fingerprint());
+ config.put("test.brokerFingerprint", broker.fingerprint());
+ config.put("test.serverFingerprint", server.fingerprint());
+ controllers.add(config);
+ starts.add(starters.submit(() -> manager.startRole(ServiceRole.CONTROLLER, config)));
+ }
+ for (int i = 0; i < starts.size(); i++) {
+ assertEquals(starts.get(i).get(45, TimeUnit.SECONDS), "Controller_ratis_voter-" + i);
+ assertEquals(ServiceStatus.getServiceStatus("Controller_ratis_voter-" + i), ServiceStatus.Status.GOOD);
+ assertTrue(ServiceStatus.getStatusDescription("Controller_ratis_voter-" + i).contains("HTTP"));
+ }
+
+ String endpoints = "127.0.0.1:" + ports[3] + ",127.0.0.1:" + ports[4] + ",127.0.0.1:" + ports[5];
+ Map<String, Object> brokerConfig = tls(history, broker, controller.certificate());
+ brokerConfig.put(PREFIX + "participantId", BROKER);
+ brokerConfig.put(PREFIX + "metadata.endpoints", endpoints);
+ brokerConfig.put(CommonConstants.Helix.KEY_OF_BROKER_QUERY_PORT, ports[6]);
+ assertEquals(manager.startRole(ServiceRole.BROKER, brokerConfig), BROKER);
+
+ Map<String, Object> serverConfig = tls(history, server, controller.certificate());
+ serverConfig.put(PREFIX + "participantId", SERVER);
+ serverConfig.put(PREFIX + "metadata.endpoints", endpoints);
+ serverConfig.put("pinot.server.instance.id", SERVER);
+ serverConfig.put(CommonConstants.Server.CONFIG_OF_INSTANCE_DATA_DIR, directory.resolve("server-data").toString());
+ serverConfig.put(CommonConstants.Server.CONFIG_OF_INSTANCE_SEGMENT_TAR_DIR,
+ directory.resolve("server-tar").toString());
+ serverConfig.put(CommonConstants.Helix.KEY_OF_SERVER_NETTY_PORT, ports[7]);
+ serverConfig.put(CommonConstants.Server.CONFIG_OF_ADMIN_API_PORT, ports[8]);
+ assertEquals(manager.startRole(ServiceRole.SERVER, serverConfig), SERVER);
+
+ try (MetadataRpcClient metadata = client(ports[4], controller, controller.certificate());
+ MetadataRpcClient brokerClient = client(ports[4], broker, controller.certificate());
+ MetadataRpcClient serverClient = client(ports[4], server, controller.certificate())) {
+ assertEquals(metadata.read(MetadataRpc.ReadRequest.newBuilder().setHistoryId(history)
+ .setRevision(MetadataRpc.Empty.getDefaultInstance()).build()).getStatus(), ResultStatus.OK);
+ MetadataRpc.MembershipReadRequest read = MetadataRpc.MembershipReadRequest.newBuilder()
+ .setHistoryId(history).build();
+ MetadataRpc.ParticipantState brokerState = brokerClient.membershipRead(read).getParticipant();
+ MetadataRpc.ParticipantState serverState = serverClient.membershipRead(read).getParticipant();
+ assertFalse(brokerState.getReady(), "Membership alone must not open broker routing");
+ assertTrue(serverState.getReady(), "The server has installed its complete own-session image");
+ assertEquals(brokerState.getCapabilitiesCount(), 0);
+ assertEquals(serverState.getCapabilitiesCount(), 0);
+ assertEquals(brokerState.getEndpointsCount(), 0);
+ assertEquals(serverState.getEndpointsCount(), 0);
+ assertEquals(ServiceStatus.getServiceStatus(BROKER), ServiceStatus.Status.BAD);
+ assertEquals(ServiceStatus.getServiceStatus(SERVER), ServiceStatus.Status.GOOD);
+ for (int i = 6; i < ports.length; i++) {
+ try (ServerSocket unused = new ServerSocket(ports[i])) {
+ assertTrue(unused.isBound(), "Native query/admin listener must remain closed");
+ }
+ }
+
+ assertTrue(manager.stopPinotInstanceById(SERVER));
+ assertTrue(serverClient.membershipRead(read).getParticipant().getRetired());
+ assertEquals(manager.startRole(ServiceRole.SERVER, serverConfig), SERVER);
+ MetadataRpc.ParticipantToken replacement = serverClient.membershipRead(read).getParticipant().getParticipant();
+ assertEquals(replacement.getRegistrationEpoch(), serverState.getParticipant().getRegistrationEpoch() + 1);
+ assertNotEquals(replacement.getIncarnation(), serverState.getParticipant().getIncarnation());
+
+ assertTrue(manager.stopPinotInstanceById("Controller_ratis_voter-0"));
+ controllers.getFirst().put(PREFIX + "bootstrap", false);
+ controllers.getFirst().put("test.bootstrapPermitMillis", 2000L);
+ assertEquals(manager.startRole(ServiceRole.CONTROLLER, controllers.getFirst()), "Controller_ratis_voter-0");
+ assertEquals(metadata.membershipRead(read).getControl().getPolicy().getPermitLifetimeMillis(), 1000L,
+ "Restart must retain committed policy instead of resetting it from bootstrap configuration");
+ }
+ } finally {
+ starters.shutdownNow();
+ starters.awaitTermination(10, TimeUnit.SECONDS);
+ // Participants retire while a quorum remains available; the manager's arbitrary iteration order is unsuitable.
+ manager.stopPinotInstanceById(BROKER);
+ manager.stopPinotInstanceById(SERVER);
+ manager.stopAll();
+ FileUtils.deleteDirectory(directory.toFile());
+ }
+ }
+
+ private static Map<String, Object> tls(String history, CertificateFiles certificate, Path trust) {
+ Map<String, Object> config = new HashMap<>();
+ config.put("pinot.coordination.backend", "RATIS");
+ config.put(PREFIX + "historyId", history);
+ config.put(PREFIX + "tls.certChain", certificate.certificate().toString());
+ config.put(PREFIX + "tls.privateKey", certificate.key().toString());
+ config.put(PREFIX + "tls.trustCerts", trust.toString());
+ return config;
+ }
+
+ private static MetadataRpcClient client(int port, CertificateFiles certificate, Path trust) throws IOException {
+ return new MetadataRpcClient("127.0.0.1", port, certificate.certificate(), certificate.key(), trust, 5000);
+ }
+
+ private static int[] ports(int count) throws IOException {
+ List<ServerSocket> reservations = new ArrayList<>();
+ int[] ports = new int[count];
+ try {
+ for (int i = 0; i < count; i++) {
+ ServerSocket socket = new ServerSocket(0);
+ reservations.add(socket);
+ ports[i] = socket.getLocalPort();
+ }
+ return ports;
+ } finally {
+ for (ServerSocket socket : reservations) {
+ socket.close();
+ }
+ }
+ }
+
+ private static CertificateFiles certificate(Path directory, String name) throws Exception {
+ Path keyStoreFile = directory.resolve(name + ".p12");
+ Path certificateFile = directory.resolve(name + ".pem");
+ Path keyFile = directory.resolve(name + "-key.pem");
+ String keytool = Path.of(System.getProperty("java.home"), "bin", "keytool").toString();
+ Process process = new ProcessBuilder(keytool, "-genkeypair", "-alias", "test", "-keyalg", "EC", "-groupname",
+ "secp256r1", "-dname", "CN=" + name, "-ext", "SAN=DNS:localhost,IP:127.0.0.1", "-ext",
+ "EKU=serverAuth,clientAuth", "-validity", "2", "-storetype", "PKCS12", "-keystore", keyStoreFile.toString(),
+ "-storepass", "test-password", "-keypass", "test-password", "-noprompt").redirectErrorStream(true).start();
+ if (!process.waitFor(30, TimeUnit.SECONDS)) {
+ process.destroyForcibly();
+ throw new IOException("Timed out generating TLS test certificate");
+ }
+ assertEquals(process.exitValue(), 0, new String(process.getInputStream().readAllBytes(), StandardCharsets.UTF_8));
+ KeyStore keyStore = KeyStore.getInstance("PKCS12");
+ try (var input = Files.newInputStream(keyStoreFile)) {
+ keyStore.load(input, "test-password".toCharArray());
+ }
+ X509Certificate certificate = (X509Certificate) keyStore.getCertificate("test");
+ Files.writeString(certificateFile, pem("CERTIFICATE", certificate.getEncoded()));
+ Files.writeString(keyFile, pem("PRIVATE KEY", keyStore.getKey("test", "test-password".toCharArray()).getEncoded()));
+ return new CertificateFiles(certificateFile, keyFile, ControllerAccessPolicy.fingerprint(certificate));
+ }
+
+ private static String pem(String kind, byte[] bytes) {
+ return "-----BEGIN " + kind + "-----\n" + Base64.getMimeEncoder(64, new byte[]{'\n'}).encodeToString(bytes)
+ + "\n-----END " + kind + "-----\n";
+ }
+
+ private record CertificateFiles(Path certificate, Path key, String fingerprint) {
+ }
+
+ /// Test-only clock assumptions and exact certificate bindings; this is not a deployment qualification provider.
+ public static final class TestProvider implements RatisControllerProvider {
+ private String _controller;
+ private String _broker;
+ private String _server;
+ private long _permitMillis;
+
+ @Override
+ public void init(PinotConfiguration config) {
+ _controller = config.getProperty("test.controllerFingerprint");
+ _broker = config.getProperty("test.brokerFingerprint");
+ _server = config.getProperty("test.serverFingerprint");
+ _permitMillis = config.getProperty("test.bootstrapPermitMillis", 1000L);
+ }
+
+ @Override
+ public MembershipClockSettings membershipClockSettings() {
+ return new MembershipClockSettings(() -> new ClockReading(System.nanoTime(), 0, true), 1_000_000, 0);
+ }
+
+ @Override
+ public ControllerAccessPolicy controllerAccessPolicy() {
+ return fingerprint -> fingerprint.equals(_controller) ? new ControllerIdentity("test-controller", "test-scope")
+ : null;
+ }
+
+ @Override
+ public ParticipantAccessPolicy participantAccessPolicy() {
+ return fingerprint -> {
+ if (fingerprint.equals(_broker)) {
+ return new ParticipantIdentity(BROKER, Role.BROKER, Set.of());
+ }
+ if (fingerprint.equals(_server)) {
+ return new ParticipantIdentity(SERVER, Role.SERVER, Set.of());
+ }
+ return fingerprint.equals(_controller) ? new ParticipantIdentity("Controller_test", Role.CONTROLLER, Set.of())
+ : null;
+ };
+ }
+
+ @Override
+ public Policy bootstrapPolicy() {
+ return new Policy(1, 100, 1000, _permitMillis, 1000, 100_000);
+ }
+ }
+}
diff --git a/pinot-tools/src/test/java/org/apache/pinot/tools/utils/PinotConfigUtilsTest.java b/pinot-tools/src/test/java/org/apache/pinot/tools/utils/PinotConfigUtilsTest.java
index 8a05c544927..b6242656061 100644
--- a/pinot-tools/src/test/java/org/apache/pinot/tools/utils/PinotConfigUtilsTest.java
+++ b/pinot-tools/src/test/java/org/apache/pinot/tools/utils/PinotConfigUtilsTest.java
@@ -26,6 +26,7 @@ import java.nio.charset.StandardCharsets;
import java.util.Map;
import org.apache.commons.configuration2.ex.ConfigurationException;
import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.coordination.CoordinationBackend;
import org.apache.pinot.controller.ControllerConf;
import org.testng.annotations.Test;
@@ -34,6 +35,25 @@ import static org.testng.Assert.*;
public class PinotConfigUtilsTest {
+ @Test
+ public void testNativeControllerConfigDoesNotRequireHelixOrHttpSettings()
+ throws Exception {
+ File file = File.createTempFile("native-controller", ".properties");
+ try {
+ FileUtils.writeStringToFile(file, CoordinationBackend.CONFIG_KEY + "=RATIS\n", StandardCharsets.UTF_8);
+ Map<String, Object> properties = PinotConfigUtils.generateControllerConf(file.getAbsolutePath());
+ ControllerConf config = new ControllerConf(properties);
+ assertEquals(CoordinationBackend.from(config), CoordinationBackend.RATIS);
+ assertThrows(IllegalStateException.class, config::getZkStr);
+ assertNull(config.getHelixClusterName());
+ assertThrows(ConfigurationException.class, () -> PinotConfigUtils.validateControllerConfig(new ControllerConf()));
+ config.setProperty(CoordinationBackend.CONFIG_KEY, "unknown");
+ assertThrows(IllegalArgumentException.class, () -> PinotConfigUtils.validateControllerConfig(config));
+ } finally {
+ FileUtils.deleteQuietly(file);
+ }
+ }
+
@Test
public void testGenerateControllerConf()
throws SocketException, UnknownHostException {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org