@VisibleForTesting
  static void processIterativeServerResponse(StreamingReducer reducer, ExecutorService executorService,
      Map<ServerRoutingInstance, Iterator<Server.ServerResponse>> serverResponseMap, long reduceTimeOutMs,
      ExecutionStatsAggregator aggregator)
      throws Exception {
    int cnt = 0;
    CompletableFuture<Void>[] futures = new CompletableFuture[serverResponseMap.size()];
    // based on ideas from on https://stackoverflow.com/questions/19348248/waiting-on-a-list-of-future
    // and https://stackoverflow.com/questions/23301598/transform-java-future-into-a-completablefuture
    // Future created via ExecutorService.submit() can be created by CompletableFuture.supplyAsync()
    for (Map.Entry<ServerRoutingInstance, Iterator<Server.ServerResponse>> entry : serverResponseMap.entrySet()) {
      futures[cnt++] = CompletableFuture.runAsync(() -> {
        Iterator<Server.ServerResponse> streamingResponses = entry.getValue();
        try {
          while (streamingResponses.hasNext()) {
            Server.ServerResponse streamingResponse = streamingResponses.next();
            DataTable dataTable = DataTableFactory.getDataTable(streamingResponse.getPayload().asReadOnlyByteBuffer());
            // null dataSchema is a metadata-only block.
            if (dataTable.getDataSchema() != null) {
              reducer.reduce(entry.getKey(), dataTable);
            } else {
              aggregator.aggregate(entry.getKey(), dataTable);
            }
          }
        } catch (Exception e) {
          LOGGER.error("Unable to process streaming response. Failure occurred!", e);
          throw new RuntimeException("Unable to process streaming response. Failure occurred!", e);
        }
      }, executorService);
    }
    CompletableFuture<Void> syncWaitPoint = CompletableFuture.allOf(futures);
    try {
      syncWaitPoint.get(reduceTimeOutMs, TimeUnit.MILLISECONDS);
    } catch (Exception ex) {
      syncWaitPoint.cancel(true);
      throw ex;
    }
  }
@Test
  public void testThreadExceptionTransfer() {
    // simulate a thread exception in gRPC call and verify that the thread can transfer the exception
    Iterator<Server.ServerResponse> mockedResponse = (Iterator<Server.ServerResponse>) mock(Iterator.class);
    when(mockedResponse.hasNext()).thenReturn(true);
    String exceptionMessage = "Some exception";
    RuntimeException innerException = new RuntimeException(exceptionMessage);
    when(mockedResponse.next()).thenThrow(innerException);
    ExecutorService threadPoolService = Executors.newFixedThreadPool(1);
    ServerRoutingInstance routingInstance = new ServerRoutingInstance("localhost", 9527, TableType.OFFLINE);
    // supposedly we can use TestNG's annotation like @Test(expectedExceptions = { IOException.class }) to verify
    // here we hope to verify deeper to make sure the thrown exception is nested inside the exception
    assertTrue(verifyException(() -> {
          StreamingReduceService.processIterativeServerResponse(mock(StreamingReducer.class),
              threadPoolService,
              ImmutableMap.of(routingInstance, mockedResponse),
              1000,
              mock(ExecutionStatsAggregator.class));
          return null;
        }, cause -> cause.getMessage().contains(exceptionMessage))
    );
  }