docs: public Reporter execute() throws IOException {
          Map<ClusterNode, Iterator<Pair<ByteArray, Versioned<byte[]>>>> nodeFetchIteratorMap;
          nodeFetchIteratorMap = new HashMap<ClusterNode, Iterator<Pair<ByteArray, Versioned<byte[]>>>>();
          /* start fetch from each node */
          for(ClusterNode clusterNode: clusterNodeList) {
              AdminClient adminClient = adminClients.get(clusterNode.getPrefixId());
              List<Integer> singlePartition = new ArrayList<Integer>();
              singlePartition.add(partitionId);
              if(logger.isDebugEnabled()) {
                  logger.debug("Start fetch request to Node[" + clusterNode.toString()
                          + "] for partition[" + partitionId + "] of store[" + storeName + "]");
              }
  
              Iterator<Pair<ByteArray, Versioned<byte[]>>> fetchIterator;
              fetchIterator = adminClient.bulkFetchOps.fetchEntries(clusterNode.getNode().getId(),
                      storeName,
                      singlePartition,
                      null,
                      false);
              nodeFetchIteratorMap.put(clusterNode, fetchIterator);
          }
          keyFetchTracker = new KeyFetchTracker(clusterNodeList.size());
  
          /* start to fetch */
          boolean fetchFinished;
          do {
              fetchFinished = true;
              for(Map.Entry<ClusterNode, Iterator<Pair<ByteArray, Versioned<byte[]>>>> nodeFetchIteratorMapEntry: nodeFetchIteratorMap.entrySet()) {
                  ClusterNode clusterNode = nodeFetchIteratorMapEntry.getKey();
                  Iterator<Pair<ByteArray, Versioned<byte[]>>> fetchIterator = nodeFetchIteratorMapEntry.getValue();
                  if(fetchIterator.hasNext()) {
                      fetchFinished = false;
                      reporter.recordScans(1);
  
                      Pair<ByteArray, Versioned<byte[]>> fetchedEntry = fetchIterator.next();
                      ByteArray key = fetchedEntry.getFirst();
                      Versioned<byte[]> versioned = fetchedEntry.getSecond();
  
                      // record fetch
                      recordFetch(clusterNode, key, versioned);
  
                      // try sweep last key fetched by this iterator
                      keyFetchTracker.recordFetch(clusterNode, key);
                      if(logger.isTraceEnabled()) {
                          logger.trace("fetched " + new String(key.get()));
                          logger.trace("map has keys: " + keyValueNodeSetMap.size());
                      }
                      trySweepAll();
                      if(logger.isTraceEnabled()) {
                          logger.trace("sweeped; keys left: " + keyValueNodeSetMap.size());
                      }
                  }
              }
  
              // stats reporting
              if(logger.isInfoEnabled()) {
                  String report = reporter.tryProgressReport();
                  if(report != null) {
                      for(String line: report.split("\n")) {
                          logger.info(line);
                      }
                  }
              }
          } while(!fetchFinished);
  
          /* adminClient shutdown */
          for(AdminClient adminClient: adminClients) {
              if(adminClient != null) {
                  adminClient.close();
              }
          }
  
          // clean keys not sufficient for write
          cleanIneligibleKeys(keyValueNodeSetMap, requiredWrites);
  
          keyFetchTracker.finishAll();
          trySweepAll();
  
          reporter.processInconsistentKeys(storeName, partitionId, keyValueNodeSetMap);
  
          return reporter;
      }