walterddr commented on code in PR #10791:
URL: https://github.com/apache/pinot/pull/10791#discussion_r1201473967
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/service/QueryServer.java:
##########
@@ -83,28 +88,33 @@ public void shutdown() {
@Override
public void submit(Worker.QueryRequest request,
StreamObserver<Worker.QueryResponse> responseObserver) {
// Deserialize the request
- DistributedStagePlan distributedStagePlan;
+ List<DistributedStagePlan> distributedStagePlans;
Map<String, String> requestMetadataMap;
try {
- distributedStagePlan =
QueryPlanSerDeUtils.deserialize(request.getStagePlan());
+ distributedStagePlans = QueryPlanSerDeUtils.deserialize(request);
requestMetadataMap = request.getMetadataMap();
} catch (Exception e) {
LOGGER.error("Caught exception while deserializing the request: {}",
request, e);
responseObserver.onError(Status.INVALID_ARGUMENT.withDescription("Bad
request").withCause(e).asException());
return;
}
-
- try {
- _queryRunner.processQuery(distributedStagePlan, requestMetadataMap);
- responseObserver.onNext(Worker.QueryResponse.newBuilder()
- .putMetadata(QueryConfig.KEY_OF_SERVER_RESPONSE_STATUS_OK,
"").build());
- responseObserver.onCompleted();
- } catch (Throwable t) {
- responseObserver.onNext(Worker.QueryResponse.newBuilder()
- .putMetadata(QueryConfig.KEY_OF_SERVER_RESPONSE_STATUS_ERROR,
QueryException.getTruncatedStackTrace(t))
- .build());
- responseObserver.onCompleted();
- }
+ distributedStagePlans.forEach(distributedStagePlan -> {
+ _executorService.submit(() -> {
+ try {
+ _queryRunner.processQuery(distributedStagePlan,
requestMetadataMap);
+ responseObserver.onNext(Worker.QueryResponse.newBuilder()
+ .putMetadata(QueryConfig.KEY_OF_SERVER_RESPONSE_STATUS_OK,
"").build());
+ responseObserver.onCompleted();
+ } catch (Throwable t) {
+ responseObserver.onNext(Worker.QueryResponse.newBuilder()
+ .putMetadata(QueryConfig.KEY_OF_SERVER_RESPONSE_STATUS_ERROR,
+ QueryException.getTruncatedStackTrace(t))
+ .build());
+ responseObserver.onCompleted();
Review Comment:
this is not the right way to handle it. we should only return onNext() once.
if we were to submit multiple server request for running then we should wait
until all of them returns then return a single success or any error should
gathered together and return 1 error. this potentially can return multiple
onNext with errors.
I suggest
- short term: we don't catch this and simple let the submission go through
successfully. then return onCompleted once.
- long term: have a fork-join on all submitted futures, return success if
all future return success and error when something is wrong.
--
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: commits-unsubscr...@pinot.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
