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 35853ce5d0. `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 35853ce5d0. 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/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:
I looked at this, and I don't think the probe can make that call soundly
with pings off. Nothing is routed to a server while it's out of routing, so
nothing would ever come back on the new channel, and holding UNHEALTHY until
then would keep a recovered server out for good. UNKNOWN doesn't help either:
when no other transport has an opinion, an all-UNKNOWN round counts as
recovered (`ConnectionFailureDetectorTest` pins that, because after an eviction
there's nothing left to verify). Telling a frozen JVM from a live one takes an
answer from the application, and that's what pings are for.
What changes instead is the default. With the report now behind its own
flag, off by default (see the thread on the handler), a wedged server is never
taken out of routing by this path and there's no readmission loop; its channel
just gets replaced. With the flag on and pings off, each readmission can cost
up to the threshold of queries per channel before the server is reported again.
The flag's doc says so and points to pings as the better option where every
server supports them.
##########
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 35853ce5d0: `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]
