HDFS Centrailized Cache会放到哪个DataNode上

  /**
   * Chooses datanode locations for caching from a list of valid possibilities.
   * Non-stale nodes are chosen before stale nodes.
   *
   * @param possibilities List of candidate datanodes
   * @param neededCached Number of replicas needed
   * @param staleInterval Age of a stale datanode
   * @return A list of chosen datanodes
   */
  private static List<DatanodeDescriptor> chooseDatanodesForCaching(
      final List<DatanodeDescriptor> possibilities, final int neededCached,
      final long staleInterval) {
    // Make a copy that we can modify
    List<DatanodeDescriptor> targets =
        new ArrayList<DatanodeDescriptor>(possibilities);
    // Selected targets
    List<DatanodeDescriptor> chosen = new LinkedList<DatanodeDescriptor>();

    // Filter out stale datanodes
    List<DatanodeDescriptor> stale = new LinkedList<DatanodeDescriptor>();
    Iterator<DatanodeDescriptor> it = targets.iterator();
    while (it.hasNext()) {
      DatanodeDescriptor d = it.next();
      if (d.isStale(staleInterval)) {
        it.remove();
        stale.add(d);
      }
    }
    // Select targets
    while (chosen.size() < neededCached) {
      // Try to use stale nodes if we're out of non-stale nodes, else we're done
      if (targets.isEmpty()) {
        if (!stale.isEmpty()) {
          targets = stale;
        } else {
          break;
        }
      }
      // Select a random target
      DatanodeDescriptor target =
          chooseRandomDatanodeByRemainingCapacity(targets);
      chosen.add(target);
      targets.remove(target);
    }
    return chosen;
  }

 /**
  * Add new entries to the PendingCached list.
  *
  * @param neededCached     The number of replicas that need to be cached.
  * @param cachedBlock      The block which needs to be cached.
  * @param cached           A list of DataNodes currently caching the block.
  * @param pendingCached    A list of DataNodes that will soon cache the
  *                         block.
  */
 private void addNewPendingCached(final int neededCached,
     CachedBlock cachedBlock, List<DatanodeDescriptor> cached,
     List<DatanodeDescriptor> pendingCached) {
   // To figure out which replicas can be cached, we consult the
   // blocksMap.  We don't want to try to cache a corrupt replica, though.
   BlockInfoContiguous blockInfo = blockManager.
         getStoredBlock(new Block(cachedBlock.getBlockId()));
   if (blockInfo == null) {
     LOG.debug("Block {}: can't add new cached replicas," +
         " because there is no record of this block " +
         "on the NameNode.", cachedBlock.getBlockId());
     return;
   }
   if (!blockInfo.isComplete()) {
     LOG.debug("Block {}: can't cache this block, because it is not yet"
         + " complete.", cachedBlock.getBlockId());
     return;
   }
   // Filter the list of replicas to only the valid targets
   List<DatanodeDescriptor> possibilities =
       new LinkedList<DatanodeDescriptor>();
   int numReplicas = blockInfo.getCapacity();
   Collection<DatanodeDescriptor> corrupt =
       blockManager.getCorruptReplicas(blockInfo);
   int outOfCapacity = 0;
   for (int i = 0; i < numReplicas; i++) {
     DatanodeDescriptor datanode = blockInfo.getDatanode(i);
     if (datanode == null) {
       continue;
     }
     if (datanode.isDecommissioned() || datanode.isDecommissionInProgress()) {
       continue;
     }
     if (corrupt != null && corrupt.contains(datanode)) {
       continue;
     }
     if (pendingCached.contains(datanode) || cached.contains(datanode)) {
       continue;
     }
     long pendingBytes = 0;
     // Subtract pending cached blocks from effective capacity
     Iterator<CachedBlock> it = datanode.getPendingCached().iterator();
     while (it.hasNext()) {
       CachedBlock cBlock = it.next();
       BlockInfoContiguous info =
           blockManager.getStoredBlock(new Block(cBlock.getBlockId()));
       if (info != null) {
         pendingBytes -= info.getNumBytes();
       }
     }
     it = datanode.getPendingUncached().iterator();
     // Add pending uncached blocks from effective capacity
     while (it.hasNext()) {
       CachedBlock cBlock = it.next();
       BlockInfoContiguous info =
           blockManager.getStoredBlock(new Block(cBlock.getBlockId()));
       if (info != null) {
         pendingBytes += info.getNumBytes();
       }
     }
     long pendingCapacity = pendingBytes + datanode.getCacheRemaining();
     if (pendingCapacity < blockInfo.getNumBytes()) {
       LOG.trace("Block {}: DataNode {} is not a valid possibility " +
           "because the block has size {}, but the DataNode only has {}" +
           "bytes of cache remaining ({} pending bytes, {} already cached.",
           blockInfo.getBlockId(), datanode.getDatanodeUuid(),
           blockInfo.getNumBytes(), pendingCapacity, pendingBytes,
           datanode.getCacheRemaining());
       outOfCapacity++;
       continue;
     }
     possibilities.add(datanode);
   }
   List<DatanodeDescriptor> chosen = chooseDatanodesForCaching(possibilities,
       neededCached, blockManager.getDatanodeManager().getStaleInterval());
   for (DatanodeDescriptor datanode : chosen) {
     LOG.trace("Block {}: added to PENDING_CACHED on DataNode {}",
         blockInfo.getBlockId(), datanode.getDatanodeUuid());
     pendingCached.add(datanode);
     boolean added = datanode.getPendingCached().add(cachedBlock);
     assert added;
   }
   // We were unable to satisfy the requested replication factor
   if (neededCached > chosen.size()) {
     LOG.debug("Block {}: we only have {} of {} cached replicas."
             + " {} DataNodes have insufficient cache capacity.",
         blockInfo.getBlockId(),
         (cachedBlock.getReplication() - neededCached + chosen.size()),
         cachedBlock.getReplication(), outOfCapacity
     );
   }
 }