code: private void processRaidEncodingTaskAsync() {
      // Should not schedule raiding during safe mode
      if (isInSafeMode()) {
        return;
      }
      
      final int nodes = heartbeats.size();
      List<RaidBlockInfo> tasksToProcess = new ArrayList<RaidBlockInfo>(Math.min(
          raidEncodingTasks.size(),
          ReplicationConfigKeys.raidEncodingTaskMultiplier * nodes));
      
      for (int i = 0; i < ReplicationConfigKeys.raidEncodingTaskMultiplier; i++) {
        writeLock();
        try {
          NameNode.getNameNodeMetrics().numRaidEncodingTasks.set(
              raidEncodingTasks.size());
          raidEncodingTasks.pollNToList(nodes, tasksToProcess);
          if (raidEncodingTasks.isEmpty()) {
            break;
          }
        } finally {
          writeUnlock();
        }
      }
  
      ArrayList<RaidBlockInfo> tasksToAdd = new ArrayList<RaidBlockInfo>();
      for (RaidBlockInfo rbi : tasksToProcess) {
        if (NameNode.stateChangeLog.isDebugEnabled()) {
          NameNode.stateChangeLog
              .debug("BLOCK* NameSystem.processRaidEncodingTaskAsync: " + rbi);
        }
        if (processRaidEncodingTask(rbi)) {
          tasksToAdd.add(rbi);
        }
      }
      if (!tasksToAdd.isEmpty()) {
        writeLock();
        try {
          for (RaidBlockInfo rbi: tasksToAdd) {
            raidEncodingTasks.add(rbi);
          }
        } finally {
          writeUnlock();
        }
      }
    }
docs: This is called from the RaidEncodingTaskEncodingMonitor to process
  encoding tasks
  Each call will poll (raidEncodingTaskMultiplier*nodes) stripes from
  raidEncodingTasks.
  Then for each stripe, it will compute necessary information to encode
  the stripe and add encoding tasks to datanodes' queue.
  If it fails to compute, it will add back to the raidEncodingTasks
  for the next round