{"action":"created","comment":{"url":"https://api.github.com/repos/linkedin/pinot/pulls/comments/91422003","pull_request_review_id":11930413,"id":91422003,"diff_hunk":"@@ -243,383 +239,396 @@ public void validateRequest(BrokerRequest brokerRequest) {\n     }\n   }\n \n-  private BucketingSelection getBucketingSelection(BrokerRequest brokerRequest) {\n-    final Map<SegmentId, ServerInstance> bucketMap = new HashMap<>();\n-    return new BucketingSelection(bucketMap);\n-  }\n-\n   /**\n-   * Main method to process the request. Following lifecycle stages:\n-   * 1. This method will first find the candidate servers to be queried for each set of segments from the routing table\n-   * 2. The second stage will be to select servers for each segment set.\n-   * 3. Scatter-Gather of request\n-   * 4. Gather response from the servers.\n-   * 5. Deserialize the responses and errors.\n-   * 6. Reduce (Merge) the responses. Create a broker response to be returned.\n+   * Main method to process the request.\n+   * <p>Following lifecycle stages:\n+   * <ul>\n+   *   <li>1. Find the candidate servers to be queried for each set of segments from the routing table.</li>\n+   *   <li>2. Select servers for each segment set and scatter request to the servers.</li>\n+   *   <li>3. Gather response from the servers.</li>\n+   *   <li>4. Deserialize the responses.</li>\n+   *   <li>5. Reduce (merge) the responses and create a broker response to be returned.</li>\n+   * </ul>\n    *\n-   * @param request Broker Request to be sent\n-   * @return Broker response\n+   * @param brokerRequest broker request to be processed.\n+   * @param scatterGatherStats scatter-gather statistics.\n+   * @param requestId broker request ID.\n+   * @return broker response.\n    * @throws InterruptedException\n    */\n-  //TODO: Define a broker response class and return\n-  public Object processBrokerRequest(final BrokerRequest request, BucketingSelection overriddenSelection,\n-      final ScatterGatherStats scatterGatherStats, final long requestId) throws InterruptedException {\n-\n-    ResponseType responseType = BrokerResponseFactory.getResponseType(request.getResponseFormat());\n+  @Nonnull\n+  public BrokerResponse processBrokerRequest(@Nonnull BrokerRequest brokerRequest,\n+      @Nonnull ScatterGatherStats scatterGatherStats, long requestId)\n+      throws InterruptedException {\n+    String tableName = brokerRequest.getQuerySource().getTableName();\n+    ResponseType responseType = BrokerResponseFactory.getResponseType(brokerRequest.getResponseFormat());\n     LOGGER.debug(\"Broker Response Type: {}\", responseType.name());\n \n-    if (request.getQuerySource() == null || request.getQuerySource().getTableName() == null) {\n-      LOGGER.info(\"Query contains null table.\");\n-      return new BrokerResponseNative();\n-    }\n-\n-    List<String> matchedTables = getMatchedTables(request);\n-    ReduceService<? extends BrokerResponse> reduceService = _reduceServiceRegistry.get(responseType);\n-\n-    if (matchedTables.size() > 1) {\n-      return processFederatedBrokerRequest(request, reduceService, scatterGatherStats, requestId);\n-    }\n-    if (matchedTables.size() == 1) {\n-      return processSingleTableBrokerRequest(request, reduceService, matchedTables.get(0), scatterGatherStats,\n-          requestId);\n+    List<String> matchedTables = getMatchedTables(tableName);\n+    int numMatchedTables = matchedTables.size();\n+    if (numMatchedTables == 0) {\n+      // No table matches the broker request.\n+      LOGGER.warn(\"No table matches the name: {}\", tableName);\n+      _brokerMetrics.addMeteredTableValue(tableName, BrokerMeter.RESOURCE_MISSING_EXCEPTIONS, 1);\n+      return BrokerResponseFactory.getNoTableHitBrokerResponse(responseType);\n+    } else {\n+      ReduceService reduceService = _reduceServiceRegistry.get(responseType);\n+      // TODO: wire up the customized BucketingSelection.\n+      if (numMatchedTables == 1) {\n+        // Broker request hits one table.\n+        brokerRequest.getQuerySource().setTableName(matchedTables.get(0));\n+        return processSingleTableBrokerRequest(_optimizer.optimize(brokerRequest), reduceService, null,\n+            scatterGatherStats, requestId);\n+      } else {\n+        // Broker request hits multiple tables.\n+        List<BrokerRequest> perTableRequests =\n+            Arrays.asList(getOfflineBrokerRequest(brokerRequest), getRealtimeBrokerRequest(brokerRequest));\n+        return processFederatedBrokerRequest(brokerRequest, reduceService, perTableRequests, null, scatterGatherStats,\n+            requestId);\n+      }\n     }\n-    return BrokerResponseFactory.getNoTableHitBrokerResponse(responseType);\n   }\n \n   /**\n-   * Given a request, will look up routing table to see how many tables are matched there.\n+   * Given a table name, look up routing table and get all tables (OFFLINE/REALTIME) matched.\n    *\n-   * @param request\n-   * @return\n+   * @param tableName table name to be looked up.\n+   * @return list of matched tables.\n    */\n-  private List<String> getMatchedTables(BrokerRequest request) {\n-    List<String> matchedTables = new ArrayList<String>();\n-    String tableName = TableNameBuilder.OFFLINE_TABLE_NAME_BUILDER.forTable(request.getQuerySource().getTableName());\n-    if (_routingTable.routingTableExists(tableName)) {\n-      matchedTables.add(tableName);\n-    }\n-    tableName = TableNameBuilder.REALTIME_TABLE_NAME_BUILDER.forTable(request.getQuerySource().getTableName());\n-    if (_routingTable.routingTableExists(tableName)) {\n-      matchedTables.add(tableName);\n-    }\n-    // For backward compatible\n+  @Nonnull\n+  private List<String> getMatchedTables(@Nonnull String tableName) {\n+    List<String> matchedTables = new ArrayList<>();\n+    String offlineTableName = TableNameBuilder.OFFLINE_TABLE_NAME_BUILDER.forTable(tableName);\n+    if (_routingTable.routingTableExists(offlineTableName)) {\n+      matchedTables.add(offlineTableName);\n+    }\n+    String realtimeTableName = TableNameBuilder.REALTIME_TABLE_NAME_BUILDER.forTable(tableName);\n+    if (_routingTable.routingTableExists(realtimeTableName)) {\n+      matchedTables.add(realtimeTableName);\n+    }\n+    // For backward compatible.\n     if (matchedTables.isEmpty()) {\n-      tableName = request.getQuerySource().getTableName();\n       if (_routingTable.routingTableExists(tableName)) {\n         matchedTables.add(tableName);\n       }\n     }\n     return matchedTables;\n   }\n \n-  private Object processSingleTableBrokerRequest(final BrokerRequest request, ReduceService reduceService,\n-      String matchedTableName, final ScatterGatherStats scatterGatherStats, final long requestId)\n-      throws InterruptedException {\n-    request.getQuerySource().setTableName(matchedTableName);\n-    return getDataTableFromBrokerRequest(_optimizer.optimize(request), reduceService, null, scatterGatherStats,\n-        requestId);\n-  }\n-\n-  private Object processFederatedBrokerRequest(final BrokerRequest request, ReduceService reduceService,\n-      final ScatterGatherStats scatterGatherStats, final long requestId) {\n-    List<BrokerRequest> perTableRequests = new ArrayList<BrokerRequest>();\n-    perTableRequests.add(getRealtimeBrokerRequest(request));\n-    perTableRequests.add(getOfflineBrokerRequest(request));\n-    try {\n-      return getDataTableFromBrokerRequestList(request, reduceService, perTableRequests, null, scatterGatherStats,\n-          requestId);\n-    } catch (Exception e) {\n-      LOGGER.error(\"Caught exception while processing federated broker request\", e);\n-      Utils.rethrowException(e);\n-      throw new AssertionError(\"Should not reach this\");\n-    }\n-  }\n-\n-  private BrokerRequest getOfflineBrokerRequest(BrokerRequest request) {\n-    BrokerRequest offlineRequest = request.deepCopy();\n-    String hybridTableName = request.getQuerySource().getTableName();\n+  /**\n+   * Given a broker request, use it to create an offline broker request.\n+   *\n+   * @param brokerRequest original broker request.\n+   * @return offline broker request.\n+   */\n+  @Nonnull\n+  private BrokerRequest getOfflineBrokerRequest(@Nonnull BrokerRequest brokerRequest) {\n+    BrokerRequest offlineRequest = brokerRequest.deepCopy();\n+    String hybridTableName = brokerRequest.getQuerySource().getTableName();\n     String offlineTableName = TableNameBuilder.OFFLINE_TABLE_NAME_BUILDER.forTable(hybridTableName);\n     offlineRequest.getQuerySource().setTableName(offlineTableName);\n     attachTimeBoundary(hybridTableName, offlineRequest, true);\n     return _optimizer.optimize(offlineRequest);\n   }\n \n-  private BrokerRequest getRealtimeBrokerRequest(BrokerRequest request) {\n-    BrokerRequest realtimeRequest = request.deepCopy();\n-    String hybridTableName = request.getQuerySource().getTableName();\n+  /**\n+   * Given a broker request, use it to create a realtime broker request.\n+   *\n+   * @param brokerRequest original broker request.\n+   * @return realtime broker request.\n+   */\n+  @Nonnull\n+  private BrokerRequest getRealtimeBrokerRequest(@Nonnull BrokerRequest brokerRequest) {\n+    BrokerRequest realtimeRequest = brokerRequest.deepCopy();\n+    String hybridTableName = brokerRequest.getQuerySource().getTableName();\n     String realtimeTableName = TableNameBuilder.REALTIME_TABLE_NAME_BUILDER.forTable(hybridTableName);\n     realtimeRequest.getQuerySource().setTableName(realtimeTableName);\n     attachTimeBoundary(hybridTableName, realtimeRequest, false);\n     return _optimizer.optimize(realtimeRequest);\n   }\n \n-  private void attachTimeBoundary(String hybridTableName, BrokerRequest offlineRequest, boolean isOfflineRequest) {\n-    TimeBoundaryInfo timeBoundaryInfo = _timeBoundaryService\n-        .getTimeBoundaryInfoFor(TableNameBuilder.OFFLINE_TABLE_NAME_BUILDER.forTable(hybridTableName));\n+  /**\n+   * Attach time boundary to a broker request.\n+   *\n+   * @param hybridTableName hybrid table name.\n+   * @param brokerRequest original broker request.\n+   * @param isOfflineRequest flag for offline/realtime request.\n+   */\n+  private void attachTimeBoundary(@Nonnull String hybridTableName, @Nonnull BrokerRequest brokerRequest,\n+      boolean isOfflineRequest) {\n+    TimeBoundaryInfo timeBoundaryInfo = _timeBoundaryService.getTimeBoundaryInfoFor(\n+        TableNameBuilder.OFFLINE_TABLE_NAME_BUILDER.forTable(hybridTableName));\n     if (timeBoundaryInfo == null || timeBoundaryInfo.getTimeColumn() == null\n         || timeBoundaryInfo.getTimeValue() == null) {\n+      LOGGER.warn(\"No time boundary attached for table: {}\", hybridTableName);\n       return;\n     }\n+\n+    // Create a range filter based on the request type.\n+    String timeValue = timeBoundaryInfo.getTimeValue();\n     FilterQuery timeFilterQuery = new FilterQuery();\n     timeFilterQuery.setOperator(FilterOperator.RANGE);\n     timeFilterQuery.setColumn(timeBoundaryInfo.getTimeColumn());\n     timeFilterQuery.setNestedFilterQueryIds(new ArrayList<Integer>());\n-    List<String> values = new ArrayList<String>();\n+    List<String> values = new ArrayList<>();\n     if (isOfflineRequest) {\n-      values.add(\"(*\\t\\t\" + timeBoundaryInfo.getTimeValue() + \")\");\n+      values.add(\"(*\\t\\t\" + timeValue + \")\");\n     } else {\n-      values.add(\"[\" + timeBoundaryInfo.getTimeValue() + \"\\t\\t*)\");\n+      values.add(\"[\" + timeValue + \"\\t\\t*)\");\n     }\n     timeFilterQuery.setValue(values);\n     timeFilterQuery.setId(-1);\n-    FilterQuery currentFilterQuery = offlineRequest.getFilterQuery();\n+\n+    // Attach the range filter to the current filter.\n+    FilterQuery currentFilterQuery = brokerRequest.getFilterQuery();\n     if (currentFilterQuery != null) {\n       FilterQuery andFilterQuery = new FilterQuery();\n       andFilterQuery.setOperator(FilterOperator.AND);\n-      List<Integer> nestedFilterQueryIds = new ArrayList<Integer>();\n+      List<Integer> nestedFilterQueryIds = new ArrayList<>();\n       nestedFilterQueryIds.add(currentFilterQuery.getId());\n       nestedFilterQueryIds.add(timeFilterQuery.getId());\n       andFilterQuery.setNestedFilterQueryIds(nestedFilterQueryIds);\n       andFilterQuery.setId(-2);\n-      FilterQueryMap filterSubQueryMap = offlineRequest.getFilterSubQueryMap();\n+      FilterQueryMap filterSubQueryMap = brokerRequest.getFilterSubQueryMap();\n       filterSubQueryMap.putToFilterQueryMap(timeFilterQuery.getId(), timeFilterQuery);\n       filterSubQueryMap.putToFilterQueryMap(andFilterQuery.getId(), andFilterQuery);\n-\n-      offlineRequest.setFilterQuery(andFilterQuery);\n-      offlineRequest.setFilterSubQueryMap(filterSubQueryMap);\n+      brokerRequest.setFilterQuery(andFilterQuery);\n+      brokerRequest.setFilterSubQueryMap(filterSubQueryMap);\n     } else {\n       FilterQueryMap filterSubQueryMap = new FilterQueryMap();\n       filterSubQueryMap.putToFilterQueryMap(timeFilterQuery.getId(), timeFilterQuery);\n-      offlineRequest.setFilterQuery(timeFilterQuery);\n-      offlineRequest.setFilterSubQueryMap(filterSubQueryMap);\n+      brokerRequest.setFilterQuery(timeFilterQuery);\n+      brokerRequest.setFilterSubQueryMap(filterSubQueryMap);\n     }\n   }\n \n-  private Object getDataTableFromBrokerRequest(final BrokerRequest request, final ReduceService reduceService,\n-      BucketingSelection overriddenSelection, final ScatterGatherStats scatterGatherStats, final long requestId)\n+  /**\n+   * Process single broker request on single table.\n+   *\n+   * @param brokerRequest broker request with modified table name.\n+   * @param reduceService reduce service.\n+   * @param bucketingSelection customized bucketing selection.\n+   * @param scatterGatherStats scatter-gather statistics.\n+   * @param requestId request ID.\n+   * @return broker response.\n+   * @throws InterruptedException\n+   */\n+  @Nonnull\n+  private BrokerResponse processSingleTableBrokerRequest(@Nonnull BrokerRequest brokerRequest,\n+      @Nonnull ReduceService reduceService, @Nullable BucketingSelection bucketingSelection,\n+      @Nonnull ScatterGatherStats scatterGatherStats, long requestId)\n       throws InterruptedException {\n-    // Step1\n-    final long routingStartTime = System.nanoTime();\n-    RoutingTableLookupRequest rtRequest = new RoutingTableLookupRequest(request.getQuerySource().getTableName(),\n-        extractRoutingOptionsFromBrokerRequest(request));\n-    Map<ServerInstance, SegmentIdSet> segmentServices = _routingTable.findServers(rtRequest);\n+    String tableName = brokerRequest.getQuerySource().getTableName();\n+    ResponseType responseType = BrokerResponseFactory.getResponseType(brokerRequest.getResponseFormat());\n+\n+    // Step 1: find the candidate servers to be queried for each set of segments from the routing table.\n+    // TODO: add checks for whether all segments are covered.\n+    long routingStartTime = System.nanoTime();\n+    Map<ServerInstance, SegmentIdSet> segmentServices = findCandidateServers(brokerRequest);\n     if (segmentServices == null || segmentServices.isEmpty()) {\n-      LOGGER.warn(\"Not found ServerInstances to Segments Mapping:\");\n-      ResponseType responseType = BrokerResponseFactory.getResponseType(request.getResponseFormat());\n+      LOGGER.warn(\"No server found for table: {}\", tableName);\n+      _brokerMetrics.addMeteredTableValue(tableName, BrokerMeter.NO_SERVER_FOUND_EXCEPTIONS, 1);\n       return BrokerResponseFactory.getEmptyBrokerResponse(responseType);\n     }\n+    _brokerMetrics.addPhaseTiming(tableName, BrokerQueryPhase.QUERY_ROUTING, System.nanoTime() - routingStartTime);\n \n-    final long queryRoutingTime = System.nanoTime() - routingStartTime;\n-    _brokerMetrics.addPhaseTiming(request, BrokerQueryPhase.QUERY_ROUTING, queryRoutingTime);\n-\n-    // Step 2-4\n-    final long scatterGatherStartTime = System.nanoTime();\n-    ScatterGatherRequestImpl scatterRequest = new ScatterGatherRequestImpl(request, segmentServices, _replicaSelection,\n-        ReplicaSelectionGranularity.SEGMENT_ID_SET, request.getBucketHashKey(), 0,\n-        //TODO: Speculative Requests not yet supported\n-        overriddenSelection, requestId, _brokerTimeOutMs, _brokerId);\n-    CompositeFuture<ServerInstance, ByteBuf> response =\n+    // Step 2: select servers for each segment set and scatter request to the servers.\n+    long scatterStartTime = System.nanoTime();\n+    ScatterGatherRequestImpl scatterRequest =\n+        new ScatterGatherRequestImpl(brokerRequest, segmentServices, _replicaSelection,\n+            ReplicaSelectionGranularity.SEGMENT_ID_SET, brokerRequest.getBucketHashKey(), 0, bucketingSelection,\n+            requestId, _brokerTimeOutMs, _brokerId);\n+    CompositeFuture<ServerInstance, ByteBuf> compositeFuture =\n         _scatterGatherer.scatterGather(scatterRequest, scatterGatherStats, _brokerMetrics);\n \n-    //Step 5 - Deserialize Responses and build instance response map\n-    final Map<ServerInstance, DataTable> instanceResponseMap = new HashMap<ServerInstance, DataTable>();\n-    {\n-      Map<ServerInstance, ByteBuf> responses = null;\n-      try {\n-        responses = response.get();\n-        // The call above should have waited for all the responses to come in.\n-        Map<String, Long> responseTimes = response.getResponseTimes();\n-        scatterGatherStats.setResponseTimeMillis(responseTimes);\n-      } catch (ExecutionException e) {\n-        LOGGER.warn(\"Caught exception while fetching response\", e);\n-        _brokerMetrics.addMeteredQueryValue(request, BrokerMeter.REQUEST_FETCH_EXCEPTIONS, 1);\n-      }\n-\n-      final long scatterGatherTime = System.nanoTime() - scatterGatherStartTime;\n-      _brokerMetrics.addPhaseTiming(request, BrokerQueryPhase.SCATTER_GATHER, scatterGatherTime);\n-\n-      final long deserializationStartTime = System.nanoTime();\n-\n-      Map<ServerInstance, Throwable> errors = response.getError();\n-\n-      if (null != responses) {\n-        for (Entry<ServerInstance, ByteBuf> e : responses.entrySet()) {\n-          try {\n-            ByteBuf b = e.getValue();\n-            byte[] b2 = new byte[b.readableBytes()];\n-            if (b2 == null || b2.length == 0) {\n-              continue;\n-            }\n-            b.readBytes(b2);\n-            DataTable r2 = new DataTable(b2);\n-            if (errors != null && errors.containsKey(e.getKey())) {\n-              Throwable throwable = errors.get(e.getKey());\n-              r2.getMetadata().put(DataTable.EXCEPTION_METADATA_KEY, new RequestProcessingException(throwable).toString());\n-              _brokerMetrics.addMeteredQueryValue(request, BrokerMeter.REQUEST_FETCH_EXCEPTIONS, 1);\n-            }\n-            instanceResponseMap.put(e.getKey(), r2);\n-          } catch (Exception ex) {\n-            LOGGER.error(\n-                \"Got exceptions in collect query result for instance \" + e.getKey() + \", error: \" + ex.getMessage(),\n-                ex);\n-            _brokerMetrics.addMeteredQueryValue(request, BrokerMeter.REQUEST_DESERIALIZATION_EXCEPTIONS, 1);\n-          }\n-        }\n-      }\n-      final long deserializationTime = System.nanoTime() - deserializationStartTime;\n-      _brokerMetrics.addPhaseTiming(request, BrokerQueryPhase.DESERIALIZATION, deserializationTime);\n-    }\n-\n-    // Step 6 : Do the reduce and return\n+    // Step 3: gather response from the servers.\n+    Map<ServerInstance, ByteBuf> responseMap;\n     try {\n-      return _brokerMetrics.timeQueryPhase(request, BrokerQueryPhase.REDUCE, new Callable<BrokerResponse>() {\n-        @Override\n-        public BrokerResponse call() {\n-          BrokerResponse returnValue = reduceService.reduceOnDataTable(request, instanceResponseMap);\n-          _brokerMetrics.addMeteredQueryValue(request, BrokerMeter.DOCUMENTS_SCANNED, returnValue.getNumDocsScanned());\n-          return returnValue;\n-        }\n-      });\n+      responseMap = compositeFuture.get();\n+      Map<String, Long> responseTimes = compositeFuture.getResponseTimes();\n+      scatterGatherStats.setResponseTimeMillis(responseTimes);\n     } catch (Exception e) {\n-      // Shouldn't happen, this is only here because timeQueryPhase() can throw a checked exception, even though the nested callable can't.\n-      LOGGER.error(\"Caught exception while processing return\", e);\n-      Utils.rethrowException(e);\n-      throw new AssertionError(\"Should not reach this\");\n-    }\n+      LOGGER.error(\"Caught exception while fetching responses.\", e);\n+      _brokerMetrics.addMeteredTableValue(tableName, BrokerMeter.RESPONSE_FETCH_EXCEPTIONS, 1);\n+      return BrokerResponseFactory.getBrokerResponseWithException(responseType,\n+          QueryException.getException(QueryException.BROKER_GATHER_ERROR, e));\n+    }\n+    _brokerMetrics.addPhaseTiming(tableName, BrokerQueryPhase.SCATTER_GATHER, System.nanoTime() - scatterStartTime);\n+\n+    //Step 4: deserialize the responses.\n+    long deserializationStartTime = System.nanoTime();\n+    Map<ServerInstance, DataTable> dataTableMap = new HashMap<>();\n+    List<ProcessingException> processingExceptions = new ArrayList<>();\n+    deserializeResponses(tableName, responseMap, -1, dataTableMap, processingExceptions);\n+    _brokerMetrics.addPhaseTiming(tableName, BrokerQueryPhase.DESERIALIZATION,\n+        System.nanoTime() - deserializationStartTime);\n+\n+    // Step 5: reduce (merge) the responses and create a broker response to be returned.\n+    long reduceStartTime = System.nanoTime();\n+    BrokerResponse brokerResponse = reduceService.reduceOnDataTable(brokerRequest, dataTableMap, _brokerMetrics);\n+    _brokerMetrics.addPhaseTiming(tableName, BrokerQueryPhase.REDUCE, System.nanoTime() - reduceStartTime);\n+\n+    // Attach processing exceptions and return.\n+    brokerResponse.setExceptions(processingExceptions);\n+    return brokerResponse;\n   }\n \n-  private Object getDataTableFromBrokerRequestList(final BrokerRequest federatedBrokerRequest,\n-      final ReduceService reduceService, final List<BrokerRequest> requests, BucketingSelection overriddenSelection,\n-      final ScatterGatherStats scatterGatherStats, final long requestId)\n+  /**\n+   * Process multiple broker requests on multiple tables.","path":"pinot-transport/src/main/java/com/linkedin/pinot/requestHandler/BrokerRequestHandler.java","position":706,"original_position":706,"commit_id":"dc6666e37cdb614ec855a19ee730d028a92c542f","original_commit_id":"dc6666e37cdb614ec855a19ee730d028a92c542f","user":{"login":"Jackie-Jiang","id":17555551,"avatar_url":"https://avatars.githubusercontent.com/u/17555551?v=3","gravatar_id":"","url":"https://api.github.com/users/Jackie-Jiang","html_url":"https://github.com/Jackie-Jiang","followers_url":"https://api.github.com/users/Jackie-Jiang/followers","following_url":"https://api.github.com/users/Jackie-Jiang/following{/other_user}","gists_url":"https://api.github.com/users/Jackie-Jiang/gists{/gist_id}","starred_url":"https://api.github.com/users/Jackie-Jiang/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/Jackie-Jiang/subscriptions","organizations_url":"https://api.github.com/users/Jackie-Jiang/orgs","repos_url":"https://api.github.com/users/Jackie-Jiang/repos","events_url":"https://api.github.com/users/Jackie-Jiang/events{/privacy}","received_events_url":"https://api.github.com/users/Jackie-Jiang/received_events","type":"User","site_admin":false},"body":"done.","created_at":"2016-12-08T00:52:18Z","updated_at":"2016-12-08T00:52:18Z","html_url":"https://github.com/linkedin/pinot/pull/868#discussion_r91422003","pull_request_url":"https://api.github.com/repos/linkedin/pinot/pulls/868","_links":{"self":{"href":"https://api.github.com/repos/linkedin/pinot/pulls/comments/91422003"},"html":{"href":"https://github.com/linkedin/pinot/pull/868#discussion_r91422003"},"pull_request":{"href":"https://api.github.com/repos/linkedin/pinot/pulls/868"}}},"pull_request":{"url":"https://api.github.com/repos/linkedin/pinot/pulls/868","id":96017350,"html_url":"https://github.com/linkedin/pinot/pull/868","diff_url":"https://github.com/linkedin/pinot/pull/868.diff","patch_url":"https://github.com/linkedin/pinot/pull/868.patch","issue_url":"https://api.github.com/repos/linkedin/pinot/issues/868","number":868,"state":"open","locked":false,"title":"Enhance broker side exception handling.","user":{"login":"Jackie-Jiang","id":17555551,"avatar_url":"https://avatars.githubusercontent.com/u/17555551?v=3","gravatar_id":"","url":"https://api.github.com/users/Jackie-Jiang","html_url":"https://github.com/Jackie-Jiang","followers_url":"https://api.github.com/users/Jackie-Jiang/followers","following_url":"https://api.github.com/users/Jackie-Jiang/following{/other_user}","gists_url":"https://api.github.com/users/Jackie-Jiang/gists{/gist_id}","starred_url":"https://api.github.com/users/Jackie-Jiang/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/Jackie-Jiang/subscriptions","organizations_url":"https://api.github.com/users/Jackie-Jiang/orgs","repos_url":"https://api.github.com/users/Jackie-Jiang/repos","events_url":"https://api.github.com/users/Jackie-Jiang/events{/privacy}","received_events_url":"https://api.github.com/users/Jackie-Jiang/received_events","type":"User","site_admin":false},"body":"1. Refactor BrokerRequestHandler, catch more exceptions and put exceptions into query response.\r\n2. Refactor BrokerReduceService, add schema-mismatch information into query response.\r\n3. Fix server send zero length response.\r\n\r\nBroker side exception includes:\r\n- PQL compile exception\r\n- Query validation exception\r\n- Scatter-gather exception\r\n- Response deserialize exception\r\n- Merge response exception (schema mismatch)","created_at":"2016-12-01T02:54:18Z","updated_at":"2016-12-08T00:52:18Z","closed_at":null,"merged_at":null,"merge_commit_sha":"2b8d84aee0144c5bc4eb04869a2283f2419e0d01","assignee":null,"assignees":[],"milestone":null,"commits_url":"https://api.github.com/repos/linkedin/pinot/pulls/868/commits","review_comments_url":"https://api.github.com/repos/linkedin/pinot/pulls/868/comments","review_comment_url":"https://api.github.com/repos/linkedin/pinot/pulls/comments{/number}","comments_url":"https://api.github.com/repos/linkedin/pinot/issues/868/comments","statuses_url":"https://api.github.com/repos/linkedin/pinot/statuses/dc6666e37cdb614ec855a19ee730d028a92c542f","head":{"label":"linkedin:clean_up_broker_request_handler","ref":"clean_up_broker_request_handler","sha":"dc6666e37cdb614ec855a19ee730d028a92c542f","user":{"login":"linkedin","id":357098,"avatar_url":"https://avatars.githubusercontent.com/u/357098?v=3","gravatar_id":"","url":"https://api.github.com/users/linkedin","html_url":"https://github.com/linkedin","followers_url":"https://api.github.com/users/linkedin/followers","following_url":"https://api.github.com/users/linkedin/following{/other_user}","gists_url":"https://api.github.com/users/linkedin/gists{/gist_id}","starred_url":"https://api.github.com/users/linkedin/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/linkedin/subscriptions","organizations_url":"https://api.github.com/users/linkedin/orgs","repos_url":"https://api.github.com/users/linkedin/repos","events_url":"https://api.github.com/users/linkedin/events{/privacy}","received_events_url":"https://api.github.com/users/linkedin/received_events","type":"Organization","site_admin":false},"repo":{"id":19961085,"name":"pinot","full_name":"linkedin/pinot","owner":{"login":"linkedin","id":357098,"avatar_url":"https://avatars.githubusercontent.com/u/357098?v=3","gravatar_id":"","url":"https://api.github.com/users/linkedin","html_url":"https://github.com/linkedin","followers_url":"https://api.github.com/users/linkedin/followers","following_url":"https://api.github.com/users/linkedin/following{/other_user}","gists_url":"https://api.github.com/users/linkedin/gists{/gist_id}","starred_url":"https://api.github.com/users/linkedin/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/linkedin/subscriptions","organizations_url":"https://api.github.com/users/linkedin/orgs","repos_url":"https://api.github.com/users/linkedin/repos","events_url":"https://api.github.com/users/linkedin/events{/privacy}","received_events_url":"https://api.github.com/users/linkedin/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/linkedin/pinot","description":"A realtime distributed OLAP datastore","fork":false,"url":"https://api.github.com/repos/linkedin/pinot","forks_url":"https://api.github.com/repos/linkedin/pinot/forks","keys_url":"https://api.github.com/repos/linkedin/pinot/keys{/key_id}","collaborators_url":"https://api.github.com/repos/linkedin/pinot/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/linkedin/pinot/teams","hooks_url":"https://api.github.com/repos/linkedin/pinot/hooks","issue_events_url":"https://api.github.com/repos/linkedin/pinot/issues/events{/number}","events_url":"https://api.github.com/repos/linkedin/pinot/events","assignees_url":"https://api.github.com/repos/linkedin/pinot/assignees{/user}","branches_url":"https://api.github.com/repos/linkedin/pinot/branches{/branch}","tags_url":"https://api.github.com/repos/linkedin/pinot/tags","blobs_url":"https://api.github.com/repos/linkedin/pinot/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/linkedin/pinot/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/linkedin/pinot/git/refs{/sha}","trees_url":"https://api.github.com/repos/linkedin/pinot/git/trees{/sha}","statuses_url":"https://api.github.com/repos/linkedin/pinot/statuses/{sha}","languages_url":"https://api.github.com/repos/linkedin/pinot/languages","stargazers_url":"https://api.github.com/repos/linkedin/pinot/stargazers","contributors_url":"https://api.github.com/repos/linkedin/pinot/contributors","subscribers_url":"https://api.github.com/repos/linkedin/pinot/subscribers","subscription_url":"https://api.github.com/repos/linkedin/pinot/subscription","commits_url":"https://api.github.com/repos/linkedin/pinot/commits{/sha}","git_commits_url":"https://api.github.com/repos/linkedin/pinot/git/commits{/sha}","comments_url":"https://api.github.com/repos/linkedin/pinot/comments{/number}","issue_comment_url":"https://api.github.com/repos/linkedin/pinot/issues/comments{/number}","contents_url":"https://api.github.com/repos/linkedin/pinot/contents/{+path}","compare_url":"https://api.github.com/repos/linkedin/pinot/compare/{base}...{head}","merges_url":"https://api.github.com/repos/linkedin/pinot/merges","archive_url":"https://api.github.com/repos/linkedin/pinot/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/linkedin/pinot/downloads","issues_url":"https://api.github.com/repos/linkedin/pinot/issues{/number}","pulls_url":"https://api.github.com/repos/linkedin/pinot/pulls{/number}","milestones_url":"https://api.github.com/repos/linkedin/pinot/milestones{/number}","notifications_url":"https://api.github.com/repos/linkedin/pinot/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/linkedin/pinot/labels{/name}","releases_url":"https://api.github.com/repos/linkedin/pinot/releases{/id}","deployments_url":"https://api.github.com/repos/linkedin/pinot/deployments","created_at":"2014-05-19T23:27:48Z","updated_at":"2016-12-07T04:01:32Z","pushed_at":"2016-12-08T00:41:17Z","git_url":"git://github.com/linkedin/pinot.git","ssh_url":"git@github.com:linkedin/pinot.git","clone_url":"https://github.com/linkedin/pinot.git","svn_url":"https://github.com/linkedin/pinot","homepage":"https://github.com/linkedin/pinot/wiki","size":148453,"stargazers_count":1377,"watchers_count":1377,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":237,"mirror_url":null,"open_issues_count":89,"forks":237,"open_issues":89,"watchers":1377,"default_branch":"master"}},"base":{"label":"linkedin:master","ref":"master","sha":"381c11a8585651d27f82242664b52e5b3f90994b","user":{"login":"linkedin","id":357098,"avatar_url":"https://avatars.githubusercontent.com/u/357098?v=3","gravatar_id":"","url":"https://api.github.com/users/linkedin","html_url":"https://github.com/linkedin","followers_url":"https://api.github.com/users/linkedin/followers","following_url":"https://api.github.com/users/linkedin/following{/other_user}","gists_url":"https://api.github.com/users/linkedin/gists{/gist_id}","starred_url":"https://api.github.com/users/linkedin/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/linkedin/subscriptions","organizations_url":"https://api.github.com/users/linkedin/orgs","repos_url":"https://api.github.com/users/linkedin/repos","events_url":"https://api.github.com/users/linkedin/events{/privacy}","received_events_url":"https://api.github.com/users/linkedin/received_events","type":"Organization","site_admin":false},"repo":{"id":19961085,"name":"pinot","full_name":"linkedin/pinot","owner":{"login":"linkedin","id":357098,"avatar_url":"https://avatars.githubusercontent.com/u/357098?v=3","gravatar_id":"","url":"https://api.github.com/users/linkedin","html_url":"https://github.com/linkedin","followers_url":"https://api.github.com/users/linkedin/followers","following_url":"https://api.github.com/users/linkedin/following{/other_user}","gists_url":"https://api.github.com/users/linkedin/gists{/gist_id}","starred_url":"https://api.github.com/users/linkedin/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/linkedin/subscriptions","organizations_url":"https://api.github.com/users/linkedin/orgs","repos_url":"https://api.github.com/users/linkedin/repos","events_url":"https://api.github.com/users/linkedin/events{/privacy}","received_events_url":"https://api.github.com/users/linkedin/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/linkedin/pinot","description":"A realtime distributed OLAP datastore","fork":false,"url":"https://api.github.com/repos/linkedin/pinot","forks_url":"https://api.github.com/repos/linkedin/pinot/forks","keys_url":"https://api.github.com/repos/linkedin/pinot/keys{/key_id}","collaborators_url":"https://api.github.com/repos/linkedin/pinot/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/linkedin/pinot/teams","hooks_url":"https://api.github.com/repos/linkedin/pinot/hooks","issue_events_url":"https://api.github.com/repos/linkedin/pinot/issues/events{/number}","events_url":"https://api.github.com/repos/linkedin/pinot/events","assignees_url":"https://api.github.com/repos/linkedin/pinot/assignees{/user}","branches_url":"https://api.github.com/repos/linkedin/pinot/branches{/branch}","tags_url":"https://api.github.com/repos/linkedin/pinot/tags","blobs_url":"https://api.github.com/repos/linkedin/pinot/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/linkedin/pinot/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/linkedin/pinot/git/refs{/sha}","trees_url":"https://api.github.com/repos/linkedin/pinot/git/trees{/sha}","statuses_url":"https://api.github.com/repos/linkedin/pinot/statuses/{sha}","languages_url":"https://api.github.com/repos/linkedin/pinot/languages","stargazers_url":"https://api.github.com/repos/linkedin/pinot/stargazers","contributors_url":"https://api.github.com/repos/linkedin/pinot/contributors","subscribers_url":"https://api.github.com/repos/linkedin/pinot/subscribers","subscription_url":"https://api.github.com/repos/linkedin/pinot/subscription","commits_url":"https://api.github.com/repos/linkedin/pinot/commits{/sha}","git_commits_url":"https://api.github.com/repos/linkedin/pinot/git/commits{/sha}","comments_url":"https://api.github.com/repos/linkedin/pinot/comments{/number}","issue_comment_url":"https://api.github.com/repos/linkedin/pinot/issues/comments{/number}","contents_url":"https://api.github.com/repos/linkedin/pinot/contents/{+path}","compare_url":"https://api.github.com/repos/linkedin/pinot/compare/{base}...{head}","merges_url":"https://api.github.com/repos/linkedin/pinot/merges","archive_url":"https://api.github.com/repos/linkedin/pinot/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/linkedin/pinot/downloads","issues_url":"https://api.github.com/repos/linkedin/pinot/issues{/number}","pulls_url":"https://api.github.com/repos/linkedin/pinot/pulls{/number}","milestones_url":"https://api.github.com/repos/linkedin/pinot/milestones{/number}","notifications_url":"https://api.github.com/repos/linkedin/pinot/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/linkedin/pinot/labels{/name}","releases_url":"https://api.github.com/repos/linkedin/pinot/releases{/id}","deployments_url":"https://api.github.com/repos/linkedin/pinot/deployments","created_at":"2014-05-19T23:27:48Z","updated_at":"2016-12-07T04:01:32Z","pushed_at":"2016-12-08T00:41:17Z","git_url":"git://github.com/linkedin/pinot.git","ssh_url":"git@github.com:linkedin/pinot.git","clone_url":"https://github.com/linkedin/pinot.git","svn_url":"https://github.com/linkedin/pinot","homepage":"https://github.com/linkedin/pinot/wiki","size":148453,"stargazers_count":1377,"watchers_count":1377,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":237,"mirror_url":null,"open_issues_count":89,"forks":237,"open_issues":89,"watchers":1377,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/linkedin/pinot/pulls/868"},"html":{"href":"https://github.com/linkedin/pinot/pull/868"},"issue":{"href":"https://api.github.com/repos/linkedin/pinot/issues/868"},"comments":{"href":"https://api.github.com/repos/linkedin/pinot/issues/868/comments"},"review_comments":{"href":"https://api.github.com/repos/linkedin/pinot/pulls/868/comments"},"review_comment":{"href":"https://api.github.com/repos/linkedin/pinot/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/linkedin/pinot/pulls/868/commits"},"statuses":{"href":"https://api.github.com/repos/linkedin/pinot/statuses/dc6666e37cdb614ec855a19ee730d028a92c542f"}}}}
{"actor":{"display_login":"Jackie-Jiang"}}