xiangfu0 commented on code in PR #20021:
URL: https://github.com/apache/pinot/pull/20021#discussion_r4434501974
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -365,11 +486,24 @@ 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:
Good catch, taken. A channel that is closed by the time we write now fails
the request with a "closed before the request could be written" error and is no
longer counted in `NETTY_CONNECTION_UNWRITABLE_REJECTIONS`. The check sits
inside the not-writable branch, so the writable path is unchanged.
`testClosedChannelIsNotCountedAsUnwritable` uses a closed `EmbeddedChannel`, so
it pins both the Netty behaviour and the new branch. Fixed in 08bba83d4e.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -146,10 +174,57 @@ 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 a
+ /// channel was evicted.
+ ///
+ /// 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;
+ }
+ LOGGER.info("Evicting cached channel to server: {}",
serverRoutingInstance);
+
_brokerMetrics.addMeteredGlobalValue(BrokerMeter.NETTY_CONNECTION_CHANNEL_EVICTIONS,
1);
Review Comment:
Taken alongside the writability one, since it was only a few lines.
`evictChannel` still drops every entry, but it only returns true and meters
when the channel was active at eviction time. So a graceful restart whose
channel already got a FIN, or an entry that never connected, is not counted.
The handler log lines now count the same way ("live cached channel(s)"), and
the `BrokerMeter` comment covers re-enable, removal and the address change from
the other thread. `testEvictingEntryWithoutLiveChannelIsNotCounted` covers the
never-connected and already-closed cases. Fixed in 08bba83d4e.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java:
##########
@@ -468,12 +506,12 @@ private void processInstanceConfigChangeInternal() {
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);
- }
+ _disabledServers.remove(instanceId);
Review Comment:
Confirmed this is real, and fixed. Nothing else ever removes those entries:
the new instance's channels are keyed by the new host and port, so the old
entries stayed in the map, and if the old process went away without a FIN the
half-open socket stayed with them. Worse, if the server later moved back to the
old address, that cached and possibly half-open channel would be picked up
again.
The routing manager now calls the removed callbacks with the previous
instance whenever a server comes back under the same id with a different
hostname, Netty port or Netty TLS port. That covers a server that stays
enabled, and one that is disabled and re-enabled, where it compares against the
address the server had when it was disabled. A config change that keeps the
address calls nothing.
`testRemovedCallbackInvokedWithPreviousAddressWhenServerMoves` covers all three
cases. Fixed in 08bba83d4e.
--
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]
