xiangfu0 commented on code in PR #20022:
URL: https://github.com/apache/pinot/pull/20022#discussion_r4369185086
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java:
##########
@@ -235,11 +245,32 @@ int getPartitionPruningPreparationThreshold() {
return _partitionPruningPreparationThreshold;
}
- /// Sets a callback to be invoked when a server becomes routable: when it
starts, and when it is enabled again after
- /// it was disabled (for example while it restarted) or excluded. This is
useful for resetting gRPC channel state to
- /// avoid exponential backoff delays and connections to a stale address.
- public void setServerEnabledCallback(Consumer<ServerInstance> callback) {
+ /// Registers a callback invoked with the [ServerInstance] of a server that
becomes routable: when it starts, and when
+ /// it is enabled again after it was disabled (for example while it
restarted) or excluded. Callers use it to drop
+ /// state that may be tied to a previous process of the server: gRPC
reconnection backoff and connections to a stale
+ /// address, cached Netty channels that may be half-open. Callbacks run on
the Helix callback thread while the routing
+ /// manager holds its write lock, so they must be quick and must not block.
+ public void addServerEnabledCallback(Consumer<ServerInstance> callback) {
+ _serverEnabledCallbacks.add(callback);
+ }
+
+ /// Sets a callback to be invoked when a server becomes routable, see
[#addServerEnabledCallback(Consumer)]. Setting
+ /// it again replaces the callback set before through this method, and only
that one, so callbacks registered
+ /// through `addServerEnabledCallback` are never dropped.
+ public synchronized void setServerEnabledCallback(Consumer<ServerInstance>
callback) {
Review Comment:
Small asymmetry: `setServerEnabledCallback` is `synchronized` but
`addServerEnabledCallback`/`addServerRemovedCallback` are not. The
CopyOnWriteArrayList is thread-safe, but `set`'s remove-then-add isn't atomic
against a concurrent `add`. In practice all registration happens during
`start()` so this is benign — just flagging it in case callbacks ever get
registered dynamically later.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/AsyncQueryResponse.java:
##########
@@ -116,6 +121,20 @@ public Map<ServerRoutingInstance, ServerResponse>
getFinalResponses()
}
}
+ /// On a timeout, hands every server the request reached but that never
answered to the transport, with the time
+ /// and the channel the request was written to; the transport judges after a
grace period whether the server has gone
+ /// silent on that channel (see [ServerChannels#recordUnansweredQuery]).
+ private void recordUnansweredServers() {
+ for (Map.Entry<ServerRoutingInstance, ServerResponse> entry :
_responseMap.entrySet()) {
+ ServerResponse response = entry.getValue();
+ int submitDelayMs = response.getSubmitDelayMs();
+ // Only servers the request was actually submitted to can be blamed for
not answering it
+ if (response.getDataTable() == null && submitDelayMs >= 0) {
Review Comment:
When a query times out while still blocked acquiring the channel lock,
`response.getChannel()` is null here, so the judgment counts against whatever
channel the entry currently holds rather than the one this query targeted. That
looks intentional (and is documented on `judgeUnansweredQuery`), just
confirming it's the behavior you want for the stuck-on-lock case.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -182,7 +293,90 @@ public void preConnect(ServerRoutingInstance
serverRoutingInstance, long timeout
_serverToChannelMap.computeIfAbsent(serverRoutingInstance,
ServerChannel::new).preConnect(timeoutMs);
}
+ /// Marks the cached channel to the given routing instance suspect: the
server stopped answering queries on it
+ /// while the socket still looked active. The next query to it, or the next
[#probe], replaces it with a fresh
+ /// connection instead of trusting `isActive()`.
+ public void markSuspect(ServerRoutingInstance serverRoutingInstance) {
+ ServerChannel serverChannel =
_serverToChannelMap.get(serverRoutingInstance);
+ if (serverChannel != null) {
+ serverChannel._suspect = true;
+ }
+ }
+
+ /// The failure detector's retry probe for one channel, when pings are off.
Does nothing and returns `false` when
+ /// this broker has not queried the server over this channel (see
[#hasChannel]); otherwise reconnects the channel if
+ /// it is down, replaces it with a fresh connection if it was marked
suspect, and returns `true`, throwing if the
+ /// connection cannot be established.
+ public boolean probe(ServerRoutingInstance serverRoutingInstance)
+ throws InterruptedException, TimeoutException {
+ ServerChannel serverChannel =
_serverToChannelMap.get(serverRoutingInstance);
+ if (serverChannel == null || !serverChannel._openedByQuery) {
+ return false;
+ }
+ serverChannel.connect();
+ return true;
+ }
+
+ /// Records a response from the server on the given channel: it is alive,
whatever it answered.
+ public void recordResponse(ServerRoutingInstance serverRoutingInstance) {
+ ServerChannel serverChannel =
_serverToChannelMap.get(serverRoutingInstance);
+ if (serverChannel != null) {
+ serverChannel.recordResponse();
+ }
+ }
+
+ /// Records that a query written at `submitTimeMs` to `channel` (when known)
timed out unanswered. The verdict waits
+ /// out the grace period: the query only counts if the entry still holds
that channel and the server has still sent
+ /// nothing on it since the query was written. That lets the server's own
late timeout response through, so a burst
+ /// of slow queries, or a broker pause that delays reading the responses,
does not look like a dead channel: only a
+ /// channel the server has gone silent on does. At the threshold the channel
is marked suspect and the server is
+ /// reported through [QueryRouter#reportUnansweredServer]. No-op when the
check is disabled.
+ public void recordUnansweredQuery(ServerRoutingInstance
serverRoutingInstance, long submitTimeMs,
+ @Nullable Channel channel) {
+ if (_livenessExecutor == null) {
+ return;
+ }
+ ServerChannel serverChannel =
_serverToChannelMap.get(serverRoutingInstance);
+ if (serverChannel == null) {
+ return;
+ }
+ if (serverChannel._numPendingJudgments.incrementAndGet() >
_maxPendingJudgmentsPerChannel) {
Review Comment:
The cap is applied with an increment-then-decrement, so
`_numPendingJudgments` can transiently read `cap+1` from another thread. It's
purely a bound and self-corrects, so no real impact, but a `>= cap` pre-check
or `updateAndGet` would avoid the brief overshoot if you want it exact.
--
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]
