xiangfu0 commented on code in PR #20021:
URL: https://github.com/apache/pinot/pull/20021#discussion_r4468562119
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -146,10 +176,96 @@ 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;
+ }
+
+ /// [#evictChannel] for a server the routing manager just enabled, except
that an open channel that still carries
+ /// queries, as `carriesQueries` tells, is kept: the server was most likely
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, as
+ /// they were before channels were evicted on enable. Returns `true` if a
live connection was closed.
+ ///
+ /// A restart rarely leaves anything to keep. A process that exits on a live
host has its socket closed by the
+ /// kernel, so its channel is closed by the time the server comes back. A
channel left half-open takes a host or
+ /// network failure, and the server is back only after the host recovers and
the new process has started and loaded
+ /// its segments, by when the queries written to the old channel have timed
out. A query with a longer timeout can
+ /// keep a half-open channel past the restart; the next write then fails on
the reset from a rebooted host, or the
Review Comment:
Small gap in the "reset from a rebooted host" fallback: if the server comes
back at a new IP behind the same hostname (say, a pod rescheduled after a node
failure), the old IP never sends a reset. A kept half-open channel then
swallows new queries until the send buffer fills or the kernel gives up, since
ping-on-timeout is off by default. It's rare, since it needs a query with a
timeout longer than the restart, but you could flag the kept entry to be
evicted once its pinned queries drain, or mention it in this comment.
---
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]
