xiangfu0 commented on code in PR #20021:
URL: https://github.com/apache/pinot/pull/20021#discussion_r4432982899
##########
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:
Nit (observability): Netty's `isWritable()` is also `false` once the channel
is closed (its outbound buffer is nulled). I checked this with
`EmbeddedChannel`. So a channel that went inactive after
`connectWithoutLocking`, for example a server FIN or an eviction mid-send, is
counted in `NETTY_CONNECTION_UNWRITABLE_REJECTIONS` with a "must drain"
message. Consider checking `!channel.isActive()` first and reporting it as a
closed channel.
---
AI-generated review via IntentLab · Model: Claude Opus 5.5 (claude-opus-5-5).
##########
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:
Minor: the previous `ServerInstance` returned here is dropped, and only the
new one is passed to the callbacks. `ServerRoutingInstance` equality is
hostname+port+tableType, so if the server comes back under the same id at a
different host or port, the old `ServerChannel` entry is never evicted and
stays in the map. Consider also notifying with the previous instance when its
address differs. (This leak predates the PR, but the PR now tracks the previous
instance, so the fix is cheap.)
---
AI-generated review via IntentLab · Model: Claude Opus 5.5 (claude-opus-5-5).
##########
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:
Minor: this meters every entry removed, including ones whose channel already
got a FIN or never connected. A graceful rolling restart will bump it for every
server (up to 4 per server: OFFLINE/REALTIME × plain/TLS), and the
`onServerEnabled` log line counts the same way. If operators are meant to read
it as "stale or half-open connection dropped", consider metering only when
`channel != null && channel.isActive()`. Also, the `BrokerMeter` comment
mentions only re-enable, not removal.
---
AI-generated review via IntentLab · Model: Claude Opus 5.5 (claude-opus-5-5).
--
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]
