xiangfu0 commented on code in PR #20021:
URL: https://github.com/apache/pinot/pull/20021#discussion_r4464272389
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -146,10 +176,104 @@ public Channel sendRequest(String rawTableName,
AsyncQueryResponse asyncQueryRes
ServerRoutingInstance serverRoutingInstance, InstanceRequest
instanceRequest, long timeoutMs)
throws Exception {
byte[] requestBytes =
THREAD_LOCAL_T_SERIALIZER.get().serialize(instanceRequest);
- ServerChannel serverChannel =
_serverToChannelMap.computeIfAbsent(serverRoutingInstance, ServerChannel::new);
- // This server is now query-carrying, whoever opened the channel. See
[#hasChannel].
- serverChannel._openedByQuery = true;
- return serverChannel.sendRequest(rawTableName, asyncQueryResponse,
serverRoutingInstance, requestBytes, timeoutMs);
+ // A ping is queued even behind a full outbound queue: it is tiny, and
whether it gets answered is how the broker
+ // tells a server that drains a large request from one that stalled.
Rejecting it would fail the ping either way.
+ boolean rejectIfUnwritable = !instanceRequest.isPing();
+ // One deadline for the whole call: a sender that races an eviction
retries on a fresh entry, but the lock waits
+ // and connects of all its attempts together stay within the caller's
timeout.
+ long deadlineMs = System.currentTimeMillis() + timeoutMs;
+ for (int attempt = 1; attempt <= MAX_SEND_ATTEMPTS; attempt++) {
+ ServerChannel serverChannel =
_serverToChannelMap.computeIfAbsent(serverRoutingInstance, ServerChannel::new);
+ // This server is now query-carrying, whoever opened the channel. See
[#hasChannel].
+ serverChannel._openedByQuery = true;
+ Channel channel = serverChannel.sendRequest(rawTableName,
asyncQueryResponse, serverRoutingInstance,
+ requestBytes, rejectIfUnwritable, Math.max(0L, deadlineMs -
System.currentTimeMillis()));
+ if (channel != null) {
+ return channel;
+ }
+ // The entry was evicted between the lookup and the send (see
[#evictChannel]); it is already out of the map, so
+ // the next lookup creates a fresh one.
+ LOGGER.info("Channel to server: {} was evicted while sending a request,
retrying on a fresh channel (attempt {})",
+ serverRoutingInstance, attempt);
+ }
+ throw new IllegalStateException(
+ "Channel to server: " + serverRoutingInstance + " kept being evicted
while sending a request");
+ }
+
+ /// Closes and forgets the cached channel to the given routing instance, if
there is one. Returns `true` if that
+ /// closed a live connection. An entry whose channel had already closed (the
server sent a FIN as it shut down) or
+ /// never connected is dropped all the same but not counted, so that
+ /// [BrokerMeter#NETTY_CONNECTION_CHANNEL_EVICTIONS] reads as connections
that were still open, possibly half-open,
+ /// when the broker gave up on them, and a graceful rolling restart does not
bump it for every server.
+ ///
+ /// Called when the routing manager sees a server removed from the cluster,
or moved to another address, and when it
+ /// enables a server, which is what a restart looks like through Helix (see
[#evictChannelOfEnabledServer]). The
+ /// cached channel then belongs to a previous process: if that process went
away without a FIN (node flap, OOM kill,
+ /// a network path that stopped forwarding), the socket is half-open,
`isActive()` keeps reporting `true` until the
+ /// kernel gives up on retransmits, and every query routed to the readmitted
server would be written into it and
+ /// lost. Evicting here makes the next query reconnect, to the new process
at whatever address its hostname now
+ /// resolves to.
+ ///
+ /// Queries in flight on the evicted channel fail fast through
`channelInactive` rather than waiting out their
+ /// timeout: the process they were sent to is gone. The entry is flagged
before it is closed so that a sender that
+ /// already holds it does not reconnect it after it has left the map, which
would leak a channel nobody routes to
+ /// (see [ServerChannel#sendRequest]).
+ public boolean evictChannel(ServerRoutingInstance serverRoutingInstance) {
+ ServerChannel serverChannel =
_serverToChannelMap.remove(serverRoutingInstance);
+ if (serverChannel == null) {
+ return false;
+ }
+ // Read before the eviction closes it; a snapshot is enough, as it only
decides what is counted
+ Channel channel = serverChannel._channel;
+ boolean wasConnected = channel != null && channel.isActive();
+ serverChannel.evict();
+ if (!wasConnected) {
+ LOGGER.debug("Dropped cached entry without a live channel to server:
{}", serverRoutingInstance);
+ return false;
+ }
+ LOGGER.info("Evicted live cached channel to server: {}",
serverRoutingInstance);
+
_brokerMetrics.addMeteredValue(BrokerMeter.NETTY_CONNECTION_CHANNEL_EVICTIONS,
1,
+ serverRoutingInstance.getShortName());
+ return true;
+ }
+
+ /// Notes that the routing manager disabled the server, so that
[#evictChannelOfEnabledServer] can later tell a
+ /// channel that heard from the server since then. Has no effect when
nothing is cached for the server.
+ public void markServerDisabled(ServerRoutingInstance serverRoutingInstance) {
+ ServerChannel serverChannel =
_serverToChannelMap.get(serverRoutingInstance);
+ if (serverChannel != null) {
+ serverChannel.markDisabled();
+ }
+ }
+
+ /// [#evictChannel] for a server the routing manager just enabled, except
that a channel which received data after
+ /// the server was disabled (see [#markServerDisabled]) and still carries
queries, as `carriesQueries` tells, is kept:
+ /// the server was disabled and enabled again without a restart, as when an
operator toggles it, and the queries
+ /// still running on the channel are left to finish. Returns `true` if a
live connection was closed.
+ ///
+ /// A restart leaves nothing to keep. A server disables itself as it starts
up, even after a crash, so the channel to
+ /// its previous process, gone or half-open, hears nothing after the
disable. After a graceful shutdown the channel
+ /// is closed by the time the server comes back, and the queries it drained
are no longer in flight.
+ public boolean evictChannelOfEnabledServer(ServerRoutingInstance
serverRoutingInstance,
+ Predicate<Channel> carriesQueries) {
+ ServerChannel serverChannel =
_serverToChannelMap.get(serverRoutingInstance);
+ if (serverChannel != null) {
+ Channel channel = serverChannel.getChannelHeardFromSinceDisabled();
+ if (channel != null && carriesQueries.test(channel)) {
Review Comment:
Trade-off worth a human call: the keep check needs a frame after the
disable. So a quick toggle of a live server whose in-flight queries are all
still computing evicts the healthy channel and fails them. `QueryRoutingTest`
lines 870-876 assert exactly that against a live server. In the restart case,
queries pinned to the old channel have usually timed out by the time the server
is back, so `carriesQueries` alone already separates most of it. If keeping the
stricter check, maybe note it in the release note as a change from master.
---
AI-generated review via IntentLab ยท Model: claude-opus-5-5.
--
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]
