xiangfu0 commented on code in PR #20022:
URL: https://github.com/apache/pinot/pull/20022#discussion_r4451442665
##########
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:
Good point, fixed in 2c4b4017b8. `DataTableHandler` now records the response
as soon as a frame arrives, before it decodes it, so a server whose response
fails to deserialize (a heap OOM included) still counts as having answered.
`receiveDataTable` no longer records it.
`QueryRoutingTest#testUndecodableResponseStillCountsAsAnswer` covers it.
##########
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:
Agreed, done in 2c4b4017b8. Reporting the server is now behind
`pinot.broker.failure.detector.report.unanswered.servers`, off by default. Out
of the box the check only marks the channel suspect so the next query replaces
it, which is the half-open fix and never takes a server out of routing. With
the flag on and pings off, the server is reported as before; with pings on the
flag does nothing, since the ping decides. The config doc and the release note
spell this out, and
`SingleConnectionBrokerRequestHandlerFailureDetectorTest#testUnansweredServerIsReportedOnlyWhenEnabled`
covers both settings.
##########
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:
Added in 2c4b4017b8: `NETTY_CONNECTION_SUSPECT_CHANNELS`, tagged with the
server's short name like the other per-server `NETTY_CONNECTION_*` meters,
counts channels marked suspect after the threshold of unanswered queries. With
the report flag on, each one is also a report to the failure detector, so on a
dashboard this kind of exclusion now shows up separately from connection
failures.
`ServerChannelsTest#testUnansweredQueriesAreJudgedAfterTheGracePeriod` checks
it.
--
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]
