Jackie-Jiang commented on code in PR #19178:
URL: https://github.com/apache/pinot/pull/19178#discussion_r3808465152
##########
pinot-server/src/main/java/org/apache/pinot/server/api/resources/HealthCheckResource.java:
##########
@@ -55,6 +56,10 @@ public class HealthCheckResource {
@Inject
private AtomicBoolean _shutDownInProgress;
+ @Inject
+ @Named(AdminApiApplication.SERVER_READY_TO_SERVE_QUERIES)
+ private BooleanSupplier _isServerReadyToServeQueries;
Review Comment:
(minor) Suggest renaming it for supplier
```suggestion
private BooleanSupplier _serverReadyToServeQueries;
```
##########
pinot-server/src/main/java/org/apache/pinot/server/api/resources/HealthCheckResource.java:
##########
@@ -111,6 +116,11 @@ public String checkReadiness() {
throw new WebApplicationException(errMessage,
Response.status(Response.Status.SERVICE_UNAVAILABLE).entity(errMessage).build());
}
+ if (!_isServerReadyToServeQueries.getAsBoolean()) {
+ String errMessage = "Server is not ready to serve queries";
+ throw new WebApplicationException(errMessage,
+
Response.status(Response.Status.SERVICE_UNAVAILABLE).entity(errMessage).build());
+ }
Status status = ServiceStatus.getServiceStatus(_instanceId);
Review Comment:
The service status check is part of the server serving query check. We can
probably skip the following part when `_serverReadyToServeQueries` is provided
##########
pinot-server/src/main/java/org/apache/pinot/server/api/AdminApiApplication.java:
##########
@@ -63,8 +65,8 @@ public class AdminApiApplication extends ResourceConfig {
public AdminApiApplication(ServerInstance instance, AccessControlFactory
accessControlFactory,
- ServerReloadJobStatusCache reloadJobStatusCache,
- PinotConfiguration serverConf) {
+ ServerReloadJobStatusCache reloadJobStatusCache, PinotConfiguration
serverConf,
+ BooleanSupplier isServerReadyToServeQueries) {
Review Comment:
(minor)
```suggestion
BooleanSupplier serverReadyToServeQueries) {
```
##########
pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BrokerRoutingReadyChecker.java:
##########
@@ -0,0 +1,213 @@
+/**
+ * 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 com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.io.IOException;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.LongSupplier;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+import org.apache.helix.HelixAdmin;
+import org.apache.helix.HelixManager;
+import org.apache.helix.model.ExternalView;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.pinot.common.auth.NullAuthProvider;
+import org.apache.pinot.common.utils.SimpleHttpResponse;
+import org.apache.pinot.common.utils.config.InstanceUtils;
+import org.apache.pinot.common.utils.helix.HelixHelper;
+import org.apache.pinot.common.utils.http.HttpClient;
+import org.apache.pinot.spi.auth.AuthProvider;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Checks broker routing state in the background while a server starts. The
checker remains false until every online
+/// broker reports the server as routable, or until the configured timeout
when fail-open behavior is enabled. Health
+/// endpoints only read the cached result. Brokers must return the
routing-specific response, so an older broker's
+/// normal health response cannot be mistaken for an acknowledgement. All
mutable state is confined to the scheduler
+/// thread except for the volatile ready flag read by health-check threads.
+public class BrokerRoutingReadyChecker implements AutoCloseable {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(BrokerRoutingReadyChecker.class);
+ private static final long CHECK_INTERVAL_MS = 1_000L;
+ private static final long CHECK_TIMEOUT_MS = 5_000L;
+
+ private final String _serverInstanceId;
+ private final Supplier<Set<String>> _onlineBrokersSupplier;
+ private final Predicate<Set<String>> _allBrokersReady;
+ private final ScheduledExecutorService _scheduler;
+ private final ExecutorService _requestExecutor;
+ private final LongSupplier _currentTimeMs;
+ private final long _deadlineMs;
+ private final boolean _failOpen;
+ private final AuthProvider _authProvider;
+ private final RoutingStatusClient _routingStatusClient;
+ private volatile boolean _ready;
+ private boolean _timeoutLogged;
+
+ public BrokerRoutingReadyChecker(HelixManager helixManager, String
serverInstanceId, long timeoutMs,
+ boolean failOpen, AuthProvider authProvider) {
+ this(helixManager, serverInstanceId, timeoutMs, failOpen, authProvider,
+ (uri, provider) -> HttpClient.getInstance().sendGetRequest(uri, null,
provider));
+ }
+
+ @VisibleForTesting
+ BrokerRoutingReadyChecker(HelixManager helixManager, String
serverInstanceId, long timeoutMs,
+ boolean failOpen, AuthProvider authProvider, RoutingStatusClient
routingStatusClient) {
+ _serverInstanceId = serverInstanceId;
+ HelixAdmin helixAdmin = helixManager.getClusterManagmentTool();
+ String clusterName = helixManager.getClusterName();
+ _onlineBrokersSupplier = () -> {
+ ExternalView brokerResource =
helixAdmin.getResourceExternalView(clusterName,
+ CommonConstants.Helix.BROKER_RESOURCE_INSTANCE);
+ return
Set.copyOf(HelixHelper.getOnlineInstanceFromExternalView(brokerResource));
+ };
+ _requestExecutor = Executors.newCachedThreadPool(
+ new
ThreadFactoryBuilder().setNameFormat("broker-routing-ready-request-%d").setDaemon(true).build());
+ _allBrokersReady = brokers -> checkAllBrokers(helixAdmin, clusterName,
brokers);
+ _scheduler = Executors.newSingleThreadScheduledExecutor(
+ new
ThreadFactoryBuilder().setNameFormat("broker-routing-ready-checker").setDaemon(true).build());
+ _currentTimeMs = System::currentTimeMillis;
+ _deadlineMs = _currentTimeMs.getAsLong() + timeoutMs;
+ _failOpen = failOpen;
+ _authProvider = authProvider;
+ _routingStatusClient = routingStatusClient;
+ }
+
+ @VisibleForTesting
+ BrokerRoutingReadyChecker(String serverInstanceId, Supplier<Set<String>>
onlineBrokersSupplier,
+ Predicate<Set<String>> allBrokersReady) {
+ this(serverInstanceId, onlineBrokersSupplier, allBrokersReady,
Long.MAX_VALUE, false, () -> 0L);
+ }
+
+ @VisibleForTesting
+ BrokerRoutingReadyChecker(String serverInstanceId, Supplier<Set<String>>
onlineBrokersSupplier,
Review Comment:
(minor) Make one single main constructor, and others calling it. Currently
there are 2 main constructors, which can reduce the test coverage
##########
pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BaseServerStarter.java:
##########
@@ -911,6 +914,13 @@ public void start()
_serverInstance.startQueryServer();
_helixAdmin.setConfig(_instanceConfigScope,
Map.of(Helix.IS_SHUTDOWN_IN_PROGRESS, Boolean.toString(false)));
+ if
(_serverConf.getProperty(Server.CONFIG_OF_STARTUP_ENABLE_BROKER_ROUTING_CHECK,
+ Server.DEFAULT_STARTUP_ENABLE_BROKER_ROUTING_CHECK)) {
+ _brokerRoutingReadyChecker = createBrokerRoutingReadyChecker();
+ _brokerRoutingReadyChecker.start();
Review Comment:
This doesn't need to be async. A more light-weight one would be to do
synchronized check here
##########
pinot-server/src/main/java/org/apache/pinot/server/api/resources/HealthCheckResource.java:
##########
@@ -55,6 +56,10 @@ public class HealthCheckResource {
@Inject
private AtomicBoolean _shutDownInProgress;
Review Comment:
Not introduced in this PR, but since the scope overlaps, shall we change
this to also use named injection and change it to a `BooleanSupplier`? It is
very fragile to have unnamed injection of an `AtomicBoolean`
##########
pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BrokerRoutingReadyChecker.java:
##########
@@ -0,0 +1,213 @@
+/**
+ * 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 com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.io.IOException;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.LongSupplier;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+import org.apache.helix.HelixAdmin;
+import org.apache.helix.HelixManager;
+import org.apache.helix.model.ExternalView;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.pinot.common.auth.NullAuthProvider;
+import org.apache.pinot.common.utils.SimpleHttpResponse;
+import org.apache.pinot.common.utils.config.InstanceUtils;
+import org.apache.pinot.common.utils.helix.HelixHelper;
+import org.apache.pinot.common.utils.http.HttpClient;
+import org.apache.pinot.spi.auth.AuthProvider;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Checks broker routing state in the background while a server starts. The
checker remains false until every online
+/// broker reports the server as routable, or until the configured timeout
when fail-open behavior is enabled. Health
+/// endpoints only read the cached result. Brokers must return the
routing-specific response, so an older broker's
+/// normal health response cannot be mistaken for an acknowledgement. All
mutable state is confined to the scheduler
+/// thread except for the volatile ready flag read by health-check threads.
+public class BrokerRoutingReadyChecker implements AutoCloseable {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(BrokerRoutingReadyChecker.class);
+ private static final long CHECK_INTERVAL_MS = 1_000L;
+ private static final long CHECK_TIMEOUT_MS = 5_000L;
+
+ private final String _serverInstanceId;
+ private final Supplier<Set<String>> _onlineBrokersSupplier;
+ private final Predicate<Set<String>> _allBrokersReady;
+ private final ScheduledExecutorService _scheduler;
+ private final ExecutorService _requestExecutor;
+ private final LongSupplier _currentTimeMs;
+ private final long _deadlineMs;
+ private final boolean _failOpen;
+ private final AuthProvider _authProvider;
+ private final RoutingStatusClient _routingStatusClient;
+ private volatile boolean _ready;
+ private boolean _timeoutLogged;
+
+ public BrokerRoutingReadyChecker(HelixManager helixManager, String
serverInstanceId, long timeoutMs,
+ boolean failOpen, AuthProvider authProvider) {
+ this(helixManager, serverInstanceId, timeoutMs, failOpen, authProvider,
+ (uri, provider) -> HttpClient.getInstance().sendGetRequest(uri, null,
provider));
+ }
+
+ @VisibleForTesting
+ BrokerRoutingReadyChecker(HelixManager helixManager, String
serverInstanceId, long timeoutMs,
+ boolean failOpen, AuthProvider authProvider, RoutingStatusClient
routingStatusClient) {
+ _serverInstanceId = serverInstanceId;
+ HelixAdmin helixAdmin = helixManager.getClusterManagmentTool();
+ String clusterName = helixManager.getClusterName();
+ _onlineBrokersSupplier = () -> {
+ ExternalView brokerResource =
helixAdmin.getResourceExternalView(clusterName,
+ CommonConstants.Helix.BROKER_RESOURCE_INSTANCE);
+ return
Set.copyOf(HelixHelper.getOnlineInstanceFromExternalView(brokerResource));
Review Comment:
(minor) No need to copy
##########
pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BrokerRoutingReadyChecker.java:
##########
@@ -0,0 +1,213 @@
+/**
+ * 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 com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.io.IOException;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.LongSupplier;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+import org.apache.helix.HelixAdmin;
+import org.apache.helix.HelixManager;
+import org.apache.helix.model.ExternalView;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.pinot.common.auth.NullAuthProvider;
+import org.apache.pinot.common.utils.SimpleHttpResponse;
+import org.apache.pinot.common.utils.config.InstanceUtils;
+import org.apache.pinot.common.utils.helix.HelixHelper;
+import org.apache.pinot.common.utils.http.HttpClient;
+import org.apache.pinot.spi.auth.AuthProvider;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Checks broker routing state in the background while a server starts. The
checker remains false until every online
+/// broker reports the server as routable, or until the configured timeout
when fail-open behavior is enabled. Health
+/// endpoints only read the cached result. Brokers must return the
routing-specific response, so an older broker's
+/// normal health response cannot be mistaken for an acknowledgement. All
mutable state is confined to the scheduler
+/// thread except for the volatile ready flag read by health-check threads.
+public class BrokerRoutingReadyChecker implements AutoCloseable {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(BrokerRoutingReadyChecker.class);
+ private static final long CHECK_INTERVAL_MS = 1_000L;
+ private static final long CHECK_TIMEOUT_MS = 5_000L;
+
+ private final String _serverInstanceId;
+ private final Supplier<Set<String>> _onlineBrokersSupplier;
+ private final Predicate<Set<String>> _allBrokersReady;
+ private final ScheduledExecutorService _scheduler;
+ private final ExecutorService _requestExecutor;
+ private final LongSupplier _currentTimeMs;
+ private final long _deadlineMs;
+ private final boolean _failOpen;
+ private final AuthProvider _authProvider;
+ private final RoutingStatusClient _routingStatusClient;
+ private volatile boolean _ready;
+ private boolean _timeoutLogged;
+
+ public BrokerRoutingReadyChecker(HelixManager helixManager, String
serverInstanceId, long timeoutMs,
Review Comment:
(minor) `serverInstanceId` is available through
`helixManager.getInstanceName()`
##########
pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BaseServerStarter.java:
##########
@@ -182,6 +184,7 @@ public abstract class BaseServerStarter implements
ServiceStartable {
protected QueryKillingManager _queryKillingManager;
protected DefaultClusterConfigChangeHandler _clusterConfigChangeHandler;
protected volatile boolean _isServerReadyToServeQueries = false;
+ protected volatile BrokerRoutingReadyChecker _brokerRoutingReadyChecker;
Review Comment:
This doesn't need to be `volatile`
##########
pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java:
##########
@@ -1589,6 +1590,22 @@ public static class Server {
public static final String
CONFIG_OF_STARTUP_SERVICE_STATUS_CHECK_INTERVAL_MS =
"pinot.server.startup.serviceStatusCheckIntervalMs";
public static final long DEFAULT_STARTUP_SERVICE_STATUS_CHECK_INTERVAL_MS
= 10_000L;
+ // Startup: wait for brokers to add the server to routing before reporting
the server as ready. Disabled by default
+ // for the initial rollout so mixed-version clusters retain the existing
readiness behavior.
+ public static final String CONFIG_OF_STARTUP_ENABLE_BROKER_ROUTING_CHECK =
+ "pinot.server.startup.enableBrokerRoutingCheck";
+ public static final boolean DEFAULT_STARTUP_ENABLE_BROKER_ROUTING_CHECK =
false;
+ public static final String
CONFIG_OF_STARTUP_BROKER_ROUTING_CHECK_TIMEOUT_MS =
+ "pinot.server.startup.brokerRoutingCheckTimeoutMs";
+ public static final long DEFAULT_STARTUP_BROKER_ROUTING_CHECK_TIMEOUT_MS =
60_000L;
+ // When true, report ready after the timeout even if one or more brokers
have not confirmed routing. When false,
+ // keep reporting unready and continue checking until the brokers recover.
+ public static final String
CONFIG_OF_STARTUP_BROKER_ROUTING_CHECK_FAIL_OPEN =
+ "pinot.server.startup.brokerRoutingCheckFailOpen";
+ public static final boolean DEFAULT_STARTUP_BROKER_ROUTING_CHECK_FAIL_OPEN
= true;
Review Comment:
Given the feature is `off` by default, do we want it to fail-open by default?
When people enable this feature, I'd expect them wanting more strict check
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
