xiangfu0 commented on code in PR #20021:
URL: https://github.com/apache/pinot/pull/20021#discussion_r4466170750
##########
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:
Agreed, relaxed in 99adcee549. On re-enable a channel is now kept whenever
it still has queries waiting on it, so a live toggle leaves its in-flight
queries alone, as master does, whether or not the server sent anything while it
was disabled.
The frame-after-disable condition only mattered for a channel to the
previous process that is both half-open and still carrying a query when the
server is readmitted, and that's rare. A process that exits on a live host has
its socket closed by the kernel, so the channel is already closed by then. A
channel only stays half-open after a host or network failure, and the server
comes back only once the host has recovered and the new process has started and
loaded its segments, by which point the queries pinned to the old channel have
timed out. What's left is a query whose timeout outlasts the restart. There the
next write fails on the reset from a rebooted host, or the write timeout closes
the channel once writes back up, and #20022's unanswered-query check replaces
it after a few timeouts.
Since the disable no longer matters, the server-disabled callback and the
per-channel read mark are gone.
`QueryRoutingTest#testEnabledServerKeepsChannelThatCarriesQueries` checks that
a live toggle keeps a held query's channel with nothing heard since, and
`ServerChannelsTest#testEnabledServerKeepsOnlyOpenChannelThatCarriesQueries`
covers kept, idle and closed channels.
--
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]
