xiangfu0 commented on code in PR #20005:
URL: https://github.com/apache/pinot/pull/20005#discussion_r4368872827
##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/SingleConnectionBrokerRequestHandler.java:
##########
@@ -1048,9 +1048,11 @@ public FailureDetector.ServerState
retryUnhealthyServer(String instanceId) {
LOGGER.info("Retrying unhealthy server: {}", instanceId);
ServerInstance serverInstance =
_routingManager.getEnabledServerInstanceMap().get(instanceId);
+ // Not an enabled server of the routing manager, for example a disabled
server or a server of a remote cluster.
+ // Leave it to the other retriers.
if (serverInstance == null) {
LOGGER.info("Failed to find enabled server: {} in routing manager,
skipping the retry", instanceId);
- return FailureDetector.ServerState.UNHEALTHY;
+ return FailureDetector.ServerState.UNKNOWN;
Review Comment:
Nit: worth noting this also changes the single-stage-only (MSE disabled)
path. With just this one retrier, an unhealthy id not in the routing map (e.g.
a disabled server) now returns all-UNKNOWN and recovers on the first retry
instead of after maxRetries. That's the intended fix and no routing regression,
just slightly beyond what the PR description frames around the MSE retrier.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/QueryDispatcher.java:
##########
@@ -899,6 +920,20 @@ DispatchClient
getOrCreateDispatchClient(QueryServerInstance queryServerInstance
_dispatchMaxInboundMessageSizeBytes));
}
+ /// Records the address of an instance that a query is dispatched to, as its
last known address, see
+ /// [#checkConnectivityToInstance(String)]. Only dispatches record it: a
cancel can still go to an address that the
+ /// instance has left.
+ @VisibleForTesting
+ void recordDispatchAddress(QueryServerInstance queryServerInstance) {
+ String instanceId = queryServerInstance.getInstanceId();
+ String hostnamePortKey =
+ toHostnamePortKey(queryServerInstance.getHostname(),
queryServerInstance.getQueryServicePort());
+ // Read first, so that a dispatch to an instance that did not move does
not write the map
+ if (!hostnamePortKey.equals(_dispatchAddressByInstanceId.get(instanceId)))
{
+ _dispatchAddressByInstanceId.put(instanceId, hostnamePortKey);
Review Comment:
Minor: entries here are only ever `put`, never removed (removal only happens
on `_dispatchClientMap`). Now that every dispatched instance gets an entry, a
custom WorkerManager that mints churning worker ids would accumulate dead
entries, and each one also lengthens the `containsValue` scan in
`onServerEnabled`. Bounded by fleet size in the common case, so not a blocker —
maybe a one-line class-doc note on the lifecycle, or a cleanup on server
removal.
--
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]
