You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "yashmayya (via GitHub)" <gi...@apache.org> on 2026/10/15 00:07:10 UTC
[PR] Refresh MSE gRPC channels when a server restarts [pinot]
yashmayya opened a new pull request, #19875:
URL: https://github.com/apache/pinot/pull/19875
## Problem
Multi-stage (MSE) queries can time out for hours after server pods restart together, when every server listens on the same query and mailbox ports (for example 8100 and 8101).
Two server pods can swap IP addresses during a batch restart. A gRPC channel to server A can then keep A's old IP, which now belongs to server B. B accepts the data meant for A, or runs A's stage plan, and A waits until the query times out. Nothing logs an error.
## Why a channel keeps the old address
grpc-java 1.81 and 1.84 have the same code here:
- The default load balancer is the old `PickFirstLoadBalancer`.
- When a connection drops, the channel resolves the hostname once, at that moment. A terminating pod can still be in DNS (for example with `publishNotReadyAddresses=true`). `DnsNameResolver` also skips a refresh less than 30 s after the last one.
- The next RPC, maybe much later, connects to that address and does not resolve again. If another server now listens on that IP and port, the connection works, and a working connection is never resolved again.
The existing `ServerGrpcChannelBackoffResetHandler` does not help: it only resets the backoff of a channel in TRANSIENT_FAILURE, and a stale channel is IDLE or READY.
## Fix
- **Servers:** when a peer server finishes startup, refresh the mailbox channel to it with `ManagedChannel.enterIdle()`. The next stream then resolves the hostname again. The handler is renamed to `ServerMailboxChannelRefreshHandler`.
- **Brokers:** refresh the dispatch channel to a server each time the server becomes routable. The routing callback now runs for every newly enabled server, not only for servers that the failure detector excluded.
- **Old addresses:** when a server moves to another address (for example a new random port) or leaves the cluster, shut down the channel to the old address, unless another server uses that address now. Before, each restart left one unused channel per peer.
- A refresh keeps the channel object that senders hold, and lets streams in flight finish. A channel is refreshed at most once a second.
## Limits
This makes misdelivery rare, but cannot rule it out. DNS can still return the old IP when the refresh runs. A query can reach the server between routing and the refresh. Channels from servers to brokers are not refreshed. #19873 adds a receiver-side check for these cases.
## Compatibility
- No wire, config or metric change.
- API changes:
- `BaseBrokerRoutingManager.setServerReenableCallback` is now `setServerEnabledCallback`, and runs for every newly enabled server.
- `QueryDispatcher.resetClientConnectionBackoff`, `MailboxService.resetConnectBackoff` and `ChannelManager.resetConnectBackoff` are removed. A refresh also clears the backoff.
- `MailboxService.shutdown()` also shuts down its client channels.
## Tests
- `ChannelManagerTest`: a refresh keeps the channel object and moves it to IDLE; refreshes are debounced; retire and shutdown.
- `ServerMailboxChannelRefreshHandlerTest`: refresh on startup; retire when a server moves or leaves; keep an address that another server took over; no retire when a config read fails.
- `BrokerRoutingManagerTest`: the callback runs for every newly enabled server, including a restarted one.
- `QueryDispatcherTest`: refresh and retire when a server is enabled; the dispatch channel refresh is debounced.
- The full `pinot-query-runtime`, `pinot-server` and `pinot-broker` suites pass.
This is the first of three PRs: this one, #19873 (receiver check) and a third PR (link to follow) (diagnostics).
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Refresh MSE gRPC channels when a server restarts [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #19875:
URL: https://github.com/apache/pinot/pull/19875#discussion_r4289839924
##########
pinot-server/src/main/java/org/apache/pinot/server/starter/helix/ServerMailboxChannelRefreshHandler.java:
##########
@@ -95,18 +106,28 @@ public synchronized void onInstanceConfigChange(List<InstanceConfig> instanceCon
/// Full cluster scan: used on INIT (first callback) and when instances are added/removed.
private void handleFullScan() {
Set<String> currentShuttingDown = new HashSet<>();
+ Set<String> currentServers = new HashSet<>();
for (String instance : _helixAdmin.getInstancesInCluster(_clusterName)) {
if (!InstanceTypeUtils.isServer(instance) || instance.equals(_selfInstanceId)) {
continue;
}
+ // Before reading the config: a server whose config cannot be read now has not left the cluster
+ currentServers.add(instance);
InstanceConfig config = getInstanceConfig(instance);
if (config == null) {
continue;
}
+ updateMailboxAddress(instance, config);
Review Comment:
During a full scan `updateMailboxAddress` is applied one instance at a time, so the `containsValue` guard in `retireChannel` reads a half-updated `_mailboxAddresses`. If server A (scanned first) vacates an address that server B (scanned later in the same scan) takes over, A's retire runs while B still maps to its old address, so the guard misses the takeover and shuts down a channel B will use. It self-heals via lazy recreate, and the single-instance path is fine, but the full-scan ordering isn't covered by a test. Worth a comment or a two-pass scan.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Refresh MSE gRPC channels when a server restarts [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #19875:
URL: https://github.com/apache/pinot/pull/19875#issuecomment-6181769076
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/19875?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `83.01887%` with `9 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 58.59%. Comparing base ([`7896d13`](https://app.codecov.io/gh/apache/pinot/commit/7896d13c19c10e02ec605fdb45c70d19f92511a7?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`4162aee`](https://app.codecov.io/gh/apache/pinot/commit/4162aeeb8f15c92e5b47c06ba8c1ef49232b9841?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 6 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/pinot/pull/19875?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...he/pinot/query/mailbox/channel/ChannelManager.java](https://app.codecov.io/gh/apache/pinot/pull/19875?src=pr&el=tree&filepath=pinot-query-runtime%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fquery%2Fmailbox%2Fchannel%2FChannelManager.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtcXVlcnktcnVudGltZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvcXVlcnkvbWFpbGJveC9jaGFubmVsL0NoYW5uZWxNYW5hZ2VyLmphdmE=) | 89.28% | [0 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/pinot/pull/19875?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../pinot/query/service/dispatch/QueryDispatcher.java](https://app.codecov.io/gh/apache/pinot/pull/19875?src=pr&el=tree&filepath=pinot-query-runtime%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fquery%2Fservice%2Fdispatch%2FQueryDispatcher.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtcXVlcnktcnVudGltZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvcXVlcnkvc2VydmljZS9kaXNwYXRjaC9RdWVyeURpc3BhdGNoZXIuamF2YQ==) | 76.92% | [0 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/pinot/pull/19875?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...org/apache/pinot/query/mailbox/MailboxService.java](https://app.codecov.io/gh/apache/pinot/pull/19875?src=pr&el=tree&filepath=pinot-query-runtime%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fquery%2Fmailbox%2FMailboxService.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtcXVlcnktcnVudGltZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvcXVlcnkvbWFpbGJveC9NYWlsYm94U2VydmljZS5qYXZh) | 33.33% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/19875?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...e/pinot/query/service/dispatch/DispatchClient.java](https://app.codecov.io/gh/apache/pinot/pull/19875?src=pr&el=tree&filepath=pinot-query-runtime%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fquery%2Fservice%2Fdispatch%2FDispatchClient.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtcXVlcnktcnVudGltZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvcXVlcnkvc2VydmljZS9kaXNwYXRjaC9EaXNwYXRjaENsaWVudC5qYXZh) | 88.88% | [0 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/pinot/pull/19875?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #19875 +/- ##
============================================
+ Coverage 58.57% 58.59% +0.01%
+ Complexity 7 1 -6
============================================
Files 2722 2721 -1
Lines 169962 170086 +124
Branches 27944 27966 +22
============================================
+ Hits 99559 99655 +96
- Misses 61964 61974 +10
- Partials 8439 8457 +18
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/19875/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [integration](https://app.codecov.io/gh/apache/pinot/pull/19875/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/19875/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/19875/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-25](https://app.codecov.io/gh/apache/pinot/pull/19875/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `58.59% <83.01%> (+0.01%)` | :arrow_up: |
| [lane-a](https://app.codecov.io/gh/apache/pinot/pull/19875/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [lane-b](https://app.codecov.io/gh/apache/pinot/pull/19875/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/19875/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `58.59% <83.01%> (+0.01%)` | :arrow_up: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/19875/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `58.59% <83.01%> (+0.01%)` | :arrow_up: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/19875/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `58.59% <83.01%> (+0.01%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/pinot/pull/19875?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Refresh MSE gRPC channels when a server restarts [pinot]
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya merged PR #19875:
URL: https://github.com/apache/pinot/pull/19875
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Refresh MSE gRPC channels when a server restarts [pinot]
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on code in PR #19875:
URL: https://github.com/apache/pinot/pull/19875#discussion_r4364612135
##########
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:
Good point, added `testServerEnabledKeepsChannelToAddressTakenOver` in c7c51f37b4: A is at addr1, B takes over addr1, then A moves to addr2, and B's client stays open. The test fails without the `containsValue` guard.
I kept the `containsValue` scan. It runs only when a server comes back at a new address, and it reads one entry per server.
##########
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:
I checked this, and in grpc-java 1.81 `enterIdle()` does not wait for streams in flight:
- `ManagedChannelImpl.enterIdleMode()` shuts down the name resolver and the load balancer at once, creates a new resolver, and moves the channel to IDLE.
- The old subchannel shuts down gracefully about 5 s later (`SUBCHANNEL_SHUTDOWN_DELAY_SECONDS`), so streams in flight finish on the old connection.
- The next stream exits IDLE with the new resolver.
The part that waits until no calls are active is the idle timeout, not `enterIdle()`.
I also tried it with a stream open. The channel was IDLE 1 ms after `refreshChannel`, the open stream still delivered, and the next stream arrived on a new connection. So I kept the wording. The Javadoc already says that streams in flight finish on the old connection.
##########
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:
Agreed, I'll keep it as is. It runs only when a server moves or leaves, and it reads one entry per server.
##########
pinot-server/src/main/java/org/apache/pinot/server/starter/helix/ServerMailboxChannelRefreshHandler.java:
##########
@@ -95,18 +106,28 @@ public synchronized void onInstanceConfigChange(List<InstanceConfig> instanceCon
/// Full cluster scan: used on INIT (first callback) and when instances are added/removed.
private void handleFullScan() {
Set<String> currentShuttingDown = new HashSet<>();
+ Set<String> currentServers = new HashSet<>();
for (String instance : _helixAdmin.getInstancesInCluster(_clusterName)) {
if (!InstanceTypeUtils.isServer(instance) || instance.equals(_selfInstanceId)) {
continue;
}
+ // Before reading the config: a server whose config cannot be read now has not left the cluster
+ currentServers.add(instance);
InstanceConfig config = getInstanceConfig(instance);
if (config == null) {
continue;
}
+ updateMailboxAddress(instance, config);
Review Comment:
Good catch, fixed in c7c51f37b4. The full scan now records the new address of every server first, and only then retires the old addresses. So the takeover check sees the result of the whole scan.
`testFullScanKeepsChannelToAddressTakenOverInTheSameScan` reads the server that moves away first. It failed before the change.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Refresh MSE gRPC channels when a server restarts [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org