This is an automated email from the ASF dual-hosted git repository.
shauryachats pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new bbbed251444 [broker] Honor skipUnavailableServers on mid-query server
channel-inactive (#19064)
bbbed251444 is described below
commit bbbed251444e00d58bdcbe59a6f233902eb7e79c
Author: Anurag Rai <[email protected]>
AuthorDate: Sat Aug 29 09:46:32 2026 +0530
[broker] Honor skipUnavailableServers on mid-query server channel-inactive
(#19064)
skipUnavailableServers=true only degraded to partial results when a server
was unreachable at request send time (Path 1). If a server's Netty channel
went inactive after dispatch (Path 2), or a write to it failed, the broker
force-failed the whole query with BROKER_REQUEST_SEND (425) instead of
returning partial results from the healthy servers. This was the root cause
of a logging zone-outage incident: a single impacted zone dropped all query
results.
Changes:
- Split markServerDown into markServerUnavailable (skippable: degrades to
partial results under the flag, records the down server so the failure
detector can quarantine it) and markServerCancelled (direct-memory OOM,
always fails the query - no partial data to return).
- Make the latch decrement idempotent: a server can be reported down through
more than one path (write failure closes the channel, which also fires
channelInactive), so a repeat report must not drop a healthy server's
slot.
- Track a single failed server per query; getFailedServer() may now be set
on
a partial success, so it no longer implies failure on its own.
- Add broker meters: SERVER_MARKED_DOWN_SKIPPED (global),
BROKER_RESPONSES_WITH_SEND_EXCEPTIONS (per table), and per-server
NETTY_CONNECTION_CHANNEL_ACTIVE / NETTY_CONNECTION_CHANNEL_INACTIVE tagged
with the server short name, for observability into channel flaps.
Tests: AsyncQueryResponseTest (skip/cancel/idempotency), QueryRoutingTest
Path-2 channel-inactive with and without the flag plus per-server meter
emission, ServerChannelsTest updated for the rename.
---
.../SingleConnectionBrokerRequestHandler.java | 4 +
.../apache/pinot/common/metrics/BrokerMeter.java | 15 ++
.../pinot/core/transport/AsyncQueryResponse.java | 62 +++++--
.../pinot/core/transport/DataTableHandler.java | 8 +-
.../pinot/core/transport/DirectOOMHandler.java | 2 +-
.../apache/pinot/core/transport/QueryResponse.java | 6 +-
.../apache/pinot/core/transport/QueryRouter.java | 22 ++-
.../pinot/core/transport/ServerChannels.java | 6 +-
.../core/transport/AsyncQueryResponseTest.java | 123 +++++++++++++
.../pinot/core/transport/QueryRoutingTest.java | 191 +++++++++++++++++++++
.../pinot/core/transport/ServerChannelsTest.java | 5 +-
.../apache/pinot/spi/utils/CommonConstants.java | 5 +-
12 files changed, 425 insertions(+), 24 deletions(-)
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/SingleConnectionBrokerRequestHandler.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/SingleConnectionBrokerRequestHandler.java
index b8e0cbd3af5..b9d3ad23e19 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/SingleConnectionBrokerRequestHandler.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/SingleConnectionBrokerRequestHandler.java
@@ -192,6 +192,7 @@ public class SingleConnectionBrokerRequestHandler extends
BaseSingleStageBrokerR
if (scatterResult.getSendException() != null) {
brokerResponse.addException(new
QueryProcessingException(QueryErrorCode.BROKER_REQUEST_SEND,
scatterResult.getSendException().getMessage()));
+ _brokerMetrics.addMeteredTableValue(rawTableName,
BrokerMeter.BROKER_RESPONSES_WITH_SEND_EXCEPTIONS, 1);
}
List<ServerRoutingInstance> serversNotResponded =
scatterResult.getServersNotResponded();
if (!serversNotResponded.isEmpty()) {
@@ -324,6 +325,9 @@ public class SingleConnectionBrokerRequestHandler extends
BaseSingleStageBrokerR
brokerResponse.addException(
new QueryProcessingException(QueryErrorCode.BROKER_REQUEST_SEND,
materializedViewSendException.getMessage()));
}
+ if (baseSendException != null || materializedViewSendException != null) {
+ _brokerMetrics.addMeteredTableValue(rawTableName,
BrokerMeter.BROKER_RESPONSES_WITH_SEND_EXCEPTIONS, 1);
+ }
int numServersNotResponded = serversNotResponded.size();
if (numServersNotResponded != 0) {
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerMeter.java
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerMeter.java
index a4ae4cf5f99..18fa4545f91 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerMeter.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerMeter.java
@@ -139,6 +139,15 @@ public class BrokerMeter implements AbstractMetrics.Meter {
public static final BrokerMeter
SECONDARY_WORKLOAD_BROKER_RESPONSES_WITH_PARTIAL_SERVERS_RESPONDED = create(
"SECONDARY_WORKLOAD_BROKER_RESPONSES_WITH_PARTIAL_SERVERS_RESPONDED",
"badResponses", false);
+ // This metric tracks the number of times an in-flight server was skipped
(its channel went inactive or a request
+ // send to it failed) because the query was submitted with
skipUnavailableServers=true.
+ public static final BrokerMeter SERVER_MARKED_DOWN_SKIPPED =
+ create("SERVER_MARKED_DOWN_SKIPPED", "count", true);
+
+ // This metric tracks the number of broker responses carrying a
BROKER_REQUEST_SEND (425) error
+ public static final BrokerMeter BROKER_RESPONSES_WITH_SEND_EXCEPTIONS =
create(
+ "BROKER_RESPONSES_WITH_SEND_EXCEPTIONS", "badResponses", false);
+
public static final BrokerMeter BROKER_RESPONSES_WITH_TIMEOUTS = create(
"BROKER_RESPONSES_WITH_TIMEOUTS", "badResponses", false);
@@ -196,6 +205,12 @@ public class BrokerMeter implements AbstractMetrics.Meter {
"NETTY_CONNECTION_BYTES_RECEIVED", "nettyConnection", true);
public static final BrokerMeter NETTY_CONNECTION_SEND_REQUEST_FAILURES =
create(
"NETTY_CONNECTION_SEND_REQUEST_FAILURES", "nettyConnection", true);
+ // These track server channels transitioning to active/inactive on the
broker (Netty channelActive/channelInactive).
+ // Non-global: emitted per server and tagged with the server short name (see
DataTableHandler).
+ public static final BrokerMeter NETTY_CONNECTION_CHANNEL_ACTIVE = create(
+ "NETTY_CONNECTION_CHANNEL_ACTIVE", "nettyConnection", false);
+ public static final BrokerMeter NETTY_CONNECTION_CHANNEL_INACTIVE = create(
+ "NETTY_CONNECTION_CHANNEL_INACTIVE", "nettyConnection", false);
public static final BrokerMeter PROACTIVE_CLUSTER_CHANGE_CHECK = create(
"PROACTIVE_CLUSTER_CHANGE_CHECK", "proactiveClusterChangeCheck", true);
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/transport/AsyncQueryResponse.java
b/pinot-core/src/main/java/org/apache/pinot/core/transport/AsyncQueryResponse.java
index 3469ea9752d..e0b946b4a9f 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/transport/AsyncQueryResponse.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/transport/AsyncQueryResponse.java
@@ -45,14 +45,20 @@ public class AsyncQueryResponse implements QueryResponse {
private final long _maxEndTimeMs;
private final long _timeoutMs;
private final ServerRoutingStatsManager _serverRoutingStatsManager;
+ private final boolean _skipUnavailableServers;
+ // Servers whose latch slot has already been released via a down-path
(skipServerResponse or markServerUnavailable).
+ // A server can be reported down more than once for the same query
+ private final Set<ServerRoutingInstance> _countedDownServers =
ConcurrentHashMap.newKeySet();
private volatile ServerRoutingInstance _failedServer;
private volatile Exception _exception;
public AsyncQueryResponse(QueryRouter queryRouter, long requestId,
Set<ServerRoutingInstance> serversQueried,
- long startTimeMs, long timeoutMs, ServerRoutingStatsManager
serverRoutingStatsManager) {
+ long startTimeMs, long timeoutMs, ServerRoutingStatsManager
serverRoutingStatsManager,
+ boolean skipUnavailableServers) {
_queryRouter = queryRouter;
_requestId = requestId;
+ _skipUnavailableServers = skipUnavailableServers;
int numServersQueried = serversQueried.size();
_responseMap = new
ConcurrentHashMap<>(HashUtil.getHashMapCapacity(numServersQueried));
_serverRoutingStatsManager = serverRoutingStatsManager;
@@ -189,6 +195,7 @@ public class AsyncQueryResponse implements QueryResponse {
void markQueryFailed(ServerRoutingInstance serverRoutingInstance, Exception
exception) {
_status.set(Status.FAILED);
_failedServer = serverRoutingInstance;
+ _countedDownServers.add(serverRoutingInstance);
_exception = exception;
int count = (int) _countDownLatch.getCount();
for (int i = 0; i < count; i++) {
@@ -196,18 +203,51 @@ public class AsyncQueryResponse implements QueryResponse {
}
}
- /// NOTE: the server might not be hit by the query. Only fail the query if
the query was sent to the server and the
- /// server hasn't responded yet.
- void markServerDown(ServerRoutingInstance serverRoutingInstance, Exception
exception) {
- ServerResponse serverResponse = _responseMap.get(serverRoutingInstance);
- if (serverResponse != null && serverResponse.getDataTable() == null) {
- markQueryFailed(serverRoutingInstance, exception);
+ /// Marks a server as unavailable while the query is in flight - its Netty
channel went inactive or a request write
+ /// to it failed. This is a genuine server-unavailability event that is
eligible for `skipUnavailableServers`.
+ /// NOTE: the server might not have been hit by this query. Only acts if the
query was sent to the server and the
+ /// server has not responded yet.
+ boolean markServerUnavailable(ServerRoutingInstance serverRoutingInstance,
Exception exception) {
+ if (!shouldActOnServerDown(serverRoutingInstance)) {
+ return false;
+ }
+ if (_skipUnavailableServers) {
+ // Best-effort: degrade to partial results. Record the down server so
the failure detector can quarantine it from
+ // routing, but do NOT set the query-global exception/status (no
BROKER_REQUEST_SEND error, query stays
+ // COMPLETED)
+ if (_countedDownServers.add(serverRoutingInstance)) {
+ _failedServer = serverRoutingInstance;
+ _countDownLatch.countDown();
+ return true;
+ }
+ return false;
}
+ markQueryFailed(serverRoutingInstance, exception);
+ return false;
}
- /// Wait for one less server response. This is used when the server is
skipped, as
- /// query submission will have failed we do not want to wait for the
response.
- void skipServerResponse() {
- _countDownLatch.countDown();
+ /// Cancels the query because the broker is shedding it (eg. direct-memory
OOM), not because a server is unavailable
+ void cancelQuery(ServerRoutingInstance serverRoutingInstance, Exception
exception) {
+ if (!shouldActOnServerDown(serverRoutingInstance)) {
+ return;
+ }
+ markQueryFailed(serverRoutingInstance, exception);
+ }
+
+ /// Returns `true` if a server-down event should be acted on: the server was
queried by this response and has not
+ /// responded yet
+ private boolean shouldActOnServerDown(ServerRoutingInstance
serverRoutingInstance) {
+ ServerResponse serverResponse = _responseMap.get(serverRoutingInstance);
+ return serverResponse != null && serverResponse.getDataTable() == null;
+ }
+
+ /// Releases the latch slot for a server that was skipped at
request-submission time (the send failed and the
+ /// query was submitted with `skipUnavailableServers=true`), so the query
does not wait for a response that will
+ /// never come.
+ void skipServerResponse(ServerRoutingInstance serverRoutingInstance) {
+ if (_countedDownServers.add(serverRoutingInstance)) {
+ _failedServer = serverRoutingInstance;
+ _countDownLatch.countDown();
+ }
}
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/transport/DataTableHandler.java
b/pinot-core/src/main/java/org/apache/pinot/core/transport/DataTableHandler.java
index 42abeb9028c..30bc698661f 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/transport/DataTableHandler.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/transport/DataTableHandler.java
@@ -54,12 +54,16 @@ public class DataTableHandler extends
SimpleChannelInboundHandler<ByteBuf> {
@Override
public void channelActive(ChannelHandlerContext ctx) {
LOGGER.info("Channel for server: {} is now active",
_serverRoutingInstance);
+
_brokerMetrics.addMeteredValue(BrokerMeter.NETTY_CONNECTION_CHANNEL_ACTIVE, 1,
+ _serverRoutingInstance.getShortName());
}
@Override
public void channelInactive(ChannelHandlerContext ctx) {
- LOGGER.error("Channel for server: {} is now inactive, marking server
down", _serverRoutingInstance);
- _queryRouter.markServerDown(_serverRoutingInstance,
+ LOGGER.error("Channel for server: {} is now inactive, marking server
unavailable", _serverRoutingInstance);
+
_brokerMetrics.addMeteredValue(BrokerMeter.NETTY_CONNECTION_CHANNEL_INACTIVE, 1,
+ _serverRoutingInstance.getShortName());
+ _queryRouter.markServerUnavailable(_serverRoutingInstance,
new RuntimeException(String.format("Channel for server: %s is
inactive", _serverRoutingInstance)));
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/transport/DirectOOMHandler.java
b/pinot-core/src/main/java/org/apache/pinot/core/transport/DirectOOMHandler.java
index 17424a9f707..d80847e99d3 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/transport/DirectOOMHandler.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/transport/DirectOOMHandler.java
@@ -128,7 +128,7 @@ public class DirectOOMHandler extends
ChannelInboundHandlerAdapter {
removed.closeChannel();
removed.setSilentShutdown();
});
- _queryRouter.markServerDown(_serverRoutingInstance,
+ _queryRouter.cancelQuery(_serverRoutingInstance,
new QueryCancelledException("Query cancelled as broker is out
of direct memory"));
} else if (_allChannels != null && !_allChannels.isEmpty()) { //
server side direct OOM handler
LOGGER.error("Closing channel from broker, as we are running out
of direct memory "
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/transport/QueryResponse.java
b/pinot-core/src/main/java/org/apache/pinot/core/transport/QueryResponse.java
index b0e8b157cf1..42688f4a928 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/transport/QueryResponse.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/transport/QueryResponse.java
@@ -55,7 +55,11 @@ public interface QueryResponse {
/// @return
long getServerResponseDelayMs(ServerRoutingInstance serverRoutingInstance);
- /// Returns the failed server if the query fails.
+ /// Returns the server that went down during the query. Set when the query
fails, and also when the query returns
+ /// partial results under `skipUnavailableServers` (used by the failure
detector to quarantine the server from
+ /// routing). Because it can be set on a partial success, a non-null value
does not by itself imply the query
+ /// failed; check [#getException()] / [#getStatus()] for that. At most one
server is tracked per query -
+ /// when several servers go down, this holds the most recent one.
@Nullable
ServerRoutingInstance getFailedServer();
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/transport/QueryRouter.java
b/pinot-core/src/main/java/org/apache/pinot/core/transport/QueryRouter.java
index 0f7daf4345b..e885a14788a 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/transport/QueryRouter.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/transport/QueryRouter.java
@@ -100,7 +100,7 @@ public class QueryRouter {
// Create the asynchronous query response with the request map
AsyncQueryResponse asyncQueryResponse =
new AsyncQueryResponse(this, requestId, requestMap.keySet(),
System.currentTimeMillis(), timeoutMs,
- _serverRoutingStatsManager);
+ _serverRoutingStatsManager, skipUnavailableServers);
_asyncQueryResponseMap.put(requestId, asyncQueryResponse);
for (Map.Entry<ServerRoutingInstance, InstanceRequest> entry :
requestMap.entrySet()) {
ServerRoutingInstance serverRoutingInstance = entry.getKey();
@@ -118,7 +118,7 @@ public class QueryRouter {
} catch (Exception e) {
_brokerMetrics.addMeteredTableValue(rawTableName,
BrokerMeter.REQUEST_SEND_EXCEPTIONS, 1);
if (skipUnavailableServers) {
- asyncQueryResponse.skipServerResponse();
+ asyncQueryResponse.skipServerResponse(serverRoutingInstance);
} else {
markQueryFailed(requestId, serverRoutingInstance,
asyncQueryResponse, e);
break;
@@ -188,9 +188,23 @@ public class QueryRouter {
}
}
- void markServerDown(ServerRoutingInstance serverRoutingInstance, Exception
exception) {
+ /// Marks a server as unavailable for every in-flight query. Called when a
server's channel goes inactive
+ /// ([DataTableHandler]) or a request write to it fails ([ServerChannels]).
Queries submitted with
+ /// `skipUnavailableServers=true` degrade to partial results for this
genuine unavailability; others are failed.
+ void markServerUnavailable(ServerRoutingInstance serverRoutingInstance,
Exception exception) {
for (AsyncQueryResponse asyncQueryResponse :
_asyncQueryResponseMap.values()) {
- asyncQueryResponse.markServerDown(serverRoutingInstance, exception);
+ if (asyncQueryResponse.markServerUnavailable(serverRoutingInstance,
exception)) {
+
_brokerMetrics.addMeteredGlobalValue(BrokerMeter.SERVER_MARKED_DOWN_SKIPPED, 1);
+ }
+ }
+ }
+
+ /// Cancels every in-flight query. Unlike
+ /// [#markServerUnavailable], this always fails the queries even under
`skipUnavailableServers`: all
+ /// channels are being closed so there is no partial data to return.
+ void cancelQuery(ServerRoutingInstance serverRoutingInstance, Exception
exception) {
+ for (AsyncQueryResponse asyncQueryResponse :
_asyncQueryResponseMap.values()) {
+ asyncQueryResponse.cancelQuery(serverRoutingInstance, exception);
}
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java
b/pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java
index d1eab6e2e3f..2bf34540cf3 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java
@@ -256,8 +256,10 @@ public class ServerChannels {
} else {
LOGGER.error("Write failure to server: {} for table: {}",
serverRoutingInstance, rawTableName, f.cause());
_brokerMetrics.addMeteredGlobalValue(BrokerMeter.NETTY_CONNECTION_SEND_REQUEST_FAILURES,
1);
- asyncQueryResponse.markServerDown(serverRoutingInstance,
- new RuntimeException("Failed to send request to server: " +
serverRoutingInstance, f.cause()));
+ if (asyncQueryResponse.markServerUnavailable(serverRoutingInstance,
+ new RuntimeException("Failed to send request to server: " +
serverRoutingInstance, f.cause()))) {
+
_brokerMetrics.addMeteredGlobalValue(BrokerMeter.SERVER_MARKED_DOWN_SKIPPED, 1);
+ }
_channel.close();
}
});
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/transport/AsyncQueryResponseTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/transport/AsyncQueryResponseTest.java
new file mode 100644
index 00000000000..7d4c242ff45
--- /dev/null
+++
b/pinot-core/src/test/java/org/apache/pinot/core/transport/AsyncQueryResponseTest.java
@@ -0,0 +1,123 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.core.transport;
+
+import java.util.Map;
+import java.util.Set;
+import
org.apache.pinot.core.transport.server.routing.stats.ServerRoutingStatsManager;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.exception.QueryCancelledException;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.mock;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
+
+
+public class AsyncQueryResponseTest {
+ private static final ServerRoutingInstance SERVER_1 =
+ new ServerRoutingInstance("localhost", 11001, TableType.OFFLINE);
+ private static final ServerRoutingInstance SERVER_2 =
+ new ServerRoutingInstance("localhost", 11002, TableType.OFFLINE);
+
+ private AsyncQueryResponse newResponse(Set<ServerRoutingInstance> servers,
long timeoutMs,
+ boolean skipUnavailableServers) {
+ return new AsyncQueryResponse(mock(QueryRouter.class), 1L, servers,
System.currentTimeMillis(), timeoutMs,
+ mock(ServerRoutingStatsManager.class), skipUnavailableServers);
+ }
+
+ @Test
+ public void testMarkServerUnavailableSkipsUnderFlag() {
+ AsyncQueryResponse response = newResponse(Set.of(SERVER_1, SERVER_2),
10_000L, true);
+
+ boolean skipped = response.markServerUnavailable(SERVER_1, new
RuntimeException("channel inactive"));
+
+ assertTrue(skipped);
+ // The skip path must not fail the query: no exception, status not moved
to FAILED.
+ assertNull(response.getException());
+ assertEquals(response.getStatus(), QueryResponse.Status.IN_PROGRESS);
+ // The down server is recorded for the failure detector.
+ assertEquals(response.getFailedServer(), SERVER_1);
+ }
+
+ @Test
+ public void testMarkServerCancelledAlwaysFailsEvenWithSkip() {
+ // Even with skipUnavailableServers=true, a broker-initiated OOM
cancellation must fail the whole query. This guards
+ // against a future refactor silently making OOM honor the skip flag.
+ AsyncQueryResponse response = newResponse(Set.of(SERVER_1, SERVER_2),
10_000L, true);
+
+ response.cancelQuery(SERVER_1,
+ new QueryCancelledException("Query cancelled as broker is out of
direct memory"));
+
+ assertEquals(response.getStatus(), QueryResponse.Status.FAILED);
+ assertNotNull(response.getException());
+ assertEquals(response.getFailedServer(), SERVER_1);
+ }
+
+ @Test
+ public void testMarkServerUnavailableWithoutFlagFails() {
+ // Genuine unavailability but the flag is off: the whole query must still
fail.
+ AsyncQueryResponse response = newResponse(Set.of(SERVER_1, SERVER_2),
10_000L, false);
+
+ boolean skipped = response.markServerUnavailable(SERVER_1, new
RuntimeException("channel inactive"));
+
+ assertFalse(skipped);
+ assertEquals(response.getStatus(), QueryResponse.Status.FAILED);
+ assertNotNull(response.getException());
+ }
+
+ @Test
+ public void testMarkServerUnavailableIsIdempotent()
+ throws Exception {
+ // A server can be reported down twice for the same query
+ AsyncQueryResponse response = newResponse(Set.of(SERVER_1, SERVER_2),
500L, true);
+
+ assertTrue(response.markServerUnavailable(SERVER_1, new
RuntimeException("write failure")));
+ // Second report for the same server is a no-op for the latch.
+ assertFalse(response.markServerUnavailable(SERVER_1, new
RuntimeException("channel inactive")));
+
+ // SERVER_2 never responds
+ Map<ServerRoutingInstance, ServerResponse> responses =
response.getFinalResponses();
+ assertEquals(responses.size(), 2);
+ assertEquals(response.getStatus(), QueryResponse.Status.TIMED_OUT);
+ assertNull(response.getException());
+ assertEquals(response.getFailedServer(), SERVER_1);
+ }
+
+ @Test
+ public void testSkipServerResponseThenUnavailableIsIdempotent()
+ throws Exception {
+ AsyncQueryResponse response = newResponse(Set.of(SERVER_1, SERVER_2),
500L, true);
+
+ response.skipServerResponse(SERVER_1);
+ // The later channel-inactive report for the same server must be a no-op
for the latch.
+ assertFalse(response.markServerUnavailable(SERVER_1, new
RuntimeException("channel inactive")));
+
+ // Only SERVER_1's single slot was released; SERVER_2 never responds, so
the query must TIME OUT, not COMPLETE.
+ Map<ServerRoutingInstance, ServerResponse> responses =
response.getFinalResponses();
+ assertEquals(responses.size(), 2);
+ assertEquals(response.getStatus(), QueryResponse.Status.TIMED_OUT);
+ assertNull(response.getException());
+ // The send-time skip also records the server for quarantine.
+ assertEquals(response.getFailedServer(), SERVER_1);
+ }
+}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/transport/QueryRoutingTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/transport/QueryRoutingTest.java
index c21ac8c6a29..86a12592192 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/transport/QueryRoutingTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/transport/QueryRoutingTest.java
@@ -19,12 +19,15 @@
package org.apache.pinot.core.transport;
import com.google.common.util.concurrent.Futures;
+import io.netty.channel.ChannelHandlerContext;
import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
import org.apache.pinot.common.datatable.DataTable;
import org.apache.pinot.common.datatable.DataTable.MetadataKey;
+import org.apache.pinot.common.metrics.BrokerMeter;
import org.apache.pinot.common.metrics.BrokerMetrics;
import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.common.request.BrokerRequest;
@@ -39,6 +42,8 @@ import org.apache.pinot.spi.accounting.ThreadAccountantUtils;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.exception.QueryErrorCode;
+import org.apache.pinot.spi.metrics.PinotMetricUtils;
+import org.apache.pinot.spi.metrics.PinotMetricsRegistry;
import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.sql.parsers.CalciteSqlCompiler;
import org.apache.pinot.util.TestUtils;
@@ -140,6 +145,15 @@ public class QueryRoutingTest {
return queryScheduler;
}
+ private QueryRouter newIsolatedQueryRouter() {
+ Map<String, Object> properties = new HashMap<>();
+
properties.put(CommonConstants.Broker.AdaptiveServerSelector.CONFIG_OF_ENABLE_STATS_COLLECTION,
true);
+ ServerRoutingStatsManager statsManager =
+ new ServerRoutingStatsManager(new PinotConfiguration(properties),
mock(BrokerMetrics.class));
+ statsManager.init();
+ return new QueryRouter("testBroker", null, null, statsManager,
ThreadAccountantUtils.getNoOpAccountant());
+ }
+
@Test
public void testValidResponse()
throws Exception {
@@ -555,6 +569,183 @@ public class QueryRoutingTest {
_serverRoutingStatsManager.fetchNumInFlightRequestsForServer(serverInstance2.getInstanceId()).intValue(),
0);
}
+ @Test
+ public void testSkipUnavailableServerChannelInactive()
+ throws Exception {
+ long requestId = 123;
+ DataSchema dataSchema =
+ new DataSchema(new String[]{"column1"}, new
DataSchema.ColumnDataType[]{DataSchema.ColumnDataType.STRING});
+ DataTableBuilder builder =
DataTableBuilderFactory.getDataTableBuilder(dataSchema);
+ builder.startRow();
+ builder.setColumn(0, "value1");
+ builder.finishRow();
+ DataTable dataTableSuccess = builder.build();
+ dataTableSuccess.getMetadata().put(MetadataKey.REQUEST_ID.getName(),
Long.toString(requestId));
+ byte[] successResponseBytes = dataTableSuccess.toBytes();
+
+ // The healthy server responds after a delay; the second server has a warm
channel established during submitQuery
+ // and its channel then goes inactive mid-flight (Path 2) before it can
respond.
+ long healthyDelayMs = 1000L;
+ long timeoutMs = 10_000L;
+ _queryServer = getQueryServer((int) healthyDelayMs, successResponseBytes,
0);
+ int healthyPort = startAndGetPort(_queryServer);
+ // Long delay so this server never actually responds during the test
+ QueryServer unavailableServer = getQueryServer(60_000,
successResponseBytes, 0);
+ int unavailablePort = startAndGetPort(unavailableServer);
+ // Isolated broker
+ QueryRouter queryRouter = newIsolatedQueryRouter();
+
+ try {
+ ServerInstance healthyInstance = new ServerInstance("localhost",
healthyPort);
+ ServerInstance unavailableInstance = new ServerInstance("localhost",
unavailablePort);
+ ServerRoutingInstance healthyRoutingInstance =
+ healthyInstance.toServerRoutingInstance(TableType.OFFLINE,
ServerInstance.RoutingType.NETTY);
+ ServerRoutingInstance unavailableRoutingInstance =
+ unavailableInstance.toServerRoutingInstance(TableType.OFFLINE,
ServerInstance.RoutingType.NETTY);
+ Map<ServerInstance, SegmentsToQuery> routingTable =
+ Map.of(healthyInstance, new SegmentsToQuery(List.of(), List.of()),
+ unavailableInstance, new SegmentsToQuery(List.of(), List.of()));
+
+ BrokerRequest brokerRequest =
+ CalciteSqlCompiler.compileToBrokerRequest("SET
skipUnavailableServers=true; SELECT * FROM testTable");
+ long startTime = System.currentTimeMillis();
+ AsyncQueryResponse asyncQueryResponse =
+ queryRouter.submitQuery(requestId, "testTable", brokerRequest,
routingTable, null, null, timeoutMs);
+ // Confirm the request was actually dispatched to the unavailable server
(its channel went live) so this is a
+ // genuine Path 2 (channel dies AFTER dispatch), not a send-time
failure. The server is up, so the write succeeds
+ // within milliseconds.
+ ServerResponse unavailableDispatch =
asyncQueryResponse.getCurrentResponses().get(unavailableRoutingInstance);
+ TestUtils.waitForCondition(aVoid ->
unavailableDispatch.getRequestSentDelayMs() >= 0, 10L, 5000L,
+ "Request was not dispatched to the unavailable server");
+ // Drive the mid-flight channel-inactive event through the real broker
inbound handler
+ new DataTableHandler(queryRouter,
ThreadAccountantUtils.getNoOpAccountant(), unavailableRoutingInstance)
+ .channelInactive(mock(ChannelHandlerContext.class));
+
+ Map<ServerRoutingInstance, ServerResponse> response =
asyncQueryResponse.getFinalResponses();
+ long elapsed = System.currentTimeMillis() - startTime;
+
+ assertEquals(response.size(), 2);
+ ServerResponse healthyResponse = response.get(healthyRoutingInstance);
+ ServerResponse unavailableResponse =
response.get(unavailableRoutingInstance);
+ // The healthy server returned data; the unavailable server did not.
+ assertNotNull(healthyResponse.getDataTable());
+ assertNull(unavailableResponse.getDataTable());
+ // No BROKER_REQUEST_SEND (425): the query degraded to partial results
instead of failing.
+ assertNull(asyncQueryResponse.getException());
+ assertEquals(asyncQueryResponse.getStatus(),
QueryResponse.Status.COMPLETED);
+ // The down server is still recorded so the failure detector can
quarantine it from routing.
+ assertEquals(asyncQueryResponse.getFailedServer(),
unavailableRoutingInstance);
+ // We waited for the healthy server (the latch was not force-drained)
and returned well before the timeout.
+ // If the bug were present, markQueryFailed would force-drain the latch
and return almost immediately.
+ assertTrue(elapsed >= healthyDelayMs, "Expected to wait for the healthy
server, elapsed=" + elapsed);
+ assertTrue(elapsed < timeoutMs, "Expected to return before timeout,
elapsed=" + elapsed);
+ } finally {
+ unavailableServer.shutDown();
+ queryRouter.shutDown();
+ }
+ }
+
+ /// Control for [#testSkipUnavailableServerChannelInactive]: the same
channel-inactive-mid-flight scenario but
+ /// WITHOUT `skipUnavailableServers` must still fail the whole query
+ @Test
+ public void testChannelInactiveWithoutSkipFailsQuery()
+ throws Exception {
+ long requestId = 123;
+ DataSchema dataSchema =
+ new DataSchema(new String[]{"column1"}, new
DataSchema.ColumnDataType[]{DataSchema.ColumnDataType.STRING});
+ DataTableBuilder builder =
DataTableBuilderFactory.getDataTableBuilder(dataSchema);
+ builder.startRow();
+ builder.setColumn(0, "value1");
+ builder.finishRow();
+ DataTable dataTableSuccess = builder.build();
+ dataTableSuccess.getMetadata().put(MetadataKey.REQUEST_ID.getName(),
Long.toString(requestId));
+ byte[] successResponseBytes = dataTableSuccess.toBytes();
+
+ long timeoutMs = 10_000L;
+ // Healthy server delay is set well above the deterministic fail-fast path
+ long healthyDelayMs = 4000L;
+ _queryServer = getQueryServer((int) healthyDelayMs, successResponseBytes,
0);
+ int healthyPort = startAndGetPort(_queryServer);
+ // Long delay so this server never actually responds during the test
+ QueryServer unavailableServer = getQueryServer(60_000,
successResponseBytes, 0);
+ int unavailablePort = startAndGetPort(unavailableServer);
+ // Isolated broker
+ QueryRouter queryRouter = newIsolatedQueryRouter();
+
+ try {
+ ServerInstance healthyInstance = new ServerInstance("localhost",
healthyPort);
+ ServerInstance unavailableInstance = new ServerInstance("localhost",
unavailablePort);
+ ServerRoutingInstance unavailableRoutingInstance =
+ unavailableInstance.toServerRoutingInstance(TableType.OFFLINE,
ServerInstance.RoutingType.NETTY);
+ Map<ServerInstance, SegmentsToQuery> routingTable =
+ Map.of(healthyInstance, new SegmentsToQuery(List.of(), List.of()),
+ unavailableInstance, new SegmentsToQuery(List.of(), List.of()));
+
+ // No skipUnavailableServers option set.
+ long startTime = System.currentTimeMillis();
+ AsyncQueryResponse asyncQueryResponse =
+ queryRouter.submitQuery(requestId, "testTable", BROKER_REQUEST,
routingTable, null, null, timeoutMs);
+ // Confirm the request was actually dispatched to the unavailable server
(its channel went live) so this is a
+ // genuine Path 2 (channel dies AFTER dispatch), not a send-time failure.
+ ServerResponse unavailableDispatch =
asyncQueryResponse.getCurrentResponses().get(unavailableRoutingInstance);
+ TestUtils.waitForCondition(aVoid ->
unavailableDispatch.getRequestSentDelayMs() >= 0, 10L, 5000L,
+ "Request was not dispatched to the unavailable server");
+ // Drive the mid-flight channel-inactive event through the real broker
inbound handler, deterministically (see the
+ // rationale in testSkipUnavailableServerChannelInactive).
+ new DataTableHandler(queryRouter,
ThreadAccountantUtils.getNoOpAccountant(), unavailableRoutingInstance)
+ .channelInactive(mock(ChannelHandlerContext.class));
+
+ Map<ServerRoutingInstance, ServerResponse> response =
asyncQueryResponse.getFinalResponses();
+ long elapsed = System.currentTimeMillis() - startTime;
+
+ assertEquals(response.size(), 2);
+ // Without the flag the query fails: the exception is set (becomes a 425
downstream) and status is FAILED.
+ assertNotNull(asyncQueryResponse.getException());
+ assertEquals(asyncQueryResponse.getStatus(),
QueryResponse.Status.FAILED);
+ assertEquals(asyncQueryResponse.getFailedServer(),
unavailableRoutingInstance);
+ // The latch is force-drained, so the query fails fast — it returns
before the healthy server's delay elapses
+ // rather than waiting it out (and well before the timeout).
+ assertTrue(elapsed < healthyDelayMs, "Expected to fail fast before the
healthy server responded, elapsed="
+ + elapsed);
+ } finally {
+ unavailableServer.shutDown();
+ queryRouter.shutDown();
+ }
+ }
+
+ @Test
+ public void testChannelActiveInactiveEmitPerServerTaggedMeters() {
+ PinotMetricUtils.init(new PinotConfiguration());
+ // register() is a compareAndSet against NOOP with no deregister, so
another test class may already have registered
+ // a real instance. Either way, read from whatever DataTableHandler will
read via BrokerMetrics.get().
+ BrokerMetrics.register(new
BrokerMetrics(PinotMetricUtils.getPinotMetricsRegistry()));
+ BrokerMetrics brokerMetrics = BrokerMetrics.get();
+ PinotMetricsRegistry registry = brokerMetrics.getMetricsRegistry();
+
+ ServerInstance serverInstance = new ServerInstance("localhost", 12345);
+ ServerRoutingInstance routingInstance =
+ serverInstance.toServerRoutingInstance(TableType.OFFLINE,
ServerInstance.RoutingType.NETTY);
+ String shortName = routingInstance.getShortName();
+ DataTableHandler handler =
+ new DataTableHandler(_queryRouter,
ThreadAccountantUtils.getNoOpAccountant(), routingInstance);
+
+ long activeBefore = taggedMeterCount(registry,
BrokerMeter.NETTY_CONNECTION_CHANNEL_ACTIVE, shortName);
+ handler.channelActive(mock(ChannelHandlerContext.class));
+ assertEquals(taggedMeterCount(registry,
BrokerMeter.NETTY_CONNECTION_CHANNEL_ACTIVE, shortName), activeBefore + 1);
+
+ // channelInactive also calls markServerUnavailable, but no query is in
flight on _queryRouter so that is a no-op.
+ long inactiveBefore = taggedMeterCount(registry,
BrokerMeter.NETTY_CONNECTION_CHANNEL_INACTIVE, shortName);
+ handler.channelInactive(mock(ChannelHandlerContext.class));
+ assertEquals(taggedMeterCount(registry,
BrokerMeter.NETTY_CONNECTION_CHANNEL_INACTIVE, shortName),
+ inactiveBefore + 1);
+ }
+
+ private static long taggedMeterCount(PinotMetricsRegistry registry,
BrokerMeter meter, String tag) {
+ String fullName = CommonConstants.Broker.DEFAULT_METRICS_NAME_PREFIX +
meter.getMeterName() + "." + tag;
+ return PinotMetricUtils.makePinotMeter(registry,
+ PinotMetricUtils.makePinotMetricName(BrokerMetrics.class, fullName),
meter.getUnit(), TimeUnit.SECONDS).count();
+ }
+
private void waitForStatsUpdate(long taskCount) {
TestUtils.waitForCondition(aVoid -> {
return (_serverRoutingStatsManager.getCompletedTaskCount() == taskCount);
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/transport/ServerChannelsTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/transport/ServerChannelsTest.java
index 0dd1bbff383..e2b7c89db5f 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/transport/ServerChannelsTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/transport/ServerChannelsTest.java
@@ -154,7 +154,8 @@ public class ServerChannelsTest {
}
verify(mockChannel).close();
-
verify(asyncQueryResponse).markServerDown(any(ServerRoutingInstance.class),
any(Exception.class));
+
+
verify(asyncQueryResponse).markServerUnavailable(any(ServerRoutingInstance.class),
any(Exception.class));
verify(asyncQueryResponse,
never()).markRequestSent(any(ServerRoutingInstance.class), any(Integer.class));
serverChannels.shutDown();
@@ -191,7 +192,7 @@ public class ServerChannelsTest {
}
verify(asyncQueryResponse).markRequestSent(any(ServerRoutingInstance.class),
any(Integer.class));
- verify(asyncQueryResponse,
never()).markServerDown(any(ServerRoutingInstance.class), any(Exception.class));
+ verify(asyncQueryResponse,
never()).markServerUnavailable(any(ServerRoutingInstance.class),
any(Exception.class));
verify(mockChannel, never()).close();
serverChannels.shutDown();
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
index 7cdf8a56398..0ef4615ba20 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
@@ -920,7 +920,10 @@ public class CommonConstants {
// divided across all servers processing the query.
public static final String MAX_QUERY_RESPONSE_SIZE_BYTES =
"maxQueryResponseSizeBytes";
- // If query submission causes an exception, still continue to submit
the query to other servers
+ // If a server is unavailable, still return results from the other
servers instead of failing the query. This
+ // covers both a send-time failure at request submission and a
mid-query channel-inactive / write failure: the
+ // unavailable server is skipped, its down status is recorded so the
failure detector can quarantine it from
+ // routing, and the query returns partial results
public static final String SKIP_UNAVAILABLE_SERVERS =
"skipUnavailableServers";
// Ignore server-side segment missing errors and proceed without
marking the query as failed.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
