class MultipleClientReportDownloader {

  public Collection<File> downloadReports(
      AdWordsSession.Builder sessionBuilder,
      ReportDefinition reportDefinition,
      Set<Long> clientCustomerIds)
      throws ReportProcessingException {
    ExecutorService executorService = Executors.newFixedThreadPool(numThreads);
    Stopwatch stopwatch = Stopwatch.createStarted();
    logger.info("Downloading {" + reportDefinition.getReportType() + "} reports...");

    List<Long> taskIds = Lists.newArrayList();
    List<CallableReportDownloader> taskJobs = Lists.newArrayList();
    Map<Long, String> failures = Maps.newHashMap();
    ImmutableAdWordsSession session = null;
    for (Long clientCustomerId : clientCustomerIds) {
      try {
        session = AdWordsSessionUtil.buildImmutableSessionForCid(sessionBuilder, clientCustomerId);
      } catch (ValidationException e) {
        logger.error("Failed to create valid adwords session for CID " + clientCustomerId
            + ", skipping it.");
        failures.put(clientCustomerId, e.getMessage());
        continue;
      }

      taskIds.add(clientCustomerId);
      taskJobs.add(genCallableReportDownloader(session, reportDefinition));
    }

    List<Future<File>> taskResults;
    try {
      //Note that invokeAll() returns results in the same sequence as input tasks.
      taskResults = executorService.invokeAll(taskJobs);
    } catch (InterruptedException e) {
      throw new ReportProcessingException(
        "MultipleClientReportDownloader encounters InterruptedException.", e);
    }

    List<File> results = Lists.newArrayList();
    for (int i = 0; i < taskResults.size(); i++) {
      try {
        results.add(taskResults.get(i).get());
      } catch (ExecutionException e) {
        failures.put(taskIds.get(i), e.getMessage());
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new ReportProcessingException(
            "MultiClientReportDownloader encounters InterruptedException.", e);
      }
    }

    executorService.shutdown();
    stopwatch.stop();

    logger.info("Downloaded reports for " + clientCustomerIds.size() + " accounts in "
        + stopwatch.elapsed(TimeUnit.SECONDS) + " seconds.");
    logger.info("Result: " +  results.size() + " successes, " + failures.size() + " failures.");
    if (!failures.isEmpty()) {
      StringBuilder sb = new StringBuilder("*** Account IDs of download failures ***");
      sb.append(SEPARATOR);
      sb.append(Joiner.on(SEPARATOR).withKeyValueSeparator(": ").join(failures));
      logger.error(sb.toString());
    }

    return results;
  }

  public  MultipleClientReportDownloader(int numThreads);

}