Jackie-Jiang commented on code in PR #19178: URL: https://github.com/apache/pinot/pull/19178#discussion_r4251524516
########## pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BrokerRoutingReadyChecker.java: ########## @@ -0,0 +1,371 @@ +/** + * 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.HashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.BooleanSupplier; +import java.util.function.LongSupplier; +import java.util.function.Predicate; +import java.util.function.Supplier; +import javax.annotation.Nullable; +import org.apache.hc.core5.http.Header; +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.AuthProviderUtils; +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.common.utils.http.HttpClientConfig; +import org.apache.pinot.common.utils.tls.TlsUtils; +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 while a server starts. A single background thread polls every online broker until all +/// of them report the server as routable, then caches success. Readiness requests only read the cached state and never +/// perform network I/O. Broker requests are sequential and have bounded connect, pool-checkout and response timeouts, +/// so an unavailable broker cannot create unbounded tasks or 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 REQUEST_TIMEOUT_MS = 5_000L; + private static final int MAX_RESPONSE_LENGTH = 1_024; + + private final String _serverInstanceId; + private final Supplier<Set<String>> _onlineBrokersSupplier; + private final BrokersReadyEvaluator _allBrokersReady; + @Nullable + private final ScheduledExecutorService _checkExecutor; + private final RoutingStatusClient _routingStatusClient; + private final LongSupplier _currentTimeMs; + private final long _deadlineMs; + private final boolean _failOpen; + private final AuthProvider _authProvider; + private final AtomicReference<State> _state; + private boolean _timeoutLogged; + + public BrokerRoutingReadyChecker(HelixManager helixManager, long timeoutMs, boolean failOpen, + AuthProvider authProvider) { + this(createProductionContext(helixManager, authProvider), timeoutMs, failOpen, authProvider); + } + + @VisibleForTesting + BrokerRoutingReadyChecker(HelixManager helixManager, long timeoutMs, boolean failOpen, AuthProvider authProvider, + RoutingStatusClient routingStatusClient) { + this(helixManager, timeoutMs, failOpen, authProvider, routingStatusClient, System::currentTimeMillis); + } + + @VisibleForTesting + BrokerRoutingReadyChecker(HelixManager helixManager, long timeoutMs, boolean failOpen, AuthProvider authProvider, + RoutingStatusClient routingStatusClient, LongSupplier currentTimeMs) { + this(createContext(helixManager, routingStatusClient, null, authProvider), timeoutMs, failOpen, currentTimeMs, + authProvider); + } + + private BrokerRoutingReadyChecker(ProductionContext context, long timeoutMs, boolean failOpen, + AuthProvider authProvider) { + this(context, timeoutMs, failOpen, System::currentTimeMillis, authProvider); + } + + private BrokerRoutingReadyChecker(ProductionContext context, long timeoutMs, boolean failOpen, + LongSupplier currentTimeMs, AuthProvider authProvider) { + this(context._serverInstanceId, context._onlineBrokersSupplier, context._allBrokersReady, + context._checkExecutor, context._routingStatusClient, timeoutMs, failOpen, currentTimeMs, + authProvider); + } + + @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, + Predicate<Set<String>> allBrokersReady, long timeoutMs, boolean failOpen, LongSupplier currentTimeMs) { + this(serverInstanceId, onlineBrokersSupplier, (brokers, shouldStop) -> allBrokersReady.test(brokers), null, + RoutingStatusClient.NOOP, timeoutMs, failOpen, currentTimeMs, new NullAuthProvider()); + } + + private BrokerRoutingReadyChecker(String serverInstanceId, Supplier<Set<String>> onlineBrokersSupplier, + BrokersReadyEvaluator allBrokersReady, @Nullable ScheduledExecutorService checkExecutor, + RoutingStatusClient routingStatusClient, long timeoutMs, boolean failOpen, LongSupplier currentTimeMs, + AuthProvider authProvider) { + _serverInstanceId = serverInstanceId; + _onlineBrokersSupplier = onlineBrokersSupplier; + _allBrokersReady = allBrokersReady; + _checkExecutor = checkExecutor; + _routingStatusClient = routingStatusClient; + _currentTimeMs = currentTimeMs; + long nowMs = currentTimeMs.getAsLong(); + _deadlineMs = timeoutMs >= Long.MAX_VALUE - nowMs ? Long.MAX_VALUE : nowMs + Math.max(timeoutMs, 0L); + _failOpen = failOpen; + _authProvider = authProvider; + _state = new AtomicReference<>(State.CHECKING); + if (_checkExecutor != null) { + _checkExecutor.scheduleWithFixedDelay(this::check, 0L, CHECK_INTERVAL_MS, TimeUnit.MILLISECONDS); + } + } + + public boolean isReady() { + transitionToReadyOnFailOpenTimeout(); + return _state.get() == State.READY; + } + + @VisibleForTesting + AuthProvider getAuthProvider() { + return _authProvider; + } + + @VisibleForTesting + synchronized void check() { + if (_state.get() != State.CHECKING || transitionToReadyOnFailOpenTimeout()) { + return; + } + try { + Set<String> onlineBrokers = _onlineBrokersSupplier.get(); + if (!onlineBrokers.isEmpty() && _allBrokersReady.test(onlineBrokers, this::shouldStopChecking) + && _state.get() == State.CHECKING + // Do not mark the server ready if broker membership changed while acknowledgements were collected. + && onlineBrokers.equals(_onlineBrokersSupplier.get())) { + if (_state.compareAndSet(State.CHECKING, State.READY)) { + LOGGER.info("All online brokers report server {} as routable: {}", _serverInstanceId, onlineBrokers); + stopChecking(); + } + return; + } + } catch (Exception e) { + LOGGER.debug("Failed to check broker routing readiness for server {}", _serverInstanceId, e); + } + + if (transitionToReadyOnFailOpenTimeout()) { + return; + } + if (_currentTimeMs.getAsLong() >= _deadlineMs) { + if (!_timeoutLogged) { + LOGGER.warn("Timed out waiting for all online brokers to report server {} as routable; failOpen={}", + _serverInstanceId, _failOpen); + _timeoutLogged = true; + } + } + } + + private boolean shouldStopChecking() { + return _state.get() != State.CHECKING || Thread.currentThread().isInterrupted() + || transitionToReadyOnFailOpenTimeout(); + } + + private boolean transitionToReadyOnFailOpenTimeout() { + if (!_failOpen || _currentTimeMs.getAsLong() < _deadlineMs) { + return false; + } + if (_state.compareAndSet(State.CHECKING, State.READY)) { + LOGGER.warn("Timed out waiting for all online brokers to report server {} as routable; failOpen=true", + _serverInstanceId); + stopChecking(); + } + return _state.get() != State.CHECKING; + } + + private void stopChecking() { + if (_checkExecutor != null) { + _checkExecutor.shutdown(); + } + } + + private static ProductionContext createProductionContext(HelixManager helixManager, AuthProvider authProvider) { + HttpClientConfig httpClientConfig = HttpClientConfig.newBuilder() + .withMaxConns(1) + .withMaxConnsPerRoute(1) + .withConnectionTimeoutMs((int) REQUEST_TIMEOUT_MS) + .withFollowRedirects(false) + .build(); + RoutingStatusClient routingStatusClient = new SecureRoutingStatusClient( + new HttpRoutingStatusClient(new HttpClient(httpClientConfig, TlsUtils.getSslContext(), true))); + ScheduledExecutorService checkExecutor = Executors.newSingleThreadScheduledExecutor( + new ThreadFactoryBuilder().setNameFormat("broker-routing-ready-check-%d").setDaemon(true).build()); + return createContext(helixManager, routingStatusClient, checkExecutor, authProvider); + } + + private static ProductionContext createContext(HelixManager helixManager, RoutingStatusClient routingStatusClient, + @Nullable ScheduledExecutorService checkExecutor, AuthProvider authProvider) { + String serverInstanceId = helixManager.getInstanceName(); + HelixAdmin helixAdmin = helixManager.getClusterManagmentTool(); + String clusterName = helixManager.getClusterName(); + Supplier<Set<String>> onlineBrokersSupplier = () -> { + ExternalView brokerResource = helixAdmin.getResourceExternalView(clusterName, + CommonConstants.Helix.BROKER_RESOURCE_INSTANCE); + return HelixHelper.getOnlineInstanceFromExternalView(brokerResource); Review Comment: **[P2] Discover live brokers independently of table assignments** The brokerResource ExternalView represents table assignments, so a live broker with no assigned tables does not appear in this set. HelixSetupUtils creates brokerResource without partitions, and broker registration only adds the broker to existing table partitions. In a fresh cluster with healthy upgraded brokers but no tables, enabling this gate with the default failOpen=false leaves servers unready indefinitely: check() rejects the empty set, although the existing local service-status check accepts zero table resources. Provisioning that waits for server readiness before creating tables cannot proceed. Discover live broker instances independently of table assignments, or explicitly distinguish the no-table case from missing/failed discovery while retaining the membership-stability check. A discovery-level regression should include a live broker and an empty brokerResource; the current empty-supplier test does not distinguish this from having no brokers. ########## pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java: ########## @@ -1376,6 +1569,14 @@ public Set<String> getServingInstances(String tableNameWithType) { return routingEntry._instanceSelector.getServingInstances(); } + /// Returns whether the broker sees the server as routable. + public boolean isServerRoutable(String instanceId) { + // Read the map first. A new entry is inserted only after the server is marked pending, and removing the pending + // marker publishes all routing-entry updates that precede it. + return _routableServerInstanceMap.containsKey(instanceId) + && !_serversPendingRoutingUpdate.containsKey(instanceId); Review Comment: **[P2] Read readiness from a consistent routing state** These two reads can acknowledge a server that remains unready throughout an exclusion. Start with S present in the routable map but still pending after a failed table update. The reader observes map membership; a concurrent exclusion then publishes the map without S and clears its pending marker; the reader observes no marker and returns true. A failing MultiStageReplicaGroupSelector instance-partitions fetch provides a real path to the retained pending state. The routing endpoint can return 200 and the server checker can permanently cache that false acknowledgement. Disabling a pending server has the same shape. Use a consistent readiness snapshot, or verify that the captured routable map is still current after checking the pending marker. Holding the global read lock across both checks is another correctness option, with request-latency tradeoffs. Merely reversing the reads is insufficient. Add a controlled interleaving test that pauses between the reads and excludes a server with a retained pending update. ########## pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java: ########## @@ -531,6 +570,163 @@ private void processInstanceConfigChangeInternal() { newDisabledServers, _excludedServers); } + /// Initializes an empty publication marker with the tables whose primary selector can route to the server. A + /// nonempty marker belongs to an earlier pass and retains its completed acknowledgements across retries. Must run + /// before publishing the routable map so the lock-free readiness check observes the pending marker first. + @GuardedBy("_globalLock.writeLock()") + private void markServersPendingRelevantTableUpdates(List<String> servers) { + for (String server : servers) { + if (!_serversPendingRoutingUpdate.get(server).isEmpty()) { + continue; + } + Set<String> relevantTables = new HashSet<>(); + for (RoutingEntry routingEntry : _routingEntryMap.values()) { + if (routingEntry.isServerAssigned(server)) { + relevantTables.add(routingEntry.getTableNameWithType()); + } + } + _serversPendingRoutingUpdate.put(server, relevantTables); + } + } + + /// Applies the current instance change plus any retained failed work independently to each table. Successful tables + /// acknowledge only the servers they can route; failed tables retain their complete change set for the retry. + @GuardedBy("_globalLock.writeLock()") + private void updateRoutingEntriesOnInstancesChange(List<String> changedServers) { + Map<String, List<String>> pendingServersByTable = new HashMap<>(); + _serversPendingRoutingUpdate.forEach((server, tables) -> { + for (String table : tables) { + pendingServersByTable.computeIfAbsent(table, ignored -> new ArrayList<>()).add(server); + } + }); + for (RoutingEntry routingEntry : _routingEntryMap.values()) { + String tableNameWithType = routingEntry.getTableNameWithType(); + List<String> tableChangedServers = new ArrayList<>(changedServers); + List<String> pendingAcknowledgements = pendingServersByTable.get(tableNameWithType); + if (pendingAcknowledgements != null) { + for (String pendingServer : pendingAcknowledgements) { + if (!tableChangedServers.contains(pendingServer)) { Review Comment: **[P3] Use a set to combine the per-table server lists** Optional simplification: combine changedServers, pendingAcknowledgements, and the retained failed servers in a set, then convert once to the List required by the selector API. These repeated ArrayList.contains() calls add quadratic membership work per table in the worst case during bulk enablement, when changedServers and pendingAcknowledgements contain the same large server set. The work runs under the global write lock. A set union also removes the duplicated membership/add loops. This is an update-path improvement; no measured performance regression is claimed. ########## pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BrokerRoutingReadyChecker.java: ########## @@ -0,0 +1,371 @@ +/** + * 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.HashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.BooleanSupplier; +import java.util.function.LongSupplier; +import java.util.function.Predicate; +import java.util.function.Supplier; +import javax.annotation.Nullable; +import org.apache.hc.core5.http.Header; +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.AuthProviderUtils; +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.common.utils.http.HttpClientConfig; +import org.apache.pinot.common.utils.tls.TlsUtils; +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 while a server starts. A single background thread polls every online broker until all +/// of them report the server as routable, then caches success. Readiness requests only read the cached state and never +/// perform network I/O. Broker requests are sequential and have bounded connect, pool-checkout and response timeouts, +/// so an unavailable broker cannot create unbounded tasks or 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 REQUEST_TIMEOUT_MS = 5_000L; + private static final int MAX_RESPONSE_LENGTH = 1_024; + + private final String _serverInstanceId; + private final Supplier<Set<String>> _onlineBrokersSupplier; + private final BrokersReadyEvaluator _allBrokersReady; + @Nullable + private final ScheduledExecutorService _checkExecutor; + private final RoutingStatusClient _routingStatusClient; + private final LongSupplier _currentTimeMs; + private final long _deadlineMs; + private final boolean _failOpen; + private final AuthProvider _authProvider; + private final AtomicReference<State> _state; + private boolean _timeoutLogged; + + public BrokerRoutingReadyChecker(HelixManager helixManager, long timeoutMs, boolean failOpen, + AuthProvider authProvider) { + this(createProductionContext(helixManager, authProvider), timeoutMs, failOpen, authProvider); + } + + @VisibleForTesting + BrokerRoutingReadyChecker(HelixManager helixManager, long timeoutMs, boolean failOpen, AuthProvider authProvider, + RoutingStatusClient routingStatusClient) { + this(helixManager, timeoutMs, failOpen, authProvider, routingStatusClient, System::currentTimeMillis); + } + + @VisibleForTesting + BrokerRoutingReadyChecker(HelixManager helixManager, long timeoutMs, boolean failOpen, AuthProvider authProvider, + RoutingStatusClient routingStatusClient, LongSupplier currentTimeMs) { + this(createContext(helixManager, routingStatusClient, null, authProvider), timeoutMs, failOpen, currentTimeMs, + authProvider); + } + + private BrokerRoutingReadyChecker(ProductionContext context, long timeoutMs, boolean failOpen, + AuthProvider authProvider) { + this(context, timeoutMs, failOpen, System::currentTimeMillis, authProvider); + } + + private BrokerRoutingReadyChecker(ProductionContext context, long timeoutMs, boolean failOpen, + LongSupplier currentTimeMs, AuthProvider authProvider) { + this(context._serverInstanceId, context._onlineBrokersSupplier, context._allBrokersReady, + context._checkExecutor, context._routingStatusClient, timeoutMs, failOpen, currentTimeMs, + authProvider); + } + + @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, + Predicate<Set<String>> allBrokersReady, long timeoutMs, boolean failOpen, LongSupplier currentTimeMs) { + this(serverInstanceId, onlineBrokersSupplier, (brokers, shouldStop) -> allBrokersReady.test(brokers), null, + RoutingStatusClient.NOOP, timeoutMs, failOpen, currentTimeMs, new NullAuthProvider()); + } + + private BrokerRoutingReadyChecker(String serverInstanceId, Supplier<Set<String>> onlineBrokersSupplier, + BrokersReadyEvaluator allBrokersReady, @Nullable ScheduledExecutorService checkExecutor, + RoutingStatusClient routingStatusClient, long timeoutMs, boolean failOpen, LongSupplier currentTimeMs, + AuthProvider authProvider) { + _serverInstanceId = serverInstanceId; + _onlineBrokersSupplier = onlineBrokersSupplier; + _allBrokersReady = allBrokersReady; + _checkExecutor = checkExecutor; + _routingStatusClient = routingStatusClient; + _currentTimeMs = currentTimeMs; + long nowMs = currentTimeMs.getAsLong(); + _deadlineMs = timeoutMs >= Long.MAX_VALUE - nowMs ? Long.MAX_VALUE : nowMs + Math.max(timeoutMs, 0L); + _failOpen = failOpen; + _authProvider = authProvider; + _state = new AtomicReference<>(State.CHECKING); + if (_checkExecutor != null) { + _checkExecutor.scheduleWithFixedDelay(this::check, 0L, CHECK_INTERVAL_MS, TimeUnit.MILLISECONDS); + } + } + + public boolean isReady() { + transitionToReadyOnFailOpenTimeout(); + return _state.get() == State.READY; + } + + @VisibleForTesting + AuthProvider getAuthProvider() { + return _authProvider; + } + + @VisibleForTesting + synchronized void check() { + if (_state.get() != State.CHECKING || transitionToReadyOnFailOpenTimeout()) { + return; + } + try { + Set<String> onlineBrokers = _onlineBrokersSupplier.get(); + if (!onlineBrokers.isEmpty() && _allBrokersReady.test(onlineBrokers, this::shouldStopChecking) + && _state.get() == State.CHECKING + // Do not mark the server ready if broker membership changed while acknowledgements were collected. + && onlineBrokers.equals(_onlineBrokersSupplier.get())) { + if (_state.compareAndSet(State.CHECKING, State.READY)) { + LOGGER.info("All online brokers report server {} as routable: {}", _serverInstanceId, onlineBrokers); + stopChecking(); + } + return; + } + } catch (Exception e) { + LOGGER.debug("Failed to check broker routing readiness for server {}", _serverInstanceId, e); + } + + if (transitionToReadyOnFailOpenTimeout()) { + return; + } + if (_currentTimeMs.getAsLong() >= _deadlineMs) { + if (!_timeoutLogged) { + LOGGER.warn("Timed out waiting for all online brokers to report server {} as routable; failOpen={}", + _serverInstanceId, _failOpen); + _timeoutLogged = true; + } + } + } + + private boolean shouldStopChecking() { + return _state.get() != State.CHECKING || Thread.currentThread().isInterrupted() + || transitionToReadyOnFailOpenTimeout(); + } + + private boolean transitionToReadyOnFailOpenTimeout() { + if (!_failOpen || _currentTimeMs.getAsLong() < _deadlineMs) { + return false; + } + if (_state.compareAndSet(State.CHECKING, State.READY)) { + LOGGER.warn("Timed out waiting for all online brokers to report server {} as routable; failOpen=true", + _serverInstanceId); + stopChecking(); + } + return _state.get() != State.CHECKING; + } + + private void stopChecking() { + if (_checkExecutor != null) { + _checkExecutor.shutdown(); + } + } + + private static ProductionContext createProductionContext(HelixManager helixManager, AuthProvider authProvider) { + HttpClientConfig httpClientConfig = HttpClientConfig.newBuilder() + .withMaxConns(1) + .withMaxConnsPerRoute(1) + .withConnectionTimeoutMs((int) REQUEST_TIMEOUT_MS) + .withFollowRedirects(false) + .build(); Review Comment: **[P2] Disable automatic HTTP retries for the dedicated polling client** This client disables redirects but retains Apache's automatic retries for 429/503. In the pinned HttpClient 5.6.4, HttpRequestRetryExec gracefully closes the first retryable response before returning control to Pinot, draining its body before the 1,024-byte reader can enforce its limit. A chunked 503 that keeps supplying data can therefore occupy this single worker and consume an arbitrarily large response. The same retry path honors Retry-After: 3600 with an hour-long sleep: DefaultHttpRequestRetryStrategy compares only its default one-second delay against the response timeout, then calculates the header-derived delay separately. With the default fail-closed behavior, the checker cannot observe broker recovery during that drain/sleep period. Fail-open health checks still evaluate their deadline independently. Disable automatic retries for this dedicated client, preserving existing defaults for other callers; the scheduled checker already handles retries. Extend testBoundedResponseDoesNotDrainChunkedRemainder to cover 429/503 and add a Retry-After case—the current slow-body test covers only 200. -- 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]
