xiangfu0 commented on code in PR #19875:
URL: https://github.com/apache/pinot/pull/19875#discussion_r4278302292
##########
pinot-server/src/main/java/org/apache/pinot/server/starter/helix/ServerMailboxChannelRefreshHandler.java:
##########
@@ -145,24 +167,51 @@ private static boolean
isShutdownInProgress(InstanceConfig config) {
config.getRecord().getStringField(CommonConstants.Helix.IS_SHUTDOWN_IN_PROGRESS,
"false"));
}
+ /// Records the server's current mailbox address, and retires the channel to
its previous one if it moved.
+ private void updateMailboxAddress(String instanceId, InstanceConfig config) {
+ Pair<String, Integer> address = getMailboxAddress(instanceId, config);
+ Pair<String, Integer> previousAddress =
+ address != null ? _mailboxAddresses.put(instanceId, address) :
_mailboxAddresses.remove(instanceId);
+ if (previousAddress != null && !previousAddress.equals(address)) {
+ retireChannel(instanceId, previousAddress);
+ }
+ }
- private void resetBackoffForServer(String instanceId, InstanceConfig config)
{
+ private void retireChannel(String instanceId, Pair<String, Integer> address)
{
+ // Another server may have taken over the address
+ if (_mailboxAddresses.containsValue(address)) {
Review Comment:
Same O(N) `containsValue` scan as on the dispatcher side — runs once per
server that comes back on a new mailbox port during a rolling restart. It's
negligible for typical clusters and the ordering here is correct, so no change
needed; just flagging it in case very large clusters with mass port changes
ever show up in a profile. A reverse address->count multiset would make it O(1)
if it ever matters.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/channel/ChannelManager.java:
##########
@@ -150,18 +156,63 @@ private ManagedChannel watchState(ManagedChannel channel,
String hostname, int p
return channel;
}
- /// Resets the connection backoff for the channel to the given server if the
channel is in
- /// TRANSIENT_FAILURE state. Returns true if a reset was performed, false
otherwise.
+ /// Makes the channel to the given address resolve the hostname again and
open a new connection before its next
+ /// stream, dropping the connection it has, and any reconnect backoff.
Streams in flight finish on the old
+ /// connection. Returns false if there is no channel to the address, or if
it was refreshed less than a second ago.
+ ///
+ /// A channel resolves the hostname only when it starts, and again after a
connection fails or is lost. When the
+ /// server at that address goes away, gRPC resolves once, right then, and
keeps that answer until the next failure.
+ /// If DNS still had the old IP address at that moment (a terminating pod
stays in DNS for a while), the channel
+ /// later reconnects to the old address, and when another server took that
address over and listens on the same
+ /// port, the connection succeeds and is never questioned again. Call this
when the address can be stale, for example
+ /// when the server at the address has restarted.
///
- /// @return true if the channel was in TRANSIENT_FAILURE and backoff was
reset
- public boolean resetConnectBackoff(String hostname, int port) {
- ManagedChannel channel = _channelMap.get(Pair.of(hostname, port));
- if (channel != null && channel.getState(false) ==
ConnectivityState.TRANSIENT_FAILURE) {
- LOGGER.info("Resetting mailbox channel backoff for server: {}:{}",
hostname, port);
- channel.resetConnectBackoff();
- return true;
+ /// Moving the channel to IDLE, rather than replacing it, keeps the channel
object that concurrent senders may
+ /// already hold valid. The refreshes are limited to one a second, because
each refresh drops the connection that the
+ /// previous one opened.
+ public boolean refreshChannel(String hostname, int port) {
+ Pair<String, Integer> key = Pair.of(hostname, port);
+ ManagedChannel channel = _channelMap.get(key);
+ if (channel == null) {
+ return false;
+ }
+ long nowNs = System.nanoTime();
+ AtomicLong lastRefreshTimeNs =
+ _lastRefreshTimeNs.computeIfAbsent(key, k -> new AtomicLong(nowNs -
MIN_REFRESH_INTERVAL_NS));
+ long lastNs = lastRefreshTimeNs.get();
+ if (nowNs - lastNs < MIN_REFRESH_INTERVAL_NS ||
!lastRefreshTimeNs.compareAndSet(lastNs, nowNs)) {
+ return false;
+ }
+ channel.enterIdle();
Review Comment:
Small caveat worth a Javadoc line: `enterIdle()` only moves the channel to
IDLE (and so re-resolves on the next RPC) once there are no active RPCs/streams
on it. For the mailbox channel a long-lived in-flight stream will defer the
re-resolution until it drains, so the "before the next stream" wording can be
slightly optimistic. Not a correctness problem — it matches the "streams in
flight finish on the old connection" behavior — just worth saying explicitly.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/QueryDispatcher.java:
##########
@@ -896,16 +899,31 @@ DispatchClient
getOrCreateDispatchClient(QueryServerInstance queryServerInstance
_dispatchMaxInboundMessageSizeBytes));
}
- /// Reset the connection backoff for a server. When the GRPC channel enters
a TRANSIENT_FAILURE state from
- /// connection failures, it will fast fail requests and reconnect with
exponential backoff. This method
- /// resets the backoff so servers that have recovered can be reconnected to
immediately.
- public void resetClientConnectionBackoff(ServerInstance serverInstance) {
- String hostname = serverInstance.getHostname();
- int port = serverInstance.getQueryServicePort();
- DispatchClient dispatchClient =
_dispatchClientMap.get(toHostnamePortKey(hostname, port));
+ /// Called when a server becomes routable, which it does when it starts and
when it is enabled again after it was
+ /// disabled or excluded. The server may have restarted at a new address, so:
+ ///
+ /// - the channel to its address resolves the hostname again before the next
request, and connects without
+ /// reconnect backoff. A channel resolves only when a connection fails or
is lost, and the last time was likely
+ /// when the server went away, while DNS could still have the old IP
address. Reconnecting to that address could
+ /// reach another server that took it over.
+ /// - the channel to its previous address, if the server moved to another
port, is shut down, since nothing uses it
+ /// again.
+ public void onServerEnabled(ServerInstance serverInstance) {
+ String hostnamePortKey = toHostnamePortKey(serverInstance.getHostname(),
serverInstance.getQueryServicePort());
+ String previousKey =
_dispatchAddressByInstanceId.put(serverInstance.getInstanceId(),
hostnamePortKey);
+ if (previousKey != null && !previousKey.equals(hostnamePortKey)
+ && !_dispatchAddressByInstanceId.containsValue(previousKey)) {
Review Comment:
This retire guard — "don't shut down the old channel if another server
already took the address over" — is logic-bearing but untested on the
dispatcher side. `QueryDispatcherTest` only covers the move-to-a-fresh-port
case (`port` -> `port+1`), where `containsValue` is always false. Worth adding
a test: enable A at addr1, enable B at addr1, then move A to addr2, and assert
B's client to addr1 is left alone. (Minor aside: `containsValue` is an O(N)
scan here; fine at normal cluster sizes, just noting it runs once per
port-changing re-enable during a rolling restart.)
--
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]
