Jackie-Jiang commented on code in PR #19452:
URL: https://github.com/apache/pinot/pull/19452#discussion_r4201780833
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -169,6 +170,19 @@ public void connect(ServerRoutingInstance
serverRoutingInstance)
_serverToChannelMap.computeIfAbsent(serverRoutingInstance,
ServerChannel::new).connect();
}
+ /// Closes the channel to the server, if one is open, so that the next
request to it connects afresh.
+ ///
+ /// The broker uses this after a server fails to answer a ping. The channel
may be half-open -- nothing closes it
+ /// when the peer's node disappears without sending a FIN or an RST -- and
requests written into it are never read.
+ /// Closing it also ends whatever is still in flight to that server right
away (see [DataTableHandler]), rather than
+ /// leaving those queries to wait out their own timeouts against a server
that is not going to answer.
+ void closeChannel(ServerRoutingInstance serverRoutingInstance) {
+ ServerChannel serverChannel =
_serverToChannelMap.get(serverRoutingInstance);
+ if (serverChannel != null) {
+ serverChannel.closeChannel();
Review Comment:
[P1] Failed-ping cleanup can close a replacement connection. A ping is sent
on channel A; A disconnects and finishes its inactive callback; a concurrent
query reconnects and installs healthy channel B. Cleanup for the failed ping
then resolves this map entry and ServerChannel.closeChannel() closes the
current _channel, B, aborting the new query. Making _channel volatile does not
prevent this race. Please retain the physical channel used by each ping and
close only that channel, or compare its identity under the channel lock before
closing. Add a regression test that installs B after A fails but before ping
cleanup and verifies that B and its query remain intact.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/QueryRouter.java:
##########
@@ -176,6 +196,101 @@ public boolean connect(ServerInstance serverInstance) {
}
}
+ /// Pings the server over every channel queries to it use, and returns
whether all of them answered within
+ /// `timeoutMs`. Returns `false` when there is no such channel, as there is
nothing to ping over.
+ ///
+ /// A server that is alive answers straight away even when it is busy,
because it replies from its network thread
+ /// instead of queueing behind queries (see `InstanceRequestHandler`). So a
channel that stays silent means the server
+ /// cannot answer anything -- its node is gone, its network path is
blackholed, or its JVM is frozen -- or that the
+ /// channel itself is half-open, left over from a server that went away.
Either way queries sent over it would time
+ /// out, which is why every channel has to answer, not just one.
+ ///
+ /// Each channel that stays silent is closed so the next attempt over it
connects afresh -- see
+ /// [ServerChannels#closeChannel]. Pings record no routing stats, so they
never skew adaptive server selection.
+ public boolean ping(ServerInstance serverInstance, long timeoutMs) {
+ try {
+ List<ServerRoutingInstance> routingInstances =
channelRoutingInstances(serverInstance);
+ if (routingInstances.isEmpty()) {
+ return false;
+ }
+ long requestId = _pingRequestIdGenerator.decrementAndGet();
+ InstanceRequest pingRequest = newPingRequest(requestId, _brokerId);
+ // Like a hybrid query, the pings share one request id and one response,
which waits for every channel.
+ AsyncQueryResponse pingResponse =
+ new AsyncQueryResponse(this, requestId, new
HashSet<>(routingInstances), System.currentTimeMillis(),
+ timeoutMs, _noOpRoutingStatsManager, false);
+ _asyncQueryResponseMap.put(requestId, pingResponse);
+ ServerChannels serverChannels = channels();
+ List<ServerRoutingInstance> pingedRoutingInstances = new
ArrayList<>(routingInstances.size());
+ for (ServerRoutingInstance routingInstance : routingInstances) {
+ pingedRoutingInstances.add(routingInstance);
+ try {
+ serverChannels.sendRequest(PING_TABLE_NAME, pingResponse,
routingInstance, pingRequest, timeoutMs);
+ } catch (Exception e) {
+ // Fails the whole ping; do not spend another connect timeout on the
remaining channels.
+ pingResponse.markQueryFailed(routingInstance, e);
+ break;
+ }
+ }
+ // Removes the ping from _asyncQueryResponseMap on every outcome.
+ // A channel the ping never reached, because an earlier send failed,
stays open. That failed send already makes
+ // the ping unanswered.
+ Map<ServerRoutingInstance, ServerResponse> responses =
pingResponse.getFinalResponses();
+ boolean answered = true;
+ for (ServerRoutingInstance routingInstance : pingedRoutingInstances) {
+ if (responses.get(routingInstance).getDataTable() == null) {
+ answered = false;
+ serverChannels.closeChannel(routingInstance);
Review Comment:
[P1] One failed ping can close another healthy channel. With both OFFLINE
and REALTIME channels, the OFFLINE ping can still be awaiting its response when
REALTIME encounters an immediate send failure or channel reset.
AsyncQueryResponse.markQueryFailed() drains the entire shared latch, so
getFinalResponses() returns immediately and this loop closes OFFLINE before its
ping deadline. That channel's inactive callback then fails unrelated in-flight
queries. Please track each submitted ping's response or deadline independently
so that one channel's failure does not classify the other pending channel as
silent. Add a regression test with a briefly delayed healthy OFFLINE response
and an immediate REALTIME failure, asserting that the healthy channel and its
queries survive.
--
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]
