xiangfu0 commented on code in PR #20021:
URL: https://github.com/apache/pinot/pull/20021#discussion_r4451374155
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java:
##########
@@ -557,6 +620,50 @@ private static boolean isEnabledServer(ZNRecord
instanceConfigZNRecord) {
return true;
}
+ private void notifyServerEnabled(ServerInstance serverInstance) {
+ for (Consumer<ServerInstance> callback : _serverEnabledCallbacks) {
+ try {
+ callback.accept(serverInstance);
+ } catch (Exception e) {
+ LOGGER.error("Caught exception in server enabled callback for server:
{}, ignoring it",
+ serverInstance.getInstanceId(), e);
+ }
+ }
+ }
+
+ private void notifyServerRemoved(ServerInstance serverInstance) {
+ LOGGER.info("Server: {} has been removed from the cluster",
serverInstance.getInstanceId());
+ invokeServerRemovedCallbacks(serverInstance);
+ }
+
+ /// A server back under the same id at a different address (hostname, Netty
port or Netty TLS port) has left its
+ /// previous address for good. Listeners key what they cache by address
rather than by id, so they get the removed
+ /// callback with the previous instance: otherwise its cached channels would
never be dropped, and one gone
+ /// half-open would be picked up again if the server ever moved back to that
address. Channels at an address that
Review Comment:
Fixed in b6d544adbc. When the channel turns out to be closed and the entry
has been evicted, `sendRequestWithoutLocking` now returns null, so the caller
retries on a fresh entry like the other eviction races do, instead of failing
the query with the server set as the failed one. A close the broker didn't make
(a server FIN) still throws as before.
`ServerChannelsTest#testEvictionAfterConnectRetriesInsteadOfFailing` covers it.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -146,10 +174,67 @@ 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);
+ // 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, 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 enables a server, which is what a
restart looks like through Helix, and when it
+ /// sees a server removed from the cluster. 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.addMeteredGlobalValue(BrokerMeter.NETTY_CONNECTION_CHANNEL_EVICTIONS,
1);
+ return true;
+ }
+
+ /// Whether [NettyConfig#TCP_USER_TIMEOUT_MS] is set on the channels: it is
a Linux socket option that Netty only
+ /// exposes on the Epoll transport.
+ @VisibleForTesting
+ boolean isTcpUserTimeoutApplied() {
+ return _tcpUserTimeoutMs > 0 && _channelClass == EpollSocketChannel.class;
Review Comment:
Done in b6d544adbc, along the lines you suggested. The routing manager now
also tells the handler when an enabled server gets disabled
(`addServerDisabledCallback`), and the broker notes the disable on that
server's cached channels. On re-enable, a channel is kept if it received data
after the disable and still has queries waiting on it, so a toggle leaves those
queries to finish.
A restart doesn't take that path. A server sets the shutdown flag as it
starts up, even after a crash, so the channel to the old process hears nothing
after the disable and gets evicted. After a graceful shutdown that channel is
already closed by the time the server comes back. I added the "still has
queries waiting" condition so that a channel which answered during a graceful
drain and then went half-open (the host dying mid-shutdown) isn't kept for
nothing. Evicting an idle channel only costs a reconnect.
Tests:
`ServerChannelsTest#testEnabledServerKeepsOnlyChannelHeardFromSinceDisabledThatCarriesQueries`,
`QueryRoutingTest#testEnabledServerKeepsChannelThatCarriesQueriesAndAnsweredSinceDisabled`
against a real server, and
`BrokerRoutingManagerTest#testDisabledCallbackInvokedWhenEnabledServerIsDisabled`.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java:
##########
@@ -461,19 +501,22 @@ private void processInstanceConfigChangeInternal() {
// hit the identity fast path
instanceId = serverInstance.getInstanceId();
enabledServers.add(instanceId);
- if (_enabledServerInstanceMap.put(instanceId, serverInstance) ==
null) {
+ ServerInstance previousInstance =
_enabledServerInstanceMap.put(instanceId, serverInstance);
+ if (previousInstance == null) {
newEnabledServers.add(instanceId);
// NOTE: Remove new enabled server from excluded servers because
the server is likely being restarted
if (_excludedServers.remove(instanceId)) {
LOGGER.info("Got excluded server: {} re-enabled, including it
into the routing", instanceId);
}
- // The server likely restarted, possibly at a new address. The
callback refreshes the GRPC channel to it:
- // it clears any reconnection backoff, without which MSE queries
to this server may fail fast until the
- // next backoff retry succeeds, and it makes the channel resolve
the hostname again.
- if (_serverEnabledCallback != null) {
- _serverEnabledCallback.accept(serverInstance);
- }
+ notifyIfAddressChanged(_disabledServers.remove(instanceId),
serverInstance);
+ // The server likely restarted, possibly at a new address. The
callbacks drop what may be tied to its
+ // previous process: the GRPC channel's reconnection backoff,
without which MSE queries to this server may
+ // fail fast until the next backoff retry succeeds, and its stale
resolved address; cached Netty channels
+ // that may be half-open.
+ notifyServerEnabled(serverInstance);
Review Comment:
I took the first option in b6d544adbc. The rejection now throws its own
`UnwritableChannelException`, and `QueryRouter` fails the query (or skips the
server) without recording the server as the failed one, so the failure detector
never hears about it. A real stall is still judged by the write timeout, which
closes the channel, and that close reports the server as before.
Pings now skip the check and queue behind the full queue instead. They're
tiny, and whether one gets answered is exactly what tells a healthy server
draining a big request from a stalled one, so rejecting it would fail it either
way.
I didn't add a grace wait: it would hold the channel lock and the query
thread while the window is closed, which is what the rejection is there to
avoid. The watermark is configurable if 16 MiB is too low for a deployment with
very large requests. #20022 also no longer marks the channel suspect on a
rejection, for the same reason.
Tests: `QueryRoutingTest#testUnwritableRejectionDoesNotReportServer` (with
and without `skipUnavailableServers`) and
`ServerChannelsTest#testPingIsQueuedOnUnwritableChannel`.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -365,11 +496,31 @@ private void awaitTlsHandshake(Channel channel, long
timeoutMs)
}
}
- void sendRequestWithoutLocking(String rawTableName, AsyncQueryResponse
asyncQueryResponse,
+ /// Returns the channel the request was written to.
+ Channel sendRequestWithoutLocking(String rawTableName, AsyncQueryResponse
asyncQueryResponse,
ServerRoutingInstance serverRoutingInstance, byte[] requestBytes) {
- long startTimeMs = System.currentTimeMillis();
- // The listener runs later, by when a reconnect may have replaced
_channel: on failure it closes this connection.
+ // Pin the channel: the completion listener must act on the channel it
wrote to, not on whatever channel this
+ // entry holds by the time the write completes, or a late failure would
close a healthy reconnected channel.
Channel channel = _channel;
+ if (!channel.isWritable()) {
Review Comment:
Agreed, fixed in b6d544adbc. When a channel is given, only requests pinned
to that channel are affected now. An unpinned request hasn't been written yet,
and its own send settles it: either it gets pinned to a live channel, after
which that channel's events apply, or its write or connect fails and fails it.
The variant without a channel, used for broker-side shedding, still covers the
whole server. I updated
`AsyncQueryResponseTest#testMarkServerUnavailableIsScopedToTheRequestChannel`
and `#testBrokerInitiatedCloseDoesNotBlameTheServer` to match.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/AsyncQueryResponse.java:
##########
@@ -237,8 +263,27 @@ void cancelQuery(ServerRoutingInstance
serverRoutingInstance, Exception exceptio
/// Returns `true` if a server-down event should be acted on: the server was
queried by this response and has not
/// responded yet
private boolean shouldActOnServerDown(ServerRoutingInstance
serverRoutingInstance) {
+ return shouldActOnServerDown(serverRoutingInstance, null, true);
+ }
+
+ /// As above, and when `channel` is given the request must have been written
to that channel. A request whose
+ /// channel is not known yet (it is about to be written, possibly to a
channel that is still being connected) is
+ /// still taken down by a genuine failure of the server, as before, but
never by a close the broker itself made:
+ /// that close only concerns the queries demonstrably written to the closed
channel.
+ private boolean shouldActOnServerDown(ServerRoutingInstance
serverRoutingInstance, @Nullable Channel channel,
+ boolean reportServer) {
ServerResponse serverResponse = _responseMap.get(serverRoutingInstance);
- return serverResponse != null && serverResponse.getDataTable() == null;
+ if (serverResponse == null || serverResponse.getDataTable() != null) {
+ return false;
+ }
+ if (channel == null) {
+ return true;
+ }
+ Channel requestChannel = serverResponse.getChannel();
+ if (requestChannel == null) {
+ return reportServer;
Review Comment:
Done in b6d544adbc. `NETTY_CONNECTION_UNWRITABLE_REJECTIONS`,
`NETTY_CONNECTION_WRITE_TIMEOUTS` and `NETTY_CONNECTION_CHANNEL_EVICTIONS` are
now tagged with the server's short name through `addMeteredValue`, the same way
as `NETTY_CONNECTION_CHANNEL_INACTIVE`. They're new in this PR, so no existing
metric changes name.
--
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]
