xiangfu0 commented on code in PR #19982:
URL: https://github.com/apache/pinot/pull/19982#discussion_r4345938744
##########
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:
A one-line comment here would help: unlike `receiveDataTable`, a
deserialization failure can't be tied to a single `requestId` (the bytes didn't
parse), so this has to broadcast to every in-flight query on the server. That's
consistent with `markServerUnavailable` and safe because the channel is closed
right after, but the reasoning isn't obvious at first glance.
##########
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:
This assertion was weakened from `assertEquals(getSubmitDelayMs(), -1)`, and
it looks unrelated to the deserialization fix. `getSubmitDelayMs()` only ever
returns `-1` or a non-negative value, so `>= -1` is always true and no longer
checks anything. If the exact match started flaking it'd be worth understanding
why rather than loosening it; otherwise consider dropping this line to keep the
PR focused.
##########
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();
Review Comment:
Good that this synthetic table stays broker-side only and is never
serialized over the wire — so the `getEmptyDataTable()` builder version carries
no mixed-version/wire-compat risk here.
##########
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:
Closing the channel here is reasonable, but it now slightly contradicts the
`exceptionCaught` comment below ("deserialization ... does not corrupt the
channel's byte stream"). Could you reconcile the two — e.g. note here why a
parse failure may leave framing desynced and so warrants a close?
##########
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:
Minor / pre-existing: `receiveDataTable` doesn't consult
`_countedDownServers`, so if a genuine response for the same (query, server)
ever arrived after this synthetic path, the latch would be counted down twice.
The `ctx.close()` makes that very unlikely, so not blocking — just noting the
new path widens that window slightly.
--
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]
