noy-solvin commented on code in PR #19982:
URL: https://github.com/apache/pinot/pull/19982#discussion_r4435199698
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/DataTableHandler.java:
##########
@@ -97,6 +97,15 @@ protected void channelRead0(ChannelHandlerContext ctx,
ByteBuf msg) {
LOGGER.error("Caught exception while deserializing data table of size:
{} from server: {}", responseSize,
_serverRoutingInstance, e);
_brokerMetrics.addMeteredGlobalValue(BrokerMeter.DATA_TABLE_DESERIALIZATION_EXCEPTIONS,
1);
+ // A deserialization parse failure on malformed/corrupted response bytes
may leave framing desynchronized or
Review Comment:
Resolved. We have narrowed the `try-catch` in
`DataTableHandler.channelRead0` to strictly wrap
`DataTableFactory.getDataTable(msg.nioBuffer())`. Post-deserialization
operations (`_queryRouter.receiveDataTable(...)` and thread accountant resource
tracking) now run outside this catch block, ensuring that any post-receive
exceptions do not close a healthy channel or trigger false deserialization
error broadcasts.
Additionally, we updated the comment in `channelRead0` to explain that
channel closure is needed because broadcasting the deserialization error
releases the latch slots of all in-flight queries on this channel, and restored
the comment in `exceptionCaught` regarding framing.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/AsyncQueryResponse.java:
##########
@@ -183,6 +184,26 @@ void markRequestSent(ServerRoutingInstance
serverRoutingInstance, int requestSen
_responseMap.get(serverRoutingInstance).markRequestSent(requestSentLatencyMs);
}
+ // Issue #18866: Handle server DataTable deserialization errors without
waiting for query timeout
Review Comment:
Resolved. We have dropped all `Issue #18866:` change-history comment
prefixes across `AsyncQueryResponse.java`, `ServerResponse.java`,
`QueryErrorCode.java`, and `Query.tsx`, replacing them with descriptions of the
current behavior. We also updated the documentation for `_countedDownServers`
at line 50 of `AsyncQueryResponse.java` to explicitly document all latch slot
release paths (`receiveDataTable`, `receiveDeserializationError`,
`skipServerResponse`, `markServerUnavailable`).
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/QueryRouter.java:
##########
@@ -345,6 +345,17 @@ void receiveDataTable(ServerRoutingInstance
serverRoutingInstance, DataTable dat
}
}
+ // Unlike receiveDataTable, a deserialization failure cannot be tied to a
single requestId (the bytes didn't
+ // parse), so this method must broadcast to every in-flight query on the
server. This is consistent with
+ // markServerUnavailable and safe because the channel is closed immediately
after.
+ void markServerDeserializationError(ServerRoutingInstance
serverRoutingInstance, Exception exception,
+ int responseSize, int deserializationTimeMs) {
+ for (AsyncQueryResponse asyncQueryResponse :
_asyncQueryResponseMap.values()) {
Review Comment:
Confirmed: this trade-off is intentional and designed into the fail-fast
semantics for DataTable deserialization errors. Because response bytes that
fail deserialization cannot be parsed, their `requestId` cannot be extracted
from the table metadata. Since queries multiplex over a single Netty connection
to the server, a corrupted payload indicates an unrecoverable protocol mismatch
or stream corruption on that channel. Broadcasting `DESERIALIZATION_ERROR`
(435) to all in-flight queries on that server allows them to fail fast
immediately with partial results rather than hanging until the broker timeout
expires, and closing the connection resets framing for future requests.
Quarantining the server via `_failedServer` is appropriate because corrupted
response payloads indicate an unhealthy server or serialization incompatibility.
We have added
`QueryRoutingTest#testMultipleInFlightQueriesOnDeserializationError` to
explicitly test that multiple in-flight queries sharing the channel complete
immediately with `DESERIALIZATION_ERROR` (435) and record the failed server
without timing out.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/AsyncQueryResponse.java:
##########
@@ -183,6 +184,26 @@ void markRequestSent(ServerRoutingInstance
serverRoutingInstance, int requestSen
_responseMap.get(serverRoutingInstance).markRequestSent(requestSentLatencyMs);
}
+ // Issue #18866: Handle server DataTable deserialization errors without
waiting for query timeout
+ void receiveDeserializationError(ServerRoutingInstance
serverRoutingInstance, Exception exception,
+ int responseSize, int deserializationTimeMs) {
+ if (!shouldActOnServerDown(serverRoutingInstance)) {
+ return;
+ }
+ if (!_countedDownServers.add(serverRoutingInstance)) {
+ return;
+ }
+ _failedServer = serverRoutingInstance;
+ DataTable dataTable = DataTableBuilderFactory.getEmptyDataTable();
+ dataTable.getMetadata().put(DataTable.MetadataKey.REQUEST_ID.getName(),
Long.toString(_requestId));
+ dataTable.addException(QueryErrorCode.DESERIALIZATION_ERROR.getId(),
+ exception.getMessage() != null ? exception.getMessage() :
exception.toString());
+ ServerResponse response = _responseMap.get(serverRoutingInstance);
+ response.receiveDeserializationError(dataTable, responseSize,
deserializationTimeMs);
+ _numServersResponded.getAndIncrement();
+ _countDownLatch.countDown();
+ }
+
void receiveDataTable(ServerRoutingInstance serverRoutingInstance, DataTable
dataTable, int responseSize,
Review Comment:
Resolved. We have guarded `receiveDataTable` with `if
(!_countedDownServers.add(serverRoutingInstance)) { return; }` before
processing the response or decrementing the countdown latch. This guarantees
that latch slots are released at most once per server, preventing duplicate
decrements when Netty's `ByteToMessageDecoder` continues firing decoded frames
from a read batch after `ctx.close()`.
We also added unit tests in `AsyncQueryResponseTest`
(`testReceiveDeserializationErrorThenReceiveDataTableIsIdempotent` and
`testReceiveDataTableThenReceiveDeserializationErrorIsIdempotent`) to verify
idempotency and ensure latch countdowns and server responses are preserved
correctly.
--
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]
