docs: @Override
      public Boolean call() throws Exception {
          error.set(null);
          
          List<Callable<Boolean>> subtasks = Lists.newArrayList();
          
          // We are iterating the entire ring using an arbitrary number of threads
          if (this.concurrencyLevel != null || startToken != null|| endToken != null) {
              List<TokenRange> tokens = partitioner.splitTokenRange(
                      startToken == null ? partitioner.getMinToken() : startToken, 
                      endToken == null   ? partitioner.getMinToken() : endToken, 
                      this.concurrencyLevel == null ? 1 : this.concurrencyLevel);
              
              for (TokenRange range : tokens) {
                  subtasks.add(makeTokenRangeTask(range.getStartToken(), range.getEndToken()));
              }
          }
          // We are iterating through each token range
          else {
              List<TokenRange> ranges = keyspace.describeRing(dc, rack);
              for (TokenRange range : ranges) {
                  if (range.getStartToken().equals(range.getEndToken())) 
                      subtasks.add(makeTokenRangeTask(range.getStartToken(), range.getEndToken()));
                  else
                      subtasks.add(makeTokenRangeTask(partitioner.getTokenMinusOne(range.getStartToken()), range.getEndToken()));
              }
          }
          
          try {
              // Use a local executor
              if (executor == null) {
                  ExecutorService localExecutor = Executors
                          .newFixedThreadPool(subtasks.size(),
                              new ThreadFactoryBuilder().setDaemon(true)
                                  .setNameFormat("AstyanaxAllRowsReader-%d")
                                  .build());
                  
                  try {
                      futures.addAll(startTasks(localExecutor, subtasks));
                      return waitForTasksToFinish();
                  }
                  finally {
                      localExecutor.shutdownNow();
                  }
              }
              // Use an externally provided executor
              else {
                  futures.addAll(startTasks(executor, subtasks));
                  return waitForTasksToFinish();
              }
          }
          catch (Exception e) {
              error.compareAndSet(null, e);
              LOG.warn("AllRowsReader terminated. " + e.getMessage(), e);
              cancel();
              
              throw error.get();
          }
      }