xiangfu0 commented on code in PR #20021:
URL: https://github.com/apache/pinot/pull/20021#discussion_r4369090785
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java:
##########
@@ -447,11 +478,16 @@ private void processInstanceConfigChangeInternal() {
// Calculate new enabled and disabled servers
Set<String> enabledServers = new HashSet<>();
+ // Every server that still has an instance config, enabled or not
+ Set<String> presentServers = new HashSet<>();
List<String> newEnabledServers = new ArrayList<>();
for (ZNRecord instanceConfigZNRecord : instanceConfigZNRecords) {
// Put instance initialization logics into try-catch block to prevent
bad server configs affecting the entire
// cluster
String instanceId = instanceConfigZNRecord.getId();
+ if (InstanceTypeUtils.isServer(instanceId)) {
+ presentServers.add(instanceId);
Review Comment:
Added the note in 4b7bb7ca7d, right where `presentServers` is declared: it
holds the raw ids from the records while `_enabledServerInstanceMap` and
`_disabledServers` are keyed by the interned ids, lookups compare with
`equals()` so the two match by value, and interning only gives those maps an
identity fast path.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java:
##########
@@ -447,11 +478,16 @@ private void processInstanceConfigChangeInternal() {
// Calculate new enabled and disabled servers
Set<String> enabledServers = new HashSet<>();
+ // Every server that still has an instance config, enabled or not
+ Set<String> presentServers = new HashSet<>();
List<String> newEnabledServers = new ArrayList<>();
for (ZNRecord instanceConfigZNRecord : instanceConfigZNRecords) {
// Put instance initialization logics into try-catch block to prevent
bad server configs affecting the entire
// cluster
String instanceId = instanceConfigZNRecord.getId();
+ if (InstanceTypeUtils.isServer(instanceId)) {
+ presentServers.add(instanceId);
Review Comment:
Same comment as the thread above (posted twice); addressed there in
4b7bb7ca7d.
##########
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:
Agreed, it is user-visible. The description's release note now says it
explicitly: a query sent to a server whose channel is above the high watermark
fails at submit time (or skips that server under `skipUnavailableServers`)
instead of being queued in direct memory, and with the 16 MiB default only a
genuinely stalled channel gets there.
--
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]
