xiangfu0 commented on code in PR #20021:
URL: https://github.com/apache/pinot/pull/20021#discussion_r4451374703
##########
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:
Done in fcd376aa31, 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-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:
I took the first option in fcd376aa31. 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/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:
Agreed, fixed in fcd376aa31. 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-common/src/main/java/org/apache/pinot/common/metrics/BrokerMeter.java:
##########
@@ -232,6 +232,20 @@ public class BrokerMeter implements AbstractMetrics.Meter {
"NETTY_CONNECTION_CHANNEL_ACTIVE", "nettyConnection", false);
public static final BrokerMeter NETTY_CONNECTION_CHANNEL_INACTIVE = create(
"NETTY_CONNECTION_CHANNEL_INACTIVE", "nettyConnection", false);
+ // Requests rejected because the channel's outbound queue was above its high
watermark, i.e. the send window to the
+ // server is closed (see NettyConfig.WRITE_BUFFER_HIGH_WATER_MARK_BYTES).
Each one is a query failed fast instead of
+ // being queued in direct memory behind a stalled socket.
+ public static final BrokerMeter NETTY_CONNECTION_UNWRITABLE_REJECTIONS =
create(
Review Comment:
Done in fcd376aa31. `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]
