raghavyadav01 commented on code in PR #20022:
URL: https://github.com/apache/pinot/pull/20022#discussion_r4448470956
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/QueryRouter.java:
##########
@@ -330,12 +382,39 @@ public boolean preConnect(ServerInstance serverInstance,
TableType tableType, lo
}
}
+ /// Closes and forgets every cached channel to the given server: both table
types, over TLS and plaintext. Returns
+ /// the number of live connections closed (see
[ServerChannels#evictChannel]).
+ ///
+ /// Called through `BrokerRequestHandler#onServerEnabled` when the routing
manager enables a server, which is how a
+ /// restart shows up through Helix, and through
`BrokerRequestHandler#onServerRemoved` when a server leaves the
+ /// cluster or moves to another address (with the instance at its previous
address). See
+ /// [ServerChannels#evictChannel] for why the cached channel cannot be
trusted at that point.
+ public int evictChannels(ServerInstance serverInstance) {
+ int numEvicted = 0;
+ for (TableType tableType : TableType.values()) {
+ if (serverInstance.getPort() > 0 && _serverChannels.evictChannel(
+ serverInstance.toServerRoutingInstance(tableType,
ServerInstance.RoutingType.NETTY))) {
+ numEvicted++;
+ }
+ if (_serverChannelsTls != null && serverInstance.getNettyTlsPort() > 0
&& _serverChannelsTls.evictChannel(
+ serverInstance.toServerRoutingInstance(tableType,
ServerInstance.RoutingType.NETTY_TLS))) {
+ numEvicted++;
+ }
+ }
+ return numEvicted;
+ }
+
public void shutDown() {
_serverChannels.shutDown();
+ if (_serverChannelsTls != null) {
+ _serverChannelsTls.shutDown();
+ }
}
void receiveDataTable(ServerRoutingInstance serverRoutingInstance, DataTable
dataTable, int responseSize,
int deserializationTimeMs) {
+ // Any response proves the server and its channel alive, whether or not
the query it answers is still waiting
+
serverChannelsFor(serverRoutingInstance).recordResponse(serverRoutingInstance);
Review Comment:
This only runs once the DataTable has deserialized. If
`DataTableFactory.getDataTable` throws in `DataTableHandler#channelRead0` (say
a heap OOM on a large response), the catch just logs, so a server that did
answer still counts as silent, and five of those in a row get it reported.
Could the response be recorded when the frame arrives in `channelRead0`, before
decoding?
##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/SingleConnectionBrokerRequestHandler.java:
##########
@@ -151,6 +155,12 @@ public
SingleConnectionBrokerRequestHandler(PinotConfiguration config, String br
Preconditions.checkArgument(_pingTimeoutMs > 0, "'%s' must be positive,
got: %s",
CommonConstants.Broker.FailureDetector.CONFIG_OF_PING_TIMEOUT_MS,
_pingTimeoutMs);
_failureDetector.registerServerNotRespondedChecker(this::checkServerNotResponded);
+ } else {
+ // Without pings nothing else acts on a timeout: a server that has gone
silent on one of its channels is reported
+ // like a server the transport saw go down, and leaves routing until the
retry probe re-establishes the channel.
+ // With pings, the ping that follows every timeout makes that call, and
the silent channel is only replaced.
+ _queryRouter.setUnansweredServerReporter(serverRoutingInstance ->
_failureDetector.markServerUnhealthy(
Review Comment:
With pings off and the CONNECTION detector, five timeouts answered more than
2s late now take a server out of routing, which never happened before. A
low-concurrency tenant with heavy queries that overrun their deadline, or a
server GC pause, can hit that with a healthy server. Could the failure-detector
report sit behind its own flag, off by default, while channel replacement stays
on?
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/QueryRouter.java:
##########
@@ -177,18 +203,19 @@ public boolean hasChannel(ServerInstance serverInstance) {
return !channelRoutingInstances(serverInstance).isEmpty();
}
- /// Connects to the given server, returns `true` if the server is
successfully connected.
- ///
- /// This is the reachability probe the failure detector uses when pings are
off, opening the OFFLINE channel with a
- /// TCP connect only. Startup pre-connect uses [#preConnect] instead.
+ /// The reachability probe the failure detector uses when pings are off.
Re-establishes every channel this broker
+ /// has queried the server over, one per table type: reconnects the ones
that are down and replaces the ones marked
+ /// suspect (see [ServerChannels#recordUnansweredQuery]) with a fresh
connection, so that a half-open socket that
+ /// still reports active is never vouched for. Returns `true` if every such
channel is connected afterwards, which
+ /// includes the vacuous case of a server with no such channel: callers that
need "reachable" must check
+ /// [#hasChannel] first, as `retryUnhealthyServer` does. It used to open the
OFFLINE channel only, so on a
+ /// realtime-only cluster a handshake to a freshly restarted process passed
the probe while the channel that carried
+ /// the queries was dead. Startup pre-connect uses [#preConnect] instead.
public boolean connect(ServerInstance serverInstance) {
try {
- if (_serverChannelsTls != null) {
- _serverChannelsTls.connect(
- serverInstance.toServerRoutingInstance(TableType.OFFLINE,
ServerInstance.RoutingType.NETTY_TLS));
- } else {
- _serverChannels.connect(
- serverInstance.toServerRoutingInstance(TableType.OFFLINE,
ServerInstance.RoutingType.NETTY));
+ ServerChannels serverChannels = channels();
+ for (ServerRoutingInstance routingInstance :
channelRoutingInstances(serverInstance)) {
+ serverChannels.probe(routingInstance);
Review Comment:
When this replaces a suspect channel, the probe passes on a bare TCP
connect, and a frozen JVM's kernel still completes the handshake. So with pings
off a wedged server gets readmitted every retry and loses another round of
queries each time. Could the probe report UNHEALTHY (or UNKNOWN) after
replacing a suspect channel until something actually comes back on the new one?
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -230,12 +458,83 @@ protected void initChannel(SocketChannel ch) {
_serverRoutingInstance));
}
});
+ if (isTcpUserTimeoutApplied()) {
+ _bootstrap.option(EpollChannelOption.TCP_USER_TIMEOUT,
_tcpUserTimeoutMs);
+ }
}
void closeChannel() {
- if (_channel != null) {
- _channel.close();
+ Channel channel = _channel;
+ if (channel != null) {
+ closeByBroker(channel);
+ }
+ }
+
+ /// Closes a channel the broker decided to drop (eviction, replacement of
a suspect channel): the queries still in
+ /// flight on it fail through `channelInactive` as on any closed channel,
but the server is not reported for it.
+ private void closeByBroker(Channel channel) {
+ DataTableHandler handler =
channel.pipeline().get(DataTableHandler.class);
+ if (handler != null) {
+ handler.markClosedByBroker();
+ }
+ channel.close();
+ }
+
+ /// See [ServerChannels#evictChannel]. Flags the entry first, then closes
whatever channel it holds; no lock is
+ /// taken so that the routing manager's Helix callback is never blocked
behind a connect or write in progress.
+ void evict() {
+ _evicted = true;
+ closeChannel();
+ }
+
+ void recordResponse() {
+ synchronized (_livenessLock) {
+ _lastResponseTimeMs = System.currentTimeMillis();
+ _unansweredQueries = 0;
+ }
+ }
+
+ int getUnansweredQueries() {
+ synchronized (_livenessLock) {
+ return _unansweredQueries;
+ }
+ }
+
+ /// Starts a clean record for a new connection: not suspect, nothing
unanswered, counted as answered from now. Done
+ /// before a replacement connects as well as after, so that a judgment of
a query written to the dropped channel
+ /// that runs while the connect is in progress finds nothing to count.
+ private void resetLiveness() {
+ synchronized (_livenessLock) {
+ _suspect = false;
+ _lastResponseTimeMs = System.currentTimeMillis();
+ _unansweredQueries = 0;
+ }
+ }
+
+ /// Runs on the liveness thread a grace period after a query written at
`submitTimeMs` to `channel` timed out. See
+ /// [ServerChannels#recordUnansweredQuery].
+ void judgeUnansweredQuery(long submitTimeMs, @Nullable Channel channel) {
+ int unansweredQueries;
+ synchronized (_livenessLock) {
+ // Nothing to add once the channel is already suspect (it is reported
once, until replaced) or no longer in use,
+ // for a query written to a channel since replaced or closed (it says
nothing about the current one), or if the
+ // server answered something since the query was written. Strictly
after: a connect and the first write can
+ // share a millisecond, while a real answer never shares one with its
request. Checked and acted on under the
+ // lock, so a response or a replacement that lands meanwhile either
makes this judgment bail or resets the
+ // count right after it.
+ if (_evicted || _suspect || (channel != null && channel != _channel)
|| _lastResponseTimeMs > submitTimeMs) {
+ return;
+ }
+ unansweredQueries = ++_unansweredQueries;
+ if (unansweredQueries < _unansweredQueriesThreshold) {
+ return;
+ }
+ _unansweredQueries = 0;
+ _suspect = true;
}
+ LOGGER.warn("Server: {} has left {} consecutive queries unanswered on
its channel, each judged {}ms past its "
Review Comment:
This is the only signal when a channel goes suspect or a server is reported
for unanswered queries, so on a dashboard this kind of exclusion looks the same
as a connection failure. Could we add a broker meter here, next to the existing
NETTY_CONNECTION_* ones?
--
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]
