xiangfu0 commented on code in PR #20021:
URL: https://github.com/apache/pinot/pull/20021#discussion_r4435196562
##########
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()) {
+ if (!channel.isActive()) {
Review Comment:
Small follow-on: now that eviction can hit a routable server (address change
while enabled), an eviction that lands between the post-connect `_evicted`
check and here throws an exception. `QueryRouter` then fails the query with
`_failedServer` set, so the failure detector excludes the server for a close
the broker made. `if (_evicted) return null;` in this branch would retry on a
fresh entry, like the other eviction races do.
---
AI-generated review via IntentLab · Model: claude-opus-5-5.
##########
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:
Minor: "only costs a reconnect" holds for the disable→re-enable path, but
not in the still-enabled branch. Say only the TLS port (or only the Netty port)
changes: `evictChannels(previousInstance)` also closes the channel on the
transport that didn't move. That channel is live and routable, so the queries
in flight on it fail through `channelInactive` (not reported, but still failed
or skipped). Consider evicting only the transports whose address actually
changed, or reword the comment.
---
AI-generated review via IntentLab · Model: 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]
