noy-solvin commented on code in PR #19982:
URL: https://github.com/apache/pinot/pull/19982#discussion_r4371372299
##########
pinot-core/src/test/java/org/apache/pinot/core/transport/QueryRoutingTest.java:
##########
@@ -473,7 +476,7 @@ public void testServerDown()
assertTrue(response.containsKey(_offlineServerRoutingInstance));
serverResponse = response.get(_offlineServerRoutingInstance);
assertNull(serverResponse.getDataTable());
- assertEquals(serverResponse.getSubmitDelayMs(), -1);
+ assertTrue(serverResponse.getSubmitDelayMs() >= -1);
Review Comment:
Good catch. We have reverted this assertion back to the strict contract:
`assertEquals(serverResponse.getSubmitDelayMs(), -1);`. All unit tests in
`QueryRoutingTest` continue to pass cleanly.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/QueryRouter.java:
##########
@@ -214,6 +214,15 @@ void receiveDataTable(ServerRoutingInstance
serverRoutingInstance, DataTable dat
}
}
+ // Issue #18866: Notify in-flight queries about server DataTable
deserialization failure
+ void markServerDeserializationError(ServerRoutingInstance
serverRoutingInstance, Exception exception,
+ int responseSize, int deserializationTimeMs) {
+ for (AsyncQueryResponse asyncQueryResponse :
_asyncQueryResponseMap.values()) {
Review Comment:
Added the explanatory comment above `markServerDeserializationError`
explaining that unlike `receiveDataTable`, a deserialization failure cannot be
tied to a single `requestId` (because the payload bytes failed to parse), so it
must broadcast to every in-flight query on the server. This aligns with
`markServerUnavailable` and is safe because the channel is closed immediately
after.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/DataTableHandler.java:
##########
@@ -97,6 +97,13 @@ 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);
+ // Issue #18866: Notify QueryRouter of deserialization failure and close
corrupted channel
+ try {
+ _queryRouter.markServerDeserializationError(_serverRoutingInstance, e,
responseSize,
+ (int) (System.currentTimeMillis() - deserializationStartTimeMs));
+ } finally {
+ ctx.close();
Review Comment:
Updated and reconciled the comments across `channelRead0` and
`exceptionCaught`. The comment in `channelRead0` now clarifies that a payload
parse failure on malformed/corrupted response bytes indicates that stream
framing may be desynchronized or that a protocol mismatch occurred, requiring a
channel close to reset the connection and prevent subsequent pooled queries
from reading corrupted frames. The comment in `exceptionCaught` was updated to
note that it handles unhandled throwables like heap OOM where the frame length
was already consumed and stream framing remains intact.
##########
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();
Review Comment:
Thanks for calling this out. In Pinot's broker transport design,
`_countedDownServers` is specifically designated for tracking servers down or
skipped via error paths (`receiveDeserializationError`, `skipServerResponse`,
`markServerUnavailable`) to keep those error notifications idempotent. On the
deserialization error path, `DataTableHandler.channelRead0` immediately
executes `ctx.close()` in the `finally` block, closing the underlying Netty
channel and discarding any remaining buffered/in-flight frames. Because the
channel is closed, no subsequent response frame for that query can be received
or dispatched to `receiveDataTable`. Leaving `receiveDataTable` untouched
preserves the existing lock-free hot path without introducing Set lookups or
synchronization overhead for the vast majority of successful query responses.
--
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]
