xiangfu0 commented on code in PR #19982:
URL: https://github.com/apache/pinot/pull/19982#discussion_r4423019556
##########
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:
**MAJOR (verified race):** `receiveDataTable` doesn't check
`_countedDownServers`, and `ctx.close()` doesn't stop frames that were already
decoded. `ByteToMessageDecoder` keeps firing the rest of the read batch, and
`decodeLast` runs on inactive. So if the bytes arrive as [bad frame][good frame
for Q2], Q2's slot for A is counted down twice, and Q2 completes without
waiting for C. I reproduced this: `COMPLETED` in 0 ms, C=`null`, 435
overwritten. Please guard every countdown with the same set. For example,
return early here when `!_countedDownServers.add(server)`, and add a test for
it.
---
AI-generated review via IntentLab · Model: Claude Opus 5.5 (claude-opus-5-5).
##########
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:
MINOR: please drop the `Issue #18866:` change-history prefixes here and in
`ServerResponse`, `QueryErrorCode` and `Query.tsx`, and describe the current
behavior instead. The `_countedDownServers` doc at L50 also needs to list the
new path.
---
AI-generated review via IntentLab · Model: Claude Opus 5.5 (claude-opus-5-5).
##########
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:
**MAJOR (semantics, needs owner sign-off):** one bad response now ends
*every* in-flight query on this channel with a 435, partial results, and a
quarantined server. This ignores `skipUnavailableServers`. It also counts those
servers as "responded" and suppresses `SERVER_NOT_RESPONDING`, even though most
of them never answered. Before, one query hung. Please confirm this trade-off
is intended and add a test that has a second in-flight query on the same
channel.
---
AI-generated review via IntentLab · Model: Claude Opus 5.5 (claude-opus-5-5).
##########
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:
MINOR: `LengthFieldBasedFrameDecoder` has already consumed exactly one
frame, so a payload that fails to parse can't desync framing. Closing is needed
because the broadcast released the latch slots of other queries, so this
comment should say that instead. Also, this `catch` wraps `receiveDataTable`
and the accountant updates. Consider narrowing it so post-receive failures
don't close a healthy channel.
---
AI-generated review via IntentLab · Model: Claude Opus 5.5 (claude-opus-5-5).
--
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]
