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


##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java:
##########
@@ -2391,6 +2402,30 @@ private long setQueryTimeout(String tableNameWithType, 
Long logicalTableQueryTim
     return remainingTimeMs;
   }
 
+  /// Records an approximate end of this broker's remaining wait budget (now 
plus the remaining time, for a hybrid
+  /// table the longer of the two table types' remaining times) into every 
dispatched request's query options, see
+  /// [QueryOptionKey#REQUEST_DEADLINE_MS]. The absolute deadline lets a 
server tell a request that sat in a stalled
+  /// connection from a live one, instead of granting it a fresh timeout 
budget from its arrival. Always overwritten:
+  /// it is this broker's deadline, whatever the user set.
+  ///
+  /// `AsyncQueryResponse` measures the remaining time from its own start in 
`QueryRouter#submitQuery`, after this
+  /// runs. The server applies a clock-skew allowance; dispatch delays or 
clock differences beyond that allowance
+  /// can still cause a request to be rejected before the broker's actual wait 
ends.
+  private static void setRequestDeadline(long remainingTimeMs, @Nullable 
BrokerRequest offlineBrokerRequest,
+      @Nullable BrokerRequest realtimeBrokerRequest, @Nullable BrokerRequest 
materializedViewBrokerRequest) {
+    String requestDeadlineMs = Long.toString(System.currentTimeMillis() + 
remainingTimeMs);

Review Comment:
   Agreed: the deadline is early by the stamp-to-dispatch gap, which counts 
against the allowance along with any lead of the server clock. Today that gap 
is CPU-only work (server option setup, cancellation tracking, building the 
per-server requests, then `new AsyncQueryResponse`), so it is far under a 
millisecond against the 10 s default. If something slow ever landed there, a 
server could drop requests in the last (gap minus allowance) of the broker 
wait, and the fix would be to stamp inside `QueryRouter#submitQuery` next to 
the `AsyncQueryResponse`. I reworded the `setRequestDeadline` Javadoc to say 
that.



##########
pinot-core/src/main/java/org/apache/pinot/core/transport/InstanceRequestHandler.java:
##########
@@ -158,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;

Review Comment:
   Reworded the comment to state the real effect: an absurdly far deadline only 
makes `lateByMs` very negative, so the request still runs instead of 
overflowing into a rejection. `testHugeDeadlineIsExecuted` pins it with 
`Long.MAX_VALUE`.



##########
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);
+    return requestDeadlineMs != null && requestDeadlineMs > 0 ? 
requestDeadlineMs : null;

Review Comment:
   Yes, intentional. `requestDeadlineMs` is broker-internal, and a non-positive 
value is never a real epoch deadline, so it is treated exactly like an absent 
one (the old-broker path) and the request runs. Only a non-numeric value fails, 
and that is answered with the parse error under the request id. Unlike 
`timeoutMs`, rejecting a non-positive value here would only turn a harmless odd 
value into a failed query, and the inline comment there says so.



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