raghavyadav01 commented on code in PR #20021:
URL: https://github.com/apache/pinot/pull/20021#discussion_r4448541859
##########
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:
Native transports are off by default, so on a stock broker this never
applies. That leaves the write timeout, which only fires once the kernel send
buffer is full. With small requests at low QPS, a half-open socket keeps taking
writes while the kernel retransmits, and queries keep timing out against that
server for minutes, same as before. Could we add a check that closes a channel
when it has pinned requests outstanding and has read nothing for a while, or at
least say in the release note that TCP_USER_TIMEOUT is effectively off by
default?
##########
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:
This fires on any disabled-to-enabled transition, not only after a restart.
If someone toggles QUERIES_DISABLED or HELIX_ENABLED on a live server,
re-enabling it closes a healthy channel and fails whatever is still in flight
on it, long-timeout queries included. Could eviction skip channels that have
received data since the server was disabled?
##########
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:
When the server closes the retired channel late, a sender waiting on the
channel lock for the replacement isn't pinned yet, so it still gets failed
here. That seems to undercut the point of scoping down events to the request
channel. When a channel is given, could we act only on requests pinned to that
channel and let unpinned ones fail on their own write or connect?
##########
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:
These three new meters are global only, so when write timeouts or rejections
spike there's no way to tell from metrics which server is at fault.
NETTY_CONNECTION_CHANNEL_INACTIVE already tags the server. Could these use
addMeteredValue with the server's short name the same way?
##########
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:
One large request near 16 MiB makes the channel unwritable for everyone
until it drains, so concurrent queries to a healthy server get rejected here.
Without skipUnavailableServers they fail with the server recorded as the failed
one, so the failure detector can briefly exclude a server that is fine. Could
this rejection skip blaming the server, or only reject when the channel has
stayed unwritable past a short grace period?
--
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]
