xiangfu0 commented on code in PR #20021:
URL: https://github.com/apache/pinot/pull/20021#discussion_r4478887682
##########
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:
Good catch, and yes, it can. The moved-server eviction compares the host and
port in the instance config, so a pod that comes back at a new IP under the
same hostname doesn't trigger it. `TCP_USER_TIMEOUT` would close the channel 10
s after the first unacknowledged byte, but only on Epoll, and native transports
are off by default. The write timeout only fires once the kernel send buffer
fills, which on a long-lived, autotuned connection can take megabytes of
requests, and otherwise the kernel gives up on retransmits about 15 minutes
after the failure. #20022's unanswered-query check is on by default (only
reporting to the failure detector is opt-in) and would replace the channel
after five unanswered queries, but those queries are still lost, and this PR
shouldn't depend on the next one.
So in f4688c8d18 the channel is retired instead of kept, which goes a step
past drain-then-evict. Evicting once the pinned queries drain would still write
every query in between into the old socket. Now the next query, or the failure
detector's reconnect probe, opens a fresh connection right away. The old
channel stays open for the queries already on it and is closed a second past
the deadline of the last one, or later if a query raced the retirement. A live
toggle still lets its in-flight queries finish on the old channel, and evicting
the entry closes a retired channel along with the current one.
`ServerChannelsTest#testEnabledServerRetiresChannelThatCarriesQueries`
covers the replacement, the close at the deadline and the eviction.
`QueryRoutingTest#testEnabledServerRetiresChannelThatCarriesQueries` checks end
to end that a held query is answered on the old channel while the next query
goes out on a fresh one, and that the old channel closes once the held query is
past its deadline.
--
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]
