xiangfu0 commented on code in PR #20023:
URL: https://github.com/apache/pinot/pull/20023#discussion_r4393175867


##########
pinot-core/src/main/java/org/apache/pinot/core/transport/InstanceRequestHandler.java:
##########
@@ -154,16 +171,81 @@ protected void channelRead0(ChannelHandlerContext ctx, 
ByteBuf msg) {
     
queryRequest.getTimerContext().startNewPhaseTimer(ServerQueryPhase.REQUEST_DESERIALIZATION,
 queryArrivalTimeMs)
         .stopAndRecord();
     try {
+      if (rejectIfExpired(queryRequest, ctx, queryArrivalTimeMs)) {
+        return;
+      }
       submitQuery(queryRequest, ctx, queryArrivalTimeMs);
     } catch (Exception e) {
       long requestId = queryRequest.getRequestId();
       String tableNameWithType = queryRequest.getTableNameWithType();
-      LOGGER.error("Caught exception while submitting query request: {}", 
requestId, e);
+      LOGGER.error("Caught exception while handling query request: {}", 
requestId, e);
       sendErrorResponse(ctx, requestId, tableNameWithType, queryArrivalTimeMs,
           DataTableBuilderFactory.getEmptyDataTable(), e);
     }
   }
 
+  /// Answers a request that reached this server after the broker's deadline 
for it with a timeout error, without
+  /// executing it, and returns `true`; returns `false` when the request is to 
be executed as usual.
+  ///
+  /// The broker sends its deadline as the `requestDeadlineMs` query option, 
so a request from an old broker (no
+  /// option) is always executed. The deadline is compared against this 
server's clock, which is why it has to be more
+  /// than [Server#CONFIG_OF_REQUEST_DEADLINE_CLOCK_SKEW_MS] in the past 
before the request is dropped. Without this
+  /// check a request that spent minutes in the socket buffers of a stalled 
connection, and is delivered when the
+  /// connection recovers, is executed with a full timeout budget measured 
from its arrival, for a broker that gave up
+  /// on it long ago. The response is still sent: it costs a few bytes and 
tells the broker side that the channel is
+  /// alive, even though the broker no longer has a query waiting for it. A 
malformed option propagates to the caller,
+  /// which answers with the parse error under the right request id.
+  private boolean rejectIfExpired(ServerQueryRequest queryRequest, 
ChannelHandlerContext ctx,
+      long queryArrivalTimeMs) {
+    if (_requestDeadlineClockSkewMs < 0) {
+      return false;
+    }
+    Long requestDeadlineMs = 
QueryOptionsUtils.getRequestDeadlineMs(queryRequest.getQueryContext().getQueryOptions());
+    if (requestDeadlineMs == null) {
+      return false;
+    }
+    // Subtract rather than add, so that a huge deadline cannot overflow into 
a rejection.
+    long lateByMs = queryArrivalTimeMs - requestDeadlineMs;
+    if (lateByMs <= _requestDeadlineClockSkewMs) {

Review Comment:
   One edge worth a thought: the broker stamps the deadline as `now + 
remainingTimeMs` before dispatch even starts, so for a query with a very small 
`timeoutMs` (say a user sets 200ms), on a perfectly healthy connection the 
request can still land here more than `timeout + skew` late if dispatch and the 
network hop are slow. It would then be dropped even though nothing stalled.
   
   In practice such a query is about to time out at the broker anyway, and the 
1s default skew absorbs normal hops, so I don't think it needs a code change — 
but it might be worth a line in the config doc noting that the check assumes 
`timeoutMs` is comfortably larger than dispatch latency.



##########
pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java:
##########
@@ -307,6 +307,16 @@ public static Long getTimeoutMs(Map<String, String> 
queryOptions) {
     return checkedParseLongPositive(QueryOptionKey.TIMEOUT_MS, 
timeoutMsString);
   }
 
+  /// The broker's deadline for the request 
([QueryOptionKey#REQUEST_DEADLINE_MS]) in epoch milliseconds, or `null`
+  /// when the option is absent or non-positive, in which case the request is 
not checked against it.
+  @Nullable
+  public static Long getRequestDeadlineMs(Map<String, String> queryOptions) {
+    String requestDeadlineMsString = 
queryOptions.get(QueryOptionKey.REQUEST_DEADLINE_MS);
+    Long requestDeadlineMs =
+        checkedParseLong(QueryOptionKey.REQUEST_DEADLINE_MS, 
requestDeadlineMsString, Long.MIN_VALUE);

Review Comment:
   Small readability nit: a non-positive deadline here silently becomes `null` 
(no check), which is the right behavior, but it means `0` and `-1` are treated 
identically to "absent" without any trace. Since the server side already keys 
the whole feature off this returning null, that's fine; just flagging that a 
misconfigured broker sending `0` would be indistinguishable from an old broker 
at the server. No change needed if that's intended.



-- 
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]

Reply via email to