Hadoop3.2.1 【 HDFS 】源码分析 : 文件系统数据集 [三]

谁借莪１个温暖的怀抱￠ 2022-12-20 08:47 140阅读 0赞

### \----------- ###

*  1.前言
 *  2.ReplicaMap
 *  3.ReplicaInfo
 *  4.FsDatasetSpi
 *  5.FsDatasetImpl
 *   *  5.1.FsDatasetImpl字段
     *  5.2.FsDatasetImpl方法
     *   *  5.2.1. 操作底层FsVolumeList以及DataStorage相关方法。
         *  5.2.2. 数据块副本信息相关方法
         *  5.2.3. 操作缓存相关方法。
         *  5.2.4. 操作块池方法。
         *  5.2.5 数据块汇报相关方法。

# 1.前言 #

BlockPoolSlice负责管理单个存储目录下单个块池的所有数据块；  
FsVolumeImpl则负责管理一个完整的存储目录下所有的数据块， 也就包括了这个存储目录下多个BlockPoolSlice对象的引用。  
Datanode可以定义多个存储目录， 也就是定义多个FsVolumeImpl对象， 在HDFS中使用FsVolumeList对象统一管理Datanode上定义的多个FsVolumeImpl对象。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center]

FsDatasetSpi是FsDatasetImpl的根接口， 定义了Datanode管理数据块的接口方法。 FsDatasetImpl会通过持有一个FsVolumeList对象对Datanode上定义的所有存储目录下的数据块进行管理与操作， FsDatasetImpl还会持有一个DataStorage对象对Datanode的存储空间进行操作。 同时FsDatasetImpl会通过持有一个ReplicaMap对象维护Datanode上所有副本的信息与状态

# 2.ReplicaMap #

ReplicaMap保存了Datanode上所有数据块副本的信息。 ReplicaMap维护一个Map， 保存Datanode上所有块池（blockPoolId标识） 与这个块池保存的数据块副本信息的映射关系。

// Map of block pool Id to a set of ReplicaInfo.
      private final Map<String, FoldedTreeSet<ReplicaInfo>> map = new HashMap<>();

# 3.ReplicaInfo #

ReplicaInfo是一个抽象类， 用来描述Datanode上保存的数据块副本信息。 ReplicaInfo类对外提供了getBlockFile()、 getMetaFile()、 getVolume()、 getStorageUuid()等方法， 分别用于获取副本的数据块文件、 校验文件、 副本所在存储目录的FsVolumeImpl对象以及副  
本所在存储目录的storageUuid信息。

ReplicaInfo有如下几个子类分别用来描述处于不同状态的数据块副本。

■ ReplicaBeingWritten——对应RBW状态的副本， 副本正通过数据流管道写入Datanode。  
■ ReplicaUnderRecovery——对应RUR状态的副本， 副本正在进行恢复操作。  
■ ReplicaWaitingToBeRecovered——对应RWR状态的副本， 副本等待恢复操作。  
■ FinalizedReplica——对应FINALIZED状态的副本， 副本已经完成写操作， 并且已经提交。

# 4.FsDatasetSpi #

FsDatasetSpi是FsDatasetImpl的根接口， 继承自FSDatasetMBean接口， 定义了Datanode上管理与操作数据块副本的接口。

FsDatasetSpi定义的方法较多， 将方法分为如下几类:

■ 操作底层FsVolumeList以及DataStorage相关方法：getVolumes()、 getVolume()、getStorage()、 getVolumeInfoMap()等方法。  
■ 操作数据块相关方法：getFinalizedBlocks()、getStoredBlock()、getBlockInputStream()、 append()、 finalizeBlock()等方法。  
■ 操作缓存相关方法： cache()、 uncache()、 isCached()等方法。  
■ 操作块池方法： addBlockPool()、 shutdownBlockPool()、 deleteBlockPool()等方法。  
■ 数据块汇报相关方法： getStorageReports()、 getBlockReports()、getCacheReport()。  
■ 回收站相关方法： enableTrash()、 restoreTrash()、 trashEnabled()。

# 5.FsDatasetImpl #

FsDatasetImpl实现了FsDatasetSpi接口

## 5.1.FsDatasetImpl字段 ##

■ datanode： 当前数据节点的Datanode对象的引用。  
■ dataStorage： DataStorage对象的引用， 在FsDatasetImpl中调用这个对象开启和关闭存储空间回收站功能。  
■ volumes： FsVolumeList对象的引用， FsDatasetImpl使用FsVolumeList对象提供的功能管理Datanode上的所有数据块， 并且通过对volumes字段的修改添加和删除新的存储目录。  
■ storageMap： 在FsDatasetImpl中， 对于每一个存储目录都会构造一个DataStorage对象， storageMap字段就是用来保存这些存储的， 是storageUuid->DataStorage的映射。 FsDatasetImpl可以通过storageUuid获取对应的DataStorage对象。  
■ volumeMap： ReplicaMap类型， 用于记录Datanode上所有数据块副本的信息。  
■ cacheManager： FsDatasetCache类型， 用于将数据块缓存到内存中的工具类。  
■ asyncDiskService： FsDatasetAsyncDiskService类型， 用于在Datanode磁盘存储上进行一些异步的IO操作。 例如在进行删除数据块副本操作时， 会调用FsDatasetAsyncDiskService.deleteAsync()异步地将数据块文件从磁盘上删除。

// 当前数据节点的Datanode对象的引用。
      final DataNode datanode;
    
      // DataStorage对象的引用， 在FsDatasetImpl中调用这个对象开启和关闭存储空间回收站功能。
      final DataStorage dataStorage;
    
      // FsVolumeList对象的引用， FsDatasetImpl使用FsVolumeList对象提供的功能管理Datanode上的所有数据块，
      // 并且通过对volumes字段的修改添加和删除新的存储目录。
      private final FsVolumeList volumes;
    
      // 在FsDatasetImpl中， 对于每一个存储目录都会构造一个DataStorage对象， storageMap字段就是用来保存这些存储的， 是storageUuid->DataStorage的映射。
      // FsDatasetImpl可以通过storageUuid获取对应的DataStorage对象。
      final Map<String, DatanodeStorage> storageMap;
      final FsDatasetAsyncDiskService asyncDiskService;
      final Daemon lazyWriter;
    
      // FsDatasetCache类型， 用于将数据块缓存到内存中的工具类。
      final FsDatasetCache cacheManager;
      private final Configuration conf;
      private final int volFailuresTolerated;
      private final int volsConfigured;
      private volatile boolean fsRunning;
    
      // ReplicaMap类型， 用于记录Datanode上所有数据块副本的信息。
      final ReplicaMap volumeMap;
      final Map<String, Set<Long>> deletingBlock;
      final RamDiskReplicaTracker ramDiskReplicaTracker;
    
      // FsDatasetAsyncDiskService类型， 用于在Datanode磁盘存储上进行一些异步的IO操作。
      // 例如在进行删除数据块副本操作时， 会调用FsDatasetAsyncDiskService.deleteAsync()异步地将数据块文件从磁盘上删除。
      final RamDiskAsyncLazyPersistService asyncLazyPersistService;

## 5.2.FsDatasetImpl方法 ##

### 5.2.1. 操作底层FsVolumeList以及DataStorage相关方法。 ###

包括getVolumes()、 getVolume()、 getStorage()和getVolumeInfoMap()方法， 分别用于获取FsVolumeImpl对象的列表、 获取指定存储目录对应的FsVolumeImpl对象、 获取指定存储目录对应的DataStorage对象以及获取存储目录的卷信息。 这类方法的实现都比较简单， 基本都是直接返回对应的FsVolumeList与DataStorage对象。

@Override
      public Map<String, Object> getVolumeInfoMap() {
        final Map<String, Object> info = new HashMap<String, Object>();
        Collection<VolumeInfo> volumes = getVolumeInfo();
        for (VolumeInfo v : volumes) {
    
          final Map<String, Object> innerInfo = new HashMap<String, Object>();
    
          //存储目录中已经使用的空间
          innerInfo.put("usedSpace", v.usedSpace);
    
          //存储目录中可用的空间
          innerInfo.put("freeSpace", v.freeSpace);
    
          //存储目录中预留的空间
          innerInfo.put("reservedSpace", v.reservedSpace);
          //存储目录中副本预留的空间
          innerInfo.put("reservedSpaceForReplicas", v.reservedSpaceForReplicas);
          //存储 block 数量
          innerInfo.put("numBlocks", v.numBlocks);
          //存储目录类型
          innerInfo.put("storageType", v.storageType);
    
          //存储目录的路径
          info.put(v.directory, innerInfo);
          
        }
        return info;
      }

### 5.2.2. 数据块副本信息相关方法 ###

我们知道FsDatasetImpl使用ReplicaMap记录Datanode存储的所有副本信息，FsDatasetImpl定义了若干获取数据块副本信息的方法， 我们将这部分方法分为三类

*  获取数据块副本信息相关方法  
    这类方法提供了获取数据块副本相关信息的功能， 包括getStoredBlock()、getFinalizedBlocks()、 getReplicaString()、 getLength()等方法。 这些方法都是从FsDatasetImpl.volumeMap（ReplicaMap类型） 字段中提取出对应的副本信息（ReplicaInfo对象保存） ， 然后构造Block或者Replica对象返回即可。这里我们以getStoredBlock()方法为例， getStoredBlock()方法用于获取指定数据块的Block对象。 Block类封装了一个数据块副本的id、 数据块副本文件的长度以及时间戳。 这个方法首先调用getFile()方法从FsDatasetImpl.volumeMap字段中取出ReplicaInfo对象， 然后通过ReplicaInfo对象获取副本的数据块文件引用， 最后构造一个Block对象并返回。

@Override // FsDatasetSpi
      public Block getStoredBlock(String bpid, long blkid)
          throws IOException {
        try (AutoCloseableLock lock = datasetLock.acquire()) {
          //调用getFile()获取副本的数据块文件引用
          ReplicaInfo r = volumeMap.get(bpid, blkid);
          if (r == null) {
            return null;
          }
          //构造Block对象返回
          return new Block(blkid, r.getBytesOnDisk(), r.getGenerationStamp());
        }
      }

*  获取数据块副本IO流相关方法  
    包括getBlockInputStream()和getMetaDataInputStream()两个方法， 分别用于获取副本的数据块文件以及校验文件的IO流。 这两个方法的实现非常简单， 就是通过上面介绍的getFile()方法获得数据块文件引用， 然后打开文件构造IO流返回即可。

@Override // FsDatasetSpi
      public InputStream getBlockInputStream(ExtendedBlock b,
          long seekOffset) throws IOException {
    
        ReplicaInfo info;
        try (AutoCloseableLock lock = datasetLock.acquire()) {
          info = volumeMap.get(b.getBlockPoolId(), b.getLocalBlock());
        }
    
        if (info != null && info.getVolume().isTransientStorage()) {
          ramDiskReplicaTracker.touch(b.getBlockPoolId(), b.getBlockId());
          datanode.getMetrics().incrRamDiskBlocksReadHits();
        }
    
        if (info != null) {
          return info.getDataInputStream(seekOffset);
        } else {
          throw new IOException("No data exists for block " + b);
        }
      }

@Override // FsDatasetSpi
      public LengthInputStream getMetaDataInputStream(ExtendedBlock b)
          throws IOException {
        ReplicaInfo info = getBlockReplica(b);
        if (info == null || !info.metadataExists()) {
          return null;
        }
        return info.getMetadataInputStream(0);
      }

### 5.2.3. 操作缓存相关方法。 ###

FsDatasetImpl使用volumeMap字段记录所有副本的信息， volumeMap对象保存在内存中， 这就可能出现内存中volumeMap记录的副本状态与磁盘上存储的副本状态不一致的情况。

FsDatasetImpl提供了checkAndUpdate()方法同步内存中的副本状态和磁盘上存储的副本状态。  
这个方法是在DirectoryScanner.run()中调用的， DirectoryScanner首先调用scan()方法获取内存与磁盘不一致的数据块副本引用， 然后调用checkAndUpdate()方法同步两者的状态。

checkAndUpdate方法的逻辑如下：

1.  如果磁盘上没有这个数据块， 则从volumeMap中删除数据块
2.  如果磁盘上存在这个数据块， volumeMap中不存在， 则添加到volumeMap中。
3.  如果两者的时间戳不匹配， 则用正确的时间戳更新副本信息。
4.  如果内存中数据块的长度与磁盘上数据块的长度不匹配， 则将数据块标识为损坏（corrupt） ， 然后更新副本的长度。
5.  如果内存中记录的数据块文件与磁盘上的数据块文件不匹配， 则更新副本的数据块文件引用

**对于情况1**， 要特别注意从内存中删除数据块副本时， 不仅仅要从volumeMap中删除， 还需要调用blockScanner.deleteBlock()方法从BlockScanner对象中删除。  
**对于情况4**，当出现数据块长度不一致的情况时， 还需要调用datanode.reportBadBlocks()向Namenode汇报这个数据块副本已经损坏。  
**对于情况5**， 如果volumeMap中记录的数据块文件与磁盘扫描的数据块文件不一致， 则调用BlockPoolSlice.resolveDuplicateReplicas()方法判断保留哪一个数据块文件。

判断的逻辑是：
    	■ 选择时间戳大的文件。
    	■ 如果时间戳相同， 则选择文件长度大的文件。
    	■ 如果文件长度也相同， 则选择在持久化卷上的文件。
    	■ 如果上述三条全都一样， 则选择第一个文件。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 1]

/**
       *
       * 处理磁盘上的块和卷映射中的块之间的差异
       * 检查给定block的差异性.
       * 检查内存和磁盘中block的差异性:
       * 1. 如果block文件丢失, 删除内存中的信息.
       * 2. 如果block文件存在,内存中不存在,更新内存中的信息
       * 3. 如果时间戳不匹配,使用正确的时间错
       * 4. 如果如果内存中的block大小跟实际的blokck大小不一致,标记这个block为损坏,更新内存中的block
       * 5. 如果内存中的block和磁盘中的block不匹配,更新内存中的block文件
       *
       * Reconcile the difference between blocks on the disk and blocks in volumeMap
       *
       * Check the given block for inconsistencies.
       * Look at the current state of the block and reconcile the differences as follows:
       * <ul>
       * <li>If the block file is missing, delete the block from volumeMap</li>
       * <li>If the block file exists and the block is missing in volumeMap,
       * add the block to volumeMap <li>
       * <li>If generation stamp does not match, then update the block with right
       * generation stamp</li>
       * <li>If the block length in memory does not match the actual block file length
       * then mark the block as corrupt and update the block length in memory</li>
       * <li>If the file in {@link ReplicaInfo} does not match the file on
       * the disk, update {@link ReplicaInfo} with the correct file</li>
       * </ul>
       *
       * @param bpid block pool ID
       * @param scanInfo {@link ScanInfo} for a given block
       */
      @Override
      public void checkAndUpdate(String bpid, ScanInfo scanInfo)
          throws IOException {
    
        // 获取blockId
        long blockId = scanInfo.getBlockId();
    
        // 获取块文件
        File diskFile = scanInfo.getBlockFile();
    
        // 获取元数据文件
        File diskMetaFile = scanInfo.getMetaFile();
    
        // 获取volume信息
        FsVolumeSpi vol = scanInfo.getVolume();
    
        Block corruptBlock = null;
        ReplicaInfo memBlockInfo;
        // 获取lock
        try (AutoCloseableLock lock = datasetLock.acquire()) {
          //获取内存block信息
          memBlockInfo = volumeMap.get(bpid, blockId);
    
          // block正在持久化中, 忽略不同信息
          if (memBlockInfo != null &&
              memBlockInfo.getState() != ReplicaState.FINALIZED) {
            // Block is not finalized - ignore the difference
            return;
          }
    
          // 获取fileIoProvider
          final FileIoProvider fileIoProvider = datanode.getFileIoProvider();
    
          // 验证磁盘元数据是否存在
          final boolean diskMetaFileExists = diskMetaFile != null &&
              fileIoProvider.exists(vol, diskMetaFile);
    
          // 验证block文件是否存在
          final boolean diskFileExists = diskFile != null &&
              fileIoProvider.exists(vol, diskFile);
    
          // 生成戳标记 ?????
          final long diskGS = diskMetaFileExists ?
              Block.getGenerationStamp(diskMetaFile.getName()) :
              HdfsConstants.GRANDFATHER_GENERATION_STAMP;
    
    
          //外部存储?? 略过...
          if (vol.getStorageType() == StorageType.PROVIDED) {
            if (memBlockInfo == null) {
              // replica exists on provided store but not in memory
              ReplicaInfo diskBlockInfo =
                  new ReplicaBuilder(ReplicaState.FINALIZED)
                  .setFileRegion(scanInfo.getFileRegion())
                  .setFsVolume(vol)
                  .setConf(conf)
                  .build();
    
              volumeMap.add(bpid, diskBlockInfo);
              LOG.warn("Added missing block to memory " + diskBlockInfo);
            } else {
              // replica exists in memory but not in the provided store
              volumeMap.remove(bpid, blockId);
              LOG.warn("Deleting missing provided block " + memBlockInfo);
            }
            return;
          }
    
          // 如果磁盘文件不存在
          if (!diskFileExists) {
            //如果元数据为空
            if (memBlockInfo == null) {
              // Block file does not exist and block does not exist in memory
              // If metadata file exists then delete it
              // block文件不存在, 内存中不存在, 但是有原信息文件存在,则删除.
              if (diskMetaFileExists && fileIoProvider.delete(vol, diskMetaFile)) {
                LOG.warn("Deleted a metadata file without a block "
                    + diskMetaFile.getAbsolutePath());
              }
              return;
            }
            //如果元信息文件不存在
            if (!memBlockInfo.blockDataExists()) {
              // Block is in memory and not on the disk
              // Remove the block from volumeMap
              // block在内存中,但是不在磁盘中, 从内存中清理到元信息.
              volumeMap.remove(bpid, blockId);
    
              if (vol.isTransientStorage()) {
                ramDiskReplicaTracker.discardReplica(bpid, blockId, true);
              }
              LOG.warn("Removed block " + blockId
                  + " from memory with missing block file on the disk");
              // Finally remove the metadata file
              // 最后移除元信息文件
              if (diskMetaFileExists && fileIoProvider.delete(vol, diskMetaFile)) {
                LOG.warn("Deleted a metadata file for the deleted block "
                    + diskMetaFile.getAbsolutePath());
              }
            }
            return;
          }
          /* block文件存在于disk
           * Block file exists on the disk
           */
          if (memBlockInfo == null) {
            // Block is missing in memory - add the block to volumeMap
            // 内存中的block信息丢失, 向内存中添加block信息.
    
            //构造内存对象
            ReplicaInfo diskBlockInfo = new ReplicaBuilder(ReplicaState.FINALIZED)
                .setBlockId(blockId)
                .setLength(diskFile.length())
                .setGenerationStamp(diskGS)
                .setFsVolume(vol)
                .setDirectoryToUse(diskFile.getParentFile())
                .build();
    
            //更新内存信息
            volumeMap.add(bpid, diskBlockInfo);
            if (vol.isTransientStorage()) {
              long lockedBytesReserved =
                  cacheManager.reserve(diskBlockInfo.getNumBytes()) > 0 ?
                      diskBlockInfo.getNumBytes() : 0;
              ramDiskReplicaTracker.addReplica(
                  bpid, blockId, (FsVolumeImpl) vol, lockedBytesReserved);
            }
            LOG.warn("Added missing block to memory " + diskBlockInfo);
            return;
          }
          /*
           * Block exists in volumeMap and the block file exists on the disk
           * block 在内存和磁盘中都存在
           */
          // 对比block文件
          // Compare block files
          if (memBlockInfo.blockDataExists()) {
            if (memBlockInfo.getBlockURI().compareTo(diskFile.toURI()) != 0) {
              if (diskMetaFileExists) {
                if (memBlockInfo.metadataExists()) {
                  // We have two sets of block+meta files. Decide which one to
                  // keep.
                  ReplicaInfo diskBlockInfo =
                      new ReplicaBuilder(ReplicaState.FINALIZED)
                        .setBlockId(blockId)
                        .setLength(diskFile.length())
                        .setGenerationStamp(diskGS)
                        .setFsVolume(vol)
                        .setDirectoryToUse(diskFile.getParentFile())
                        .build();
                  ((FsVolumeImpl) vol).resolveDuplicateReplicas(bpid,
                      memBlockInfo, diskBlockInfo, volumeMap);
                }
              } else {
                if (!fileIoProvider.delete(vol, diskFile)) {
                  LOG.warn("Failed to delete " + diskFile);
                }
              }
            }
          } else {
            // Block refers to a block file that does not exist.
            // Update the block with the file found on the disk. Since the block
            // file and metadata file are found as a pair on the disk, update
            // the block based on the metadata file found on the disk
            LOG.warn("Block file in replica "
                + memBlockInfo.getBlockURI()
                + " does not exist. Updating it to the file found during scan "
                + diskFile.getAbsolutePath());
    
            // 更新内存信息
            memBlockInfo.updateWithReplica(
                StorageLocation.parse(diskFile.toString()));
    
            LOG.warn("Updating generation stamp for block " + blockId
                + " from " + memBlockInfo.getGenerationStamp() + " to " + diskGS);
            memBlockInfo.setGenerationStamp(diskGS);
          }
    
          // Compare generation stamp
          // 对比时间戳
          if (memBlockInfo.getGenerationStamp() != diskGS) {
            File memMetaFile = FsDatasetUtil.getMetaFile(diskFile, 
                memBlockInfo.getGenerationStamp());
            if (fileIoProvider.exists(vol, memMetaFile)) {
              String warningPrefix = "Metadata file in memory "
                  + memMetaFile.getAbsolutePath()
                  + " does not match file found by scan ";
              if (!diskMetaFileExists) {
                LOG.warn(warningPrefix + "null");
              } else if (memMetaFile.compareTo(diskMetaFile) != 0) {
                LOG.warn(warningPrefix + diskMetaFile.getAbsolutePath());
              }
            } else {
              // Metadata file corresponding to block in memory is missing
    
    
              // If metadata file found during the scan is on the same directory
              // as the block file, then use the generation stamp from it
              try {
                File memFile = new File(memBlockInfo.getBlockURI());
                long gs = diskMetaFileExists &&
                    diskMetaFile.getParent().equals(memFile.getParent()) ? diskGS
                    : HdfsConstants.GRANDFATHER_GENERATION_STAMP;
    
                LOG.warn("Updating generation stamp for block " + blockId
                    + " from " + memBlockInfo.getGenerationStamp() + " to " + gs);
    
                memBlockInfo.setGenerationStamp(gs);
              } catch (IllegalArgumentException e) {
                //exception arises because the URI cannot be converted to a file
                LOG.warn("Block URI could not be resolved to a file", e);
              }
            }
          }
    
          // Compare block size
          // 对比大小
          if (memBlockInfo.getNumBytes() != memBlockInfo.getBlockDataLength()) {
            // Update the length based on the block file
            corruptBlock = new Block(memBlockInfo);
            LOG.warn("Updating size of block " + blockId + " from "
                + memBlockInfo.getNumBytes() + " to "
                + memBlockInfo.getBlockDataLength());
            memBlockInfo.setNumBytes(memBlockInfo.getBlockDataLength());
          }
        }
    
        // Send corrupt block report outside the lock
        // 发送损坏block信息
        if (corruptBlock != null) {
          LOG.warn("Reporting the block " + corruptBlock
              + " as corrupt due to length mismatch");
          try {
            datanode.reportBadBlocks(new ExtendedBlock(bpid, corruptBlock),
                memBlockInfo.getVolume());
          } catch (IOException e) {
            LOG.warn("Failed to report bad block " + corruptBlock, e);
          }
        }
      }

### 5.2.4. 操作块池方法。 ###

在FsDatasetImpl实现中最重要的一部分就是操作数据块副本的方法.  
在FsDatasetImpl中操作数据块副本的方法会对副本的状态产生影响.

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 2]

\*\*createTemporary()\*\*方法的代码实现很简单，在做一系列验证之后. 首先调用FsVolumeList.getNextVolume()方法获取一个能够保存这个副本的FsVolumeImpl对象， 然后在这个FsVolumeImpl对象上调用createTmpFile()方法在该数据块副本所在块池的tmp目录中创建副本文件， 接下来构造ReplicaInPipeline对象放入FsDatasetImpl.volumeMap中保存， 最后返回这个ReplicaInPipeline对象。

@Override // FsDatasetSpi
      public ReplicaHandler createTemporary(StorageType storageType,
          String storageId, ExtendedBlock b, boolean isTransfer)
          throws IOException {
        long startTimeMs = Time.monotonicNow();
        long writerStopTimeoutMs = datanode.getDnConf().getXceiverStopTimeout();
        ReplicaInfo lastFoundReplicaInfo = null;
        boolean isInPipeline = false;
    
        // 这是一个do {} while(); 循环代码...
        do {
    
          try (AutoCloseableLock lock = datasetLock.acquire()) {
    
    
            ReplicaInfo currentReplicaInfo =  volumeMap.get(b.getBlockPoolId(), b.getBlockId());
            if (currentReplicaInfo == lastFoundReplicaInfo) {
              break;
            } else {
              //是否在构建中
              isInPipeline = currentReplicaInfo.getState() == ReplicaState.TEMPORARY
                  || currentReplicaInfo.getState() == ReplicaState.RBW;
              /* 如果当前的block不是外部存储并且是旧的block 直接拒绝
               * 否则，如果是请求转移，则接受它。
               * 否则，如果状态不是RBW/Temporary，则拒绝
               * 如果是外部存储，则忽略这个副本。
               *
               * If the current block is not PROVIDED and old, reject.
               * else If transfer request, then accept it.
               * else if state is not RBW/Temporary, then reject
               * If current block is PROVIDED, ignore the replica.
               */
              if (((currentReplicaInfo.getGenerationStamp() >= b
                  .getGenerationStamp()) || (!isTransfer && !isInPipeline))
                  && !isReplicaProvided(currentReplicaInfo)) {
                throw new ReplicaAlreadyExistsException("Block " + b
                    + " already exists in state " + currentReplicaInfo.getState()
                    + " and thus cannot be created.");
              }
              lastFoundReplicaInfo = currentReplicaInfo;
            }
    
          }
          if (!isInPipeline) {
            // 构建中, 忽略...
            continue;
          }
          // Hang too long, just bail out. This is not supposed to happen.
          // 如果挂起太久了, 抛出异常, 正常不应该发生这种情况
          long writerStopMs = Time.monotonicNow() - startTimeMs;
          if (writerStopMs > writerStopTimeoutMs) {
            LOG.warn("Unable to stop existing writer for block " + b + " after " 
                + writerStopMs + " miniseconds.");
            throw new IOException("Unable to stop existing writer for block " + b
                + " after " + writerStopMs + " miniseconds.");
          }
    
          // if lastFoundReplicaInfo is PROVIDED and FINALIZED,
          // stopWriter isn't required.
    
          // 如果lastFoundReplicaInfo 是外部存储 并且是FINALIZED . 不需要stopWriter
          if (isReplicaProvided(lastFoundReplicaInfo) &&
              lastFoundReplicaInfo.getState() == ReplicaState.FINALIZED) {
            continue;
          }
          // 停止外部写入
          // Stop the previous writer
          ((ReplicaInPipeline)lastFoundReplicaInfo).stopWriter(writerStopTimeoutMs);
        } while (true);
    
        //
        if (lastFoundReplicaInfo != null
            && !isReplicaProvided(lastFoundReplicaInfo)) {
          // 旧的块文件应该同步删除，因为如果在同一个卷中分配，它可能会和新块发生冲突。
          //
          // 在锁外作为其磁盘IO进行删除。
    
          // Old blockfile should be deleted synchronously as it might collide
          // with the new block if allocated in same volume.
          // Do the deletion outside of lock as its DISK IO.
          invalidate(b.getBlockPoolId(), new Block[] { lastFoundReplicaInfo },
              false);
        }
    
    
        try (AutoCloseableLock lock = datasetLock.acquire()) {
    
          // 获取FsVolumeReference
          FsVolumeReference ref = volumes.getNextVolume(storageType, storageId, b .getNumBytes());
    
          //获取一个FsVolumeImpl对象来存储该副本
          FsVolumeImpl v = (FsVolumeImpl) ref.getVolume();
          ReplicaInPipeline newReplicaInfo;
          try {
            // 构建Temporary 存储
            newReplicaInfo = v.createTemporary(b);
          } catch (IOException e) {
            IOUtils.cleanup(null, ref);
            throw e;
          }
          //将新构造的ReplicaInPipeline对象放入FsDatasetImpl.volumeMap中
          volumeMap.add(b.getBlockPoolId(), newReplicaInfo.getReplicaInfo());
          return new ReplicaHandler(newReplicaInfo, ref);
        }
      }
    
    
      public ReplicaInPipeline createTemporary(ExtendedBlock b) throws IOException {
        // create a temporary file to hold block in the designated volume
        //在该存储目录下的数据块所在块池的tmp子目录中创建数据块文件
        File f = createTmpFile(b.getBlockPoolId(), b.getLocalBlock());
        // 构建副本信息
        LocalReplicaInPipeline newReplicaInfo =
            new ReplicaBuilder(ReplicaState.TEMPORARY)
              .setBlockId(b.getBlockId())
              .setGenerationStamp(b.getGenerationStamp())
              .setDirectoryToUse(f.getParentFile())
              .setBytesToReserve(b.getLocalBlock().getNumBytes())
              .setFsVolume(this)
              .buildLocalReplicaInPipeline();
        return newReplicaInfo;
      }

\*\*createRbw()\*\*方法用于在rbw文件夹中创建一个副本（当HDFS客户端发起写请求而创建一个副本时， 这个副本将会被放到rbw文件夹中， 并设置为RBW状态， RBW状态的副本对于读线程是可见的） 。 这个方法是在当前Datanode接收来自客户端的写数据请求或者追加写数据请求时， 在BlockRecevier的构造方法中调用的。 createRbw()方法的实现与createTemporary()是大致相同的， 不同的是副本存放在块池目录的rbw文件夹下， 同时增加了lazyPersist的支持。

@Override // FsDatasetSpi
      public ReplicaHandler createRbw( StorageType storageType, String storageId, ExtendedBlock b, boolean allowLazyPersist) throws IOException {
    
        try (AutoCloseableLock lock = datasetLock.acquire()) {
    
          // 获取副本信息
          ReplicaInfo replicaInfo = volumeMap.get(b.getBlockPoolId(),  b.getBlockId());
    
          if (replicaInfo != null) {
            throw new ReplicaAlreadyExistsException("Block " + b +
                " already exists in state " + replicaInfo.getState() +
                " and thus cannot be created.");
          }
          // create a new block
          FsVolumeReference ref = null;
    
          // 只有块大小是操作系统 page 大小的倍数时才使用ramdisk。
          // 这大大简化了对部分使用的副本的保留。
          //
          //
          // Use ramdisk only if block size is a multiple of OS page size.
          // This simplifies reservation for partially used replicas
          // significantly.
          if (allowLazyPersist &&
              lazyWriter != null &&
              b.getNumBytes() % cacheManager.getOsPageSize() == 0 &&
              reserveLockedMemory(b.getNumBytes())) {
            try {
              // First try to place the block on a transient volume.
              ref = volumes.getNextTransientVolume(b.getNumBytes());
              datanode.getMetrics().incrRamDiskBlocksWrite();
            } catch (DiskOutOfSpaceException de) {
              // Ignore the exception since we just fall back to persistent storage.
              LOG.warn("Insufficient space for placing the block on a transient "
                  + "volume, fall back to persistent storage: "
                  + de.getMessage());
            } finally {
              if (ref == null) {
                cacheManager.release(b.getNumBytes());
              }
            }
          }
    
          if (ref == null) {
            ref = volumes.getNextVolume(storageType, storageId, b.getNumBytes());
          }
    
          FsVolumeImpl v = (FsVolumeImpl) ref.getVolume();
          // create an rbw file to hold block in the designated volume
          // 创建rbw文件以在指定卷中保存块
          if (allowLazyPersist && !v.isTransientStorage()) {
            datanode.getMetrics().incrRamDiskBlocksWriteFallback();
          }
    
          ReplicaInPipeline newReplicaInfo;
          try {
    
            // 构建newReplicaInfo
            newReplicaInfo = v.createRbw(b);
            if (newReplicaInfo.getReplicaInfo().getState() != ReplicaState.RBW) {
              throw new IOException("CreateRBW returned a replica of state "
                  + newReplicaInfo.getReplicaInfo().getState()
                  + " for block " + b.getBlockId());
            }
          } catch (IOException e) {
            IOUtils.cleanup(null, ref);
            throw e;
          }
          //持久化内存信息
          volumeMap.add(b.getBlockPoolId(), newReplicaInfo.getReplicaInfo());
          return new ReplicaHandler(newReplicaInfo, ref);
        }
      }

\*\*recoverRbw()\*\*方法用于恢复一个RBW状态的副本， 并返回这个副本的信息。 这个方法是在Datanode进行数据流管道恢复操作时， 在BlockReceiver的构造方法中调用的。 调用完recoverRbw()方法后， 数据流管道中的所有RBW状态副本的时间戳和文件长度也就同步了， 可以继续通过数据流管道写入新的数据了。  
recoverRbw()方法主要进行了三个操作：  
停止副本原先的写入线程， 并将当前线程更新为写入线程；  
对时间戳进行检查， 对rbw数据块的长度进行检查；  
检查完成之后， 更新rbw副本的时间戳， 然后将副本中没有ack的数据抛弃。

@Override // FsDatasetSpi
      public ReplicaHandler recoverRbw(
          ExtendedBlock b, long newGS, long minBytesRcvd, long maxBytesRcvd)
          throws IOException {
        LOG.info("Recover RBW replica " + b);
    
        while (true) {
          try {
            try (AutoCloseableLock lock = datasetLock.acquire()) {
    
              /// 获取副本信息
              ReplicaInfo replicaInfo = getReplicaInfo(b.getBlockPoolId(), b.getBlockId());
    
              // 检查副本的状态, 如果副本的状态不是 RBW , 抛出异常
              // check the replica's state
              if (replicaInfo.getState() != ReplicaState.RBW) {
                throw new ReplicaNotFoundException(
                    ReplicaNotFoundException.NON_RBW_REPLICA + replicaInfo);
              }
              ReplicaInPipeline rbw = (ReplicaInPipeline)replicaInfo;
              //停止副本原先的写入线程， 并将当前线程更新为写入线程
              if (!rbw.attemptToSetWriter(null, Thread.currentThread())) {
                throw new MustStopExistingWriter(rbw);
              }
              LOG.info("At " + datanode.getDisplayName() + ", Recovering " + rbw);
    
              // 回复RBW 实现
              return recoverRbwImpl(rbw, b, newGS, minBytesRcvd, maxBytesRcvd);
            }
          } catch (MustStopExistingWriter e) {
            e.getReplicaInPipeline().stopWriter(
                datanode.getDnConf().getXceiverStopTimeout());
          }
        }
      }
    
      private ReplicaHandler recoverRbwImpl(ReplicaInPipeline rbw,
          ExtendedBlock b, long newGS, long minBytesRcvd, long maxBytesRcvd)
          throws IOException {
        try (AutoCloseableLock lock = datasetLock.acquire()) {
    
          // 检查时间戳标记
          // check generation stamp
          long replicaGenerationStamp = rbw.getGenerationStamp();
          // 检查时间戳范围， 如果不在b.getGenerationStamp()至newGS之间则抛出异常
          // 副本时间戳不匹配,直接抛出异常
          if (replicaGenerationStamp < b.getGenerationStamp() ||
              replicaGenerationStamp > newGS) {
            throw new ReplicaNotFoundException(
                ReplicaNotFoundException.UNEXPECTED_GS_REPLICA + b +
                    ". Expected GS range is [" + b.getGenerationStamp() + ", " +
                    newGS + "].");
          }
    
          // 检查副本长度
          //检查副本的长度， 如果不在minBytesRcvd与maxBytesRcvd之间则抛出异常
          // check replica length
          long bytesAcked = rbw.getBytesAcked();
          long numBytes = rbw.getNumBytes();
          if (bytesAcked < minBytesRcvd || numBytes > maxBytesRcvd) {
            throw new ReplicaNotFoundException("Unmatched length replica " +
                rbw + ": BytesAcked = " + bytesAcked +
                " BytesRcvd = " + numBytes + " are not in the range of [" +
                minBytesRcvd + ", " + maxBytesRcvd + "].");
          }
    
          long bytesOnDisk = rbw.getBytesOnDisk();
          long blockDataLength = rbw.getReplicaInfo().getBlockDataLength();
          if (bytesOnDisk != blockDataLength) {
            LOG.info("Resetting bytesOnDisk to match blockDataLength (={}) for " +
                "replica {}", blockDataLength, rbw);
            bytesOnDisk = blockDataLength;
            rbw.setLastChecksumAndDataLen(bytesOnDisk, null);
          }
    
    
          if (bytesOnDisk < bytesAcked) {
            throw new ReplicaNotFoundException("Found fewer bytesOnDisk than " +
                "bytesAcked for replica " + rbw);
          }
    
          FsVolumeReference ref = rbw.getReplicaInfo()
              .getVolume().obtainReference();
          try {
            // Truncate the potentially corrupt portion.
            // If the source was client and the last node in the pipeline was lost,
            // any corrupt data written after the acked length can go unnoticed.
    
            // 清空可能损坏的部分。
            //如果source 是client，并且管道中的最后一个节点丢失，
            // 则在确认长度之后写入的任何损坏数据都会忽略。
    
            //将副本中所有没有确认的数据包删除
            if (bytesOnDisk > bytesAcked) {
              rbw.getReplicaInfo().truncateBlock(bytesAcked);
              rbw.setNumBytes(bytesAcked);
              rbw.setLastChecksumAndDataLen(bytesAcked, null);
            }
            // 设置新的时间戳
            // bump the replica's generation stamp to newGS
            rbw.getReplicaInfo().bumpReplicaGS(newGS);
          } catch (IOException e) {
            IOUtils.cleanup(null, ref);
            throw e;
          }
          return new ReplicaHandler(rbw, ref);
        }
      }

\*\*convertTemporaryToRbw()\*\*方法用于将TEMPORARY状态的副本转换为RBW状态。  
这个方法是在当前Datanode从其他Datanode成功地接收了复制（transfer） 的数据块时， 在BlockReceiver.receiveBlock()方法中调用的。 因为在当前Datanode从其他Datanode接收数据时， 处于TEMPORARY状态的副本是不完整的， 并且对于读线程是不可见的。 当数据块传输完成后， Datanode上的数据块完整了， 也就是可以进行读取操作了， 这时就需要将副本状态从TEMPORARY转换为RBW状态（只有RBW状态的副本对于客户端才是可见的） 。  
convertTemporaryToRbw()方法首先对副本状态、 时间戳以及数据块文件长度进行了检查。 如果出现时间戳不等、 数据块文件长度不匹配或者副本状态异常等情况， 则直接抛出异常。 如果这些判断都正常， 则将tmp文件夹中的副本文件移动到rbw文件夹下， 然后构造新的ReplicaBeingWritten对象并更新volumeMap.

@Override // FsDatasetSpi
      public ReplicaInPipeline convertTemporaryToRbw(
          final ExtendedBlock b) throws IOException {
    
        try (AutoCloseableLock lock = datasetLock.acquire()) {
          final long blockId = b.getBlockId();
          final long expectedGs = b.getGenerationStamp();
          final long visible = b.getNumBytes();
          LOG.info("Convert " + b + " from Temporary to RBW, visible length="
              + visible);
    
          final ReplicaInfo temp;
          {
            // get replica
            final ReplicaInfo r = volumeMap.get(b.getBlockPoolId(), blockId);
            //检查副本信息是否存在
            if (r == null) {
              throw new ReplicaNotFoundException(
                  ReplicaNotFoundException.NON_EXISTENT_REPLICA + b);
            }
            // 检查副本状态是否为TEMPORARY,否则拋出异常
            // check the replica's state
            if (r.getState() != ReplicaState.TEMPORARY) {
              throw new ReplicaAlreadyExistsException(
                  "r.getState() != ReplicaState.TEMPORARY, r=" + r);
            }
            temp = r;
          }
          //检查时间戳是否正常
          // check generation stamp
          if (temp.getGenerationStamp() != expectedGs) {
            throw new ReplicaAlreadyExistsException(
                "temp.getGenerationStamp() != expectedGs = " + expectedGs
                    + ", temp=" + temp);
          }
    
          // TODO: check writer?
          // set writer to the current thread
          // temp.setWriter(Thread.currentThread());
    
          // check length
          // 检查副本长度， 如果副本长度小于传入的长度， 则抛出异常
          final long numBytes = temp.getNumBytes();
          if (numBytes < visible) {
            throw new IOException(numBytes + " = numBytes < visible = "
                + visible + ", temp=" + temp);
          }
          // check volume
          // 检查副本所在存储目录的FsVolumeImpl是否存在， 不存在则抛出异常
          final FsVolumeImpl v = (FsVolumeImpl) temp.getVolume();
          if (v == null) {
            throw new IOException("r.getVolume() = null, temp=" + temp);
          }
    
          //将副本从tmp目录移动到rbw目录
          final ReplicaInPipeline rbw = v.convertTemporaryToRbw(b, temp);
    
          if(rbw.getState() != ReplicaState.RBW) {
            throw new IOException("Expected replica state: " + ReplicaState.RBW
                + " obtained " + rbw.getState() + " for converting block "
                + b.getBlockId());
          }
          //构造新的ReplicaInfo对象， 并在volumeMap中更新
          // overwrite the RBW in the volume map
          volumeMap.add(b.getBlockPoolId(), rbw.getReplicaInfo());
          return rbw;
        }
      }

\*\*append()\*\*方法用于对FINALIZED状态的副本执行追加写操作。 这个方法首先进行副本状态检查、 时间戳检查以及文件长度检查， 然后调用私有的append()方法进行真正的追加写操作.

私有的append()方法首先调用uncacheBlock()方法， 将当前副本从缓存中移出。 移出的原因是缓存中保存的数据块是需要确保数据块文件的校验和完全匹配的， 如果进行了append()操作， 那么数据块文件的校验和需要重新确认。

调用完uncacheBlock()方法之后， append()方法会调用ReplicaInfo.unlinkBlock()方法去除该数据块文件的硬链接。 调用这个方法的原因是在Datanode升级过程中， 会创建数据块文件的快照， 但是为了节省磁盘空间， 其实是建立了数据块文件的硬链接（例如升级过程  
中的current与previous之间的数据块就是建立了硬链接） 。 当需要在current目录中追加写（append） 一个数据块时， 如果修改了current中的文件， previous中记录的快照文件也修改了， 所以这时就需要进行unlink操作。 方法很简单， 就是在同一个目录下创建一个临时文件， 然后将源文件的内容拷贝到临时文件中， 再将临时文件改名成源文件， 这样硬链接就解除了。  
append()方法后面的部分就比较简单了， 它将数据块文件与校验和文件移动到rbw目录中， 然后构造ReplicaBeingWritten对象， 更新volumeMap， 并返回新构造的ReplicaBeingWritten对象。

/** Append to a finalized replica
       * Change a finalized replica to be a RBW replica and 
       * bump its generation stamp to be the newGS
       * 
       * @param bpid block pool Id
       * @param replicaInfo a finalized replica
       * @param newGS new generation stamp
       * @param estimateBlockLen estimate block length
       * @return a RBW replica
       * @throws IOException if moving the replica from finalized directory 
       *         to rbw directory fails
       */
      private ReplicaInPipeline append(String bpid,
          ReplicaInfo replicaInfo, long newGS, long estimateBlockLen)
          throws IOException {
    
        try (AutoCloseableLock lock = datasetLock.acquire()) {
          // If the block is cached, start uncaching it.
          if (replicaInfo.getState() != ReplicaState.FINALIZED) {
            throw new IOException("Only a Finalized replica can be appended to; "
                + "Replica with blk id " + replicaInfo.getBlockId() + " has state "
                + replicaInfo.getState());
          }
    
          // 将当前副本从缓存中移出。
          // If the block is cached, start uncaching it.
          cacheManager.uncacheBlock(bpid, replicaInfo.getBlockId());
    
          // If there are any hardlinks to the block, break them.  This ensures
          // we are not appending to a file that is part of a previous/ directory.
          replicaInfo.breakHardLinksIfNeeded();
    
          FsVolumeImpl v = (FsVolumeImpl)replicaInfo.getVolume();
    
          //构造新的ReplicaInfo对象
          ReplicaInPipeline rip = v.append(bpid, replicaInfo,  newGS, estimateBlockLen);
    
    
          if (rip.getReplicaInfo().getState() != ReplicaState.RBW) {
            throw new IOException("Append on block " + replicaInfo.getBlockId() +
                " returned a replica of state " + rip.getReplicaInfo().getState()
                + "; expected RBW");
          }
          // Replace finalized replica by a RBW replica in replicas map
          volumeMap.add(bpid, rip.getReplicaInfo());
          return rip;
        }
    
      }
    
    
    
      public ReplicaInPipeline append(String bpid, ReplicaInfo replicaInfo,
          long newGS, long estimateBlockLen) throws IOException {
    
        long bytesReserved = estimateBlockLen - replicaInfo.getNumBytes();
        if (getAvailable() < bytesReserved) {
          throw new DiskOutOfSpaceException("Insufficient space for appending to "
              + replicaInfo);
        }
    
        assert replicaInfo.getVolume() == this:
          "The volume of the replica should be the same as this volume";
    
        // 构造一个新的RBW文件
        // construct a RBW replica with the new GS
        File newBlkFile = new File(getRbwDir(bpid), replicaInfo.getBlockName());
        LocalReplicaInPipeline newReplicaInfo = new ReplicaBuilder(ReplicaState.RBW)
            .setBlockId(replicaInfo.getBlockId())
            .setLength(replicaInfo.getNumBytes())
            .setGenerationStamp(newGS)
            .setFsVolume(this)
            .setDirectoryToUse(newBlkFile.getParentFile())
            .setWriterThread(Thread.currentThread())
            .setBytesToReserve(bytesReserved)
            .buildLocalReplicaInPipeline();
    
        // 只有finalized 的副本才能被追加
        // Only a finalized replica can be appended.
        FinalizedReplica finalized = (FinalizedReplica)replicaInfo;
        // load last checksum and datalen
        newReplicaInfo.setLastChecksumAndDataLen(
            finalized.getVisibleLength(), finalized.getLastPartialChunkChecksum());
    
        // 更新volumeMap中副本的ReplicaInfo对象
        // rename meta file to rbw directory
        // rename block file to rbw directory
        newReplicaInfo.moveReplicaFrom(replicaInfo, newBlkFile);
    
        reserveSpaceForReplica(bytesReserved);
    
        //返回构造的ReplicaInfo对象
        return newReplicaInfo;
      }

\*\*recoverAppend()\*\*方法用于恢复一个失败的追加写操作。 recoverAppend()方法是在Datanode进行数据流管道恢复操作时， 在BlockReceiver的构造方法中调用的。 调用完recoverAppend()方法后， 数据流管道中所有副本的状态都恢复至RBW， 并且副本的时间  
戳和数据块文件的长度也都一致， 可以继续通过数据流管道执行追加写（append） 操作  
了。  
recoverAppend()会首先调用recoverCheck()方法对副本对应的RelicaInfo对象进行检查。 然后判断如果副本状态为FINALIZED， 则重新调用append()方法将副本状态恢复为RBW（) ； 如果副本状态为RBW， 则更新副本的时间戳。

@Override  // FsDatasetSpi
      public ReplicaHandler recoverAppend(
          ExtendedBlock b, long newGS, long expectedBlockLen) throws IOException {
        LOG.info("Recover failed append to " + b);
    
        while (true) {
          try {
    
            try (AutoCloseableLock lock = datasetLock.acquire()) {
    
              //先调用recoverCheck()方法对ReplicaInfo进行检查
              ReplicaInfo replicaInfo = recoverCheck(b, newGS, expectedBlockLen);
    
    
              FsVolumeReference ref = replicaInfo.getVolume().obtainReference();
              ReplicaInPipeline replica;
              try {
    
    
                // change the replica's state/gs etc.
                // 之后更改副本的状态以及时间戳
                if (replicaInfo.getState() == ReplicaState.FINALIZED) {
                  //FINALIZED状态的副本变为RBW状态
                  replica = append(b.getBlockPoolId(), replicaInfo,
                                   newGS, b.getNumBytes());
                } else { //RBW
    
                  //RBW状态的副本， 则直接更新时间戳即可
                  replicaInfo.bumpReplicaGS(newGS);
                  replica = (ReplicaInPipeline) replicaInfo;
    
                }
              } catch (IOException e) {
                IOUtils.cleanup(null, ref);
                throw e;
              }
              return new ReplicaHandler(replica, ref);
            }
    
    
          } catch (MustStopExistingWriter e) {
            e.getReplicaInPipeline()
                .stopWriter(datanode.getDnConf().getXceiverStopTimeout());
          }
        }
      }

recoverCheck()方法会对副本对应的ReplicaInfo对象进行检查， 主要是检查副本状态、 时间戳、 数据块文件长度是否正常。

private ReplicaInfo recoverCheck(ExtendedBlock b, long newGS, 
          long expectedBlockLen) throws IOException, MustStopExistingWriter {
        ReplicaInfo replicaInfo = getReplicaInfo(b.getBlockPoolId(), b.getBlockId());
        
        // check state
        // 检查副本状态， 只有FINALIZED以及RBW状态的副本才能进行追加写操作
        if (replicaInfo.getState() != ReplicaState.FINALIZED &&
            replicaInfo.getState() != ReplicaState.RBW) {
          throw new ReplicaNotFoundException(
              ReplicaNotFoundException.UNFINALIZED_AND_NONRBW_REPLICA + replicaInfo);
        }
    
        // check generation stamp
        // 检查时间戳， 时间戳的范围应当是在原记录的时间戳与新时间戳之间
        long replicaGenerationStamp = replicaInfo.getGenerationStamp();
        if (replicaGenerationStamp < b.getGenerationStamp() ||
            replicaGenerationStamp > newGS) {
          throw new ReplicaNotFoundException(
              ReplicaNotFoundException.UNEXPECTED_GS_REPLICA + replicaGenerationStamp
              + ". Expected GS range is [" + b.getGenerationStamp() + ", " + 
              newGS + "].");
        }
    
        //
        // stop the previous writer before check a replica's length
        long replicaLen = replicaInfo.getNumBytes();
        if (replicaInfo.getState() == ReplicaState.RBW) {
          ReplicaInPipeline rbw = (ReplicaInPipeline) replicaInfo;
    
          if (!rbw.attemptToSetWriter(null, Thread.currentThread())) {
            throw new MustStopExistingWriter(rbw);
          }
          // check length: bytesRcvd, bytesOnDisk, and bytesAcked should be the same
          // 确认bytesRcvd、 bytesOnDisk和bytesAcked相同
          if (replicaLen != rbw.getBytesOnDisk() 
              || replicaLen != rbw.getBytesAcked()) {
            throw new ReplicaAlreadyExistsException("RBW replica " + replicaInfo + 
                "bytesRcvd(" + rbw.getNumBytes() + "), bytesOnDisk(" + 
                rbw.getBytesOnDisk() + "), and bytesAcked(" + rbw.getBytesAcked() +
                ") are not the same.");
          }
        }
        
        // check block length
        // 确认副本长度和期待长度相同， 否则抛出异常
        if (replicaLen != expectedBlockLen) {
          throw new IOException("Corrupted replica " + replicaInfo + 
              " with a length of " + replicaLen + 
              " expected length is " + expectedBlockLen);
        }
        
        return replicaInfo;
      }

\*\*recoverClose()\*\*是在Datanode进行写入操作的数据流管道恢复时， 在DataXceiver.writeBlock()方法中调用的。 recoverClose()方法的实现类似于recoverAppend()方法

@Override // FsDatasetSpi
      public Replica recoverClose(ExtendedBlock b, long newGS,
          long expectedBlockLen) throws IOException {
        LOG.info("Recover failed close " + b);
        while (true) {
          try {
            try (AutoCloseableLock lock = datasetLock.acquire()) {
              // check replica's state
              ReplicaInfo replicaInfo = recoverCheck(b, newGS, expectedBlockLen);
              // bump the replica's GS
              replicaInfo.bumpReplicaGS(newGS);
              // finalize the replica if RBW
              if (replicaInfo.getState() == ReplicaState.RBW) {
                finalizeReplica(b.getBlockPoolId(), replicaInfo);
              }
              return replicaInfo;
            }
          } catch (MustStopExistingWriter e) {
            e.getReplicaInPipeline()
                .stopWriter(datanode.getDnConf().getXceiverStopTimeout());
          }
        }
      }

finalizeBlock()方法用于提交数据块。 当Datanode完成了一个数据块的写操作并成功接收到数据流管道下游节点的ack时， 会在PacketResponder.run()方法中调用finalizeBlock()提交新写入的数据块。  
finalizeBlock()方法调用了私有的finalizeReplica()方法。 这个方法首先判断副本的当前状态是不是RUR， 如果副本的当前状态是RUR， 并且该副本进行恢复操作前的状态是FINALIZED， 那么直接将副本的状态转换为FINALIZED即可； 如果是其他状态， 则调用FsVolumeImpl.addFinalizedBlock()方法将数据块文件移动到finalize目录下。 然后构造新的FinalizedReplica对象并更新FsDatasetImpl.volumeMap

private ReplicaInfo finalizeReplica(String bpid, ReplicaInfo replicaInfo)
          throws IOException {
        try (AutoCloseableLock lock = datasetLock.acquire()) {
          // Compare generation stamp of old and new replica before finalizing
          if (volumeMap.get(bpid, replicaInfo.getBlockId()).getGenerationStamp()
              > replicaInfo.getGenerationStamp()) {
            throw new IOException("Generation Stamp should be monotonically "
                + "increased.");
          }
    
          ReplicaInfo newReplicaInfo = null;
          // 如果副本处于恢复状态， 并且恢复前的状态为FINALIZED
          // 则直接将副本状态转换为FINALIZED即可
          if (replicaInfo.getState() == ReplicaState.RUR &&
              replicaInfo.getOriginalReplica().getState()
              == ReplicaState.FINALIZED) {
            newReplicaInfo = replicaInfo.getOriginalReplica();
            ((FinalizedReplica)newReplicaInfo).loadLastPartialChunkChecksum();
          } else {
            //对于其他情况， 则首先获取FsVolumeImpl对象
            FsVolumeImpl v = (FsVolumeImpl)replicaInfo.getVolume();
            if (v == null) {
              throw new IOException("No volume for block " + replicaInfo);
            }
            // 然后调用addFinalizedBlock()方法将数据块移至finalized目录下
            newReplicaInfo = v.addFinalizedBlock(
                bpid, replicaInfo, replicaInfo, replicaInfo.getBytesReserved());
            if (v.isTransientStorage()) {
              releaseLockedMemory(
                  replicaInfo.getOriginalBytesReserved()
                      - replicaInfo.getNumBytes(),
                  false);
              ramDiskReplicaTracker.addReplica(
                  bpid, replicaInfo.getBlockId(), v, replicaInfo.getNumBytes());
              datanode.getMetrics().addRamDiskBytesWrite(replicaInfo.getNumBytes());
            }
          }
          assert newReplicaInfo.getState() == ReplicaState.FINALIZED
              : "Replica should be finalized";
    
          //更新volumeMap中的副本信息
          volumeMap.add(bpid, newReplicaInfo);
    
          //返回新构造的ReplicaInfo对象
          return newReplicaInfo;
        }
      }

\*\*unfinalizeBlock()\*\*方法用于删除TEMPORARY状态的副本， 也就是复制失败的副本，这个方法直接检查副本状态， 删除数据块文件以及对应的校验和文件， 然后从FsDatasetImpl.volumeMap中移除副本的ReplicaInfo。

/**
       * Remove the temporary block file (if any)
       */
      @Override // FsDatasetSpi
      public void unfinalizeBlock(ExtendedBlock b) throws IOException {
        try (AutoCloseableLock lock = datasetLock.acquire()) {
          ReplicaInfo replicaInfo = volumeMap.get(b.getBlockPoolId(),
              b.getLocalBlock());
          if (replicaInfo != null &&
              replicaInfo.getState() == ReplicaState.TEMPORARY) {
            // remove from volumeMap
            // 从内存volumeMap 移除
            volumeMap.remove(b.getBlockPoolId(), b.getLocalBlock());
    
            // delete the on-disk temp file
            // 删除磁盘文件
            if (delBlockFromDisk(replicaInfo)) {
              LOG.warn("Block " + b + " unfinalized and removed. ");
            }
            if (replicaInfo.getVolume().isTransientStorage()) {
              ramDiskReplicaTracker.discardReplica(b.getBlockPoolId(),
                  b.getBlockId(), true);
            }
          }
        }
      }

\*\*invalidate()\*\*方法也是用来删除副本的， 与unfinalizeBlock()方法不同的是， **invalidate()方法是用于删除RBW状态以及FINALIZED状态的副本的。** 因为副本有可能被短路读取（RBW状态以及FINALIZED状态的副本对客户端是可见的） ， 并且可能存储在Datanode  
与客户端的短路读取共享内存中 ， 所以invalidate()方法需要从短路读取的共享内存中移出副本的引用。 然后invalidate()方法会将副本从Datanode缓存中删除， 最后invalidate()方法会调用异步的磁盘文件删除方法FsDatasetAsyncDiskService.deleteAsync()将副本的数据块文件从磁盘上删除.  
异步的磁盘操作降低了方法的响应时间， 底层就是使用线程池进行异步操作的。

private void invalidate(String bpid, Block[] invalidBlks, boolean async)
          throws IOException {
        final List<String> errors = new ArrayList<String>();
        for (int i = 0; i < invalidBlks.length; i++) {
          final ReplicaInfo removing;
          final FsVolumeImpl v;
          try (AutoCloseableLock lock = datasetLock.acquire()) {
            // 获取内存中的副本信息
            final ReplicaInfo info = volumeMap.get(bpid, invalidBlks[i]);
    
    
            if (info == null) {
    
              ReplicaInfo infoByBlockId =  volumeMap.get(bpid, invalidBlks[i].getBlockId());
    
              if (infoByBlockId == null) {
                // It is okay if the block is not found -- it
                // may be deleted earlier.
                LOG.info("Failed to delete replica " + invalidBlks[i]
                    + ": ReplicaInfo not found.");
              } else {
                errors.add("Failed to delete replica " + invalidBlks[i]
                    + ": GenerationStamp not matched, existing replica is "
                    + Block.toString(infoByBlockId));
              }
              continue;
            }
    
            v = (FsVolumeImpl)info.getVolume();
            if (v == null) {
              errors.add("Failed to delete replica " + invalidBlks[i]
                  +  ". No volume for replica " + info);
              continue;
            }
            try {
              File blockFile = new File(info.getBlockURI());
              if (blockFile != null && blockFile.getParentFile() == null) {
                errors.add("Failed to delete replica " + invalidBlks[i]
                    +  ". Parent not found for block file: " + blockFile);
                continue;
              }
            } catch(IllegalArgumentException e) {
              LOG.warn("Parent directory check failed; replica " + info
                  + " is not backed by a local file");
            }
    
            //从volumeMap中移除副本信息
            removing = volumeMap.remove(bpid, invalidBlks[i]);
            addDeletingBlock(bpid, removing.getBlockId());
            if (LOG.isDebugEnabled()) {
              LOG.debug("Block file " + removing.getBlockURI()
                  + " is to be deleted");
            }
            if (removing instanceof ReplicaInPipeline) {
              ((ReplicaInPipeline) removing).releaseAllBytesReserved();
            }
          }
    
          if (v.isTransientStorage()) {
    
            RamDiskReplica replicaInfo =
              ramDiskReplicaTracker.getReplica(bpid, invalidBlks[i].getBlockId());
            if (replicaInfo != null) {
              if (!replicaInfo.getIsPersisted()) {
                datanode.getMetrics().incrRamDiskBlocksDeletedBeforeLazyPersisted();
              }
              ramDiskReplicaTracker.discardReplica(replicaInfo.getBlockPoolId(),
                replicaInfo.getBlockId(), true);
            }
          }
    
          // 如果副本正在被短路读取， 并且副本信息存储在短路读取的共享内存中， 则删除
          // If a DFSClient has the replica in its cache of short-circuit file
          // descriptors (and the client is using ShortCircuitShm), invalidate it.
          datanode.getShortCircuitRegistry().processBlockInvalidation(
                    new ExtendedBlockId(invalidBlks[i].getBlockId(), bpid));
    
          // 如果副本在Datanode中被缓存， 则从缓存中删除当前副本
          // If the block is cached, start uncaching it.
          cacheManager.uncacheBlock(bpid, invalidBlks[i].getBlockId());
    
          try {
            if (async) {
              // Delete the block asynchronously to make sure we can do it fast
              // enough.
              // It's ok to unlink the block file before the uncache operation
              // finishes.
              
              // 调用异步的磁盘文件删除功能， 删除数据块文件以及校验文件
              asyncDiskService.deleteAsync(v.obtainReference(), removing,
                  new ExtendedBlock(bpid, invalidBlks[i]),
                  dataStorage.getTrashDirectoryForReplica(bpid, removing));
            } else {
              asyncDiskService.deleteSync(v.obtainReference(), removing,
                  new ExtendedBlock(bpid, invalidBlks[i]),
                  dataStorage.getTrashDirectoryForReplica(bpid, removing));
            }
          } catch (ClosedChannelException e) {
            LOG.warn("Volume " + v + " is closed, ignore the deletion task for " +
                "block " + invalidBlks[i]);
          }
        }
        if (!errors.isEmpty()) {
          StringBuilder b = new StringBuilder("Failed to delete ")
            .append(errors.size()).append(" (out of ").append(invalidBlks.length)
            .append(") replica(s):");
          for(int i = 0; i < errors.size(); i++) {
            b.append("\n").append(i).append(") ").append(errors.get(i));
          }
          throw new IOException(b.toString());
        }
      }

initReplicaRecovery()当客户端写文件时异常退出， 且未能正常关闭文件时， 文件的最后一个数据块的所有副本可能处于不同的状态。 当客户端再次打开这个文件进行操作时， 就需要先对最后一个数据块进行块恢复操作（BlockRecovery） ， 使得该数据块的所有副本状态致， 然后才能进行后续的操作。Namenode初始化块恢复操作时， **会先选定一个主恢复数据节点（Primary Datanode）来控制整个数据块恢复流程**。 然后Namenode会通过心跳携带LeaseRecovery指令给主恢复数据节点开始块恢复操作， 主恢复数据节点会通过调用InterDatanodeProtocol接口与数据流管道中的其他Datanode通信， 控制整个数据块恢复流程。  
主恢复节点首先调用InterDatanodeProtocol.initReplicaRecovery()方法获取数据流管道中各个Datanode的副本恢复信息， 然后从所有副本中选取一个最好的状态作为所有副本恢复操作的目标状态。 最后主恢复节点会调用InterDatanodeProtocol.updateReplicaUnderRecovery()方法同步所有Datanode上该副本的状态至目标状态。  
FsDatasetImpl.initReplicaRecovery()方法就用于获取当前Datanode上指定数据块副本的状态。 这个方法会返回一个ReplicaRecoveryInfo对象， 它包含了该副本在恢复前的信息和状态。 这个方法的实现比较简单， 首先从volumeMap中获取该副本的信息， 如果有写线程  
正在写这个副本， 则停止写线程以保证数据块的恢复过程不受写线程的影响。 之后检查副本的时间戳以及恢复标识（recoveryId） ， 这里的恢复标识就是Namenode为该数据块分配的新的时间戳。 当恢复操作完成后， 副本的时间戳将会被设置为恢复标识。 如果当前副本  
的时间戳大于恢复标识， 则当前恢复操作是一个过期的恢复操作， 抛出异常。 检查完时间戳和恢复标识之后， initReplicaRecovery()方法将会构造一个ReplicaUnderRecovery对象，表明当前副本正在进行副本恢复操作， 副本状态为RUR， 然后使用这个ReplicaUnderRecovery更新volumeMap中副本的信息。

static ReplicaRecoveryInfo initReplicaRecoveryImpl(String bpid, ReplicaMap map,
          Block block, long recoveryId)
              throws IOException, MustStopExistingWriter {
    
        // 从volumeMap中获取副本信息
        final ReplicaInfo replica = map.get(bpid, block.getBlockId());
        LOG.info("initReplicaRecovery: " + block + ", recoveryId=" + recoveryId + ", replica=" + replica);
    
        // //确认副本存在
        // check replica
        if (replica == null) {
          return null;
        }
    
        // 如果有写线程则停止写线程
        // stop writer if there is any
        if (replica.getState() == ReplicaState.TEMPORARY ||
            replica.getState() == ReplicaState.RBW) {
          final ReplicaInPipeline rip = (ReplicaInPipeline)replica;
    
          //确认接收的数据大于可见的数据
          if (!rip.attemptToSetWriter(null, Thread.currentThread())) {
            throw new MustStopExistingWriter(rip);
          }
    
          // check replica bytes on disk.
          // 检查副本文件
          if (replica.getBytesOnDisk() < replica.getVisibleLength()) {
            throw new IOException("getBytesOnDisk() < getVisibleLength(), rip="
                + replica);
          }
    
          //check the replica's files
          checkReplicaFiles(replica);
        }
        //
        // 检查副本时间戳
        // check generation stamp
        if (replica.getGenerationStamp() < block.getGenerationStamp()) {
          throw new IOException(
              "replica.getGenerationStamp() < block.getGenerationStamp(), block="
              + block + ", replica=" + replica);
        }
    
        // 检查recoveryId， 确认不是一个过期的恢复操作
        // check recovery id
        if (replica.getGenerationStamp() >= recoveryId) {
          throw new IOException("THIS IS NOT SUPPOSED TO HAPPEN:"
              + " replica.getGenerationStamp() >= recoveryId = " + recoveryId
              + ", block=" + block + ", replica=" + replica);
        }
    
        //构造ReplicaUnderRecovery对象， 如果已经存在RUR对象， 则更新recoveryId
        //check RUR
        final ReplicaInfo rur;
        if (replica.getState() == ReplicaState.RUR) {
          rur = replica;
          //注意， 这里要判断这个恢复操作是否是一个过期或者重复的操作
          if (rur.getRecoveryID() >= recoveryId) {
            throw new RecoveryInProgressException(
                "rur.getRecoveryID() >= recoveryId = " + recoveryId
                + ", block=" + block + ", rur=" + rur);
          }
          final long oldRecoveryID = rur.getRecoveryID();
          rur.setRecoveryID(recoveryId);
          LOG.info("initReplicaRecovery: update recovery id for " + block
              + " from " + oldRecoveryID + " to " + recoveryId);
        }
        else {
          //  ReplicaUnderRecovery对象不存在， 则构造RUR对象， 并更新volumeMap
          rur = new ReplicaBuilder(ReplicaState.RUR)
              .from(replica).setRecoveryId(recoveryId).build();
          map.add(bpid, rur);
          LOG.info("initReplicaRecovery: changing replica state for "
              + block + " from " + replica.getState()
              + " to " + rur.getState());
          if (replica.getState() == ReplicaState.TEMPORARY || replica
              .getState() == ReplicaState.RBW) {
            ((ReplicaInPipeline) replica).releaseAllBytesReserved();
          }
        }
        // 返回副本恢复前的信息
        return rur.createInfo();
      }

\*\*updateReplicaUnderRecovery()\*\*主恢复节点调用InterDatanodeProtocol.initReplicaRecovery()方法后， 将会获得数据流管道中所有恢复数据块的副本状态， **主恢复节点将会在所有副本中选取一个最好的状态作为恢复操作的目标状态**。 判断副本状态的标准是FINALIZED>RBW>RWR>RUR>TEMPORARY， 也就是副本的持久化程度越高则副本状态越好。 如果所有副本中最好的状态FINALIZED， 那么就以这个FINALIZED副本长度为目标， 其他和这个副本不一致的都排除在恢复操作外； 如果最好的状态是RBW或者RWR， 那么就选择所有副本中数据块文件长度最短的副本作为目标副本。得到了副本恢复操作的目标状态以及参与恢复的Datanode列表后， 主恢复节点会对列表中的所有Datanode调用InterDatanodeProtocol.updateReplicaUnderRecovery()方法同步该数据节点上副本的状态至目标状态.  
FsDatasetImpl.updateReplicaUnderRecovery()方法就用于将当前数据节点上副本的状态同步至目标状态。 updateReplicaUnderRecovery()方法会首先检查副本是不是RUR状态， 检查副本在磁盘上的长度是否正常， 检查副本的数据块文件和校验文件是否匹配等。 然后updateReplicaUnderRecovery()方法会调用私有的updateReplicaUnderRecovery()将当前副本的时间戳更新为恢复标识（ recoveryId） ， 将副本的数据块文件长度截短为目标长度并重新计算校验和。 最后updateReplicaUnderRecovery()方法会将副本由RUR状态更改为FINALIZED状态， 并更新FsDatasetImpl.volumeMap。 至此， 恢复操作完成， 数据流管道中的所有副本状态一致。

@Override // FsDatasetSpi
      public Replica updateReplicaUnderRecovery(
                                        final ExtendedBlock oldBlock,
                                        final long recoveryId,
                                        final long newBlockId,
                                        final long newlength) throws IOException {
        try (AutoCloseableLock lock = datasetLock.acquire()) {
          // get replica
          // 获取副本信息
          final String bpid = oldBlock.getBlockPoolId();
          final ReplicaInfo replica = volumeMap.get(bpid, oldBlock.getBlockId());
          LOG.info("updateReplica: " + oldBlock
              + ", recoveryId=" + recoveryId
              + ", length=" + newlength
              + ", replica=" + replica);
    
          // check replica
          // 检查副本信息是否存在
          if (replica == null) {
            throw new ReplicaNotFoundException(oldBlock);
          }
    
          //check replica state
          // 判断副本状态， 如果不是RUR则抛出异常
          if (replica.getState() != ReplicaState.RUR) {
            throw new IOException("replica.getState() != " + ReplicaState.RUR
                + ", replica=" + replica);
          }
    
          // 检查副本长度是否正常
          // check replica's byte on disk
          if (replica.getBytesOnDisk() != oldBlock.getNumBytes()) {
            throw new IOException("THIS IS NOT SUPPOSED TO HAPPEN:"
                + " replica.getBytesOnDisk() != block.getNumBytes(), block="
                + oldBlock + ", replica=" + replica);
          }
    
          //  检查副本的数据块文件和校验文件是否匹配
          //check replica files before update
          checkReplicaFiles(replica);
    
          // update replica
          // 将副本状态更新为目标状态
          final ReplicaInfo finalized = updateReplicaUnderRecovery(oldBlock
              .getBlockPoolId(), replica, recoveryId,
              newBlockId, newlength);
    
    
          boolean copyTruncate = newBlockId != oldBlock.getBlockId();
          if (!copyTruncate) {
            assert finalized.getBlockId() == oldBlock.getBlockId()
                && finalized.getGenerationStamp() == recoveryId
                && finalized.getNumBytes() == newlength
                : "Replica information mismatched: oldBlock=" + oldBlock
                + ", recoveryId=" + recoveryId + ", newlength=" + newlength
                + ", newBlockId=" + newBlockId + ", finalized=" + finalized;
          } else {
            assert finalized.getBlockId() == oldBlock.getBlockId()
                && finalized.getGenerationStamp() == oldBlock.getGenerationStamp()
                && finalized.getNumBytes() == oldBlock.getNumBytes()
                : "Finalized and old information mismatched: oldBlock=" + oldBlock
                + ", genStamp=" + oldBlock.getGenerationStamp()
                + ", len=" + oldBlock.getNumBytes()
                + ", finalized=" + finalized;
          }
          // check replica files after update
          // 检查副本的数据块文件和校验文件是否匹配
          checkReplicaFiles(finalized);
    
          //返回存储ID
          return finalized;
        }
      }
    
      // 私有的updateReplicaUnderRecovery()方法
      private ReplicaInfo updateReplicaUnderRecovery(
                                              String bpid,
                                              ReplicaInfo rur,
                                              long recoveryId,
                                              long newBlockId,
                                              long newlength) throws IOException {
        //check recovery id
        // 检查恢复标识， 如果是过期的恢复操作， 则抛出异常
        if (rur.getRecoveryID() != recoveryId) {
          throw new IOException("rur.getRecoveryID() != recoveryId = " + recoveryId
              + ", rur=" + rur);
        }
    
        boolean copyOnTruncate = newBlockId > 0L && rur.getBlockId() != newBlockId;
        // bump rur's GS to be recovery id
        if(!copyOnTruncate) {
          //更新副本时间戳到recoveryId
          rur.bumpReplicaGS(recoveryId);
        }
    
        //update length
        if (rur.getNumBytes() < newlength) {
          throw new IOException("rur.getNumBytes() < newlength = " + newlength
              + ", rur=" + rur);
        }
    
        if (rur.getNumBytes() > newlength) {
          if(!copyOnTruncate) {
            rur.breakHardLinksIfNeeded();
            //则调用truncate()方法截短数据块文件， 并重新计算校验和
            rur.truncateBlock(newlength);
            // update RUR with the new length
            rur.setNumBytes(newlength);
          } else {
            // Copying block to a new block with new blockId.
            // Not truncating original block.
            FsVolumeImpl volume = (FsVolumeImpl) rur.getVolume();
            ReplicaInPipeline newReplicaInfo = volume.updateRURCopyOnTruncate(
                rur, bpid, newBlockId, recoveryId, newlength);
            if (newReplicaInfo.getState() != ReplicaState.RBW) {
              throw new IOException("Append on block " + rur.getBlockId()
                  + " returned a replica of state " + newReplicaInfo.getState()
                  + "; expected RBW");
            }
    
            newReplicaInfo.setNumBytes(newlength);
            volumeMap.add(bpid, newReplicaInfo.getReplicaInfo());
            //将副本状态由RUR改为FINALIZED， 并更新volumeMap
            finalizeReplica(bpid, newReplicaInfo.getReplicaInfo());
          }
        }
        // finalize the block
        return finalizeReplica(bpid, rur);
      }

### 5.2.5 数据块汇报相关方法。 ###

FsDatasetImpl的getBlockReports()和getCacheReport()方法用于支持Datanode的块汇报以及缓存汇报操作。  
Datanode需要向HDFS集群中的所有Namenode发送块汇报.

**getBlockReports()** Datnode在启动时会通过DatanodeProtocol.blockReport()方法向Namenode汇报数据节点上存储的该命名空间的所有数据块副本的信息， 也就是Namenode对应的块池在当前Datanode上保存的所有数据块副本的信息.

Datanode会调用FsDatasetImpl.getBlockReports()方法获取指定块池在当前Datanode上保存的所有副本的信息， 这里的块池信息是由getBlockReports()方法的参数bpid指定的。需要注意的是， 块汇报只包含FINALIZED状态以及UnderConstruction（包括RBW、RWR、 RUR)状态的副本， **不包含TEMPORARY状态的副本**。 同时由于Hadoop 2.6版本中引入了异构的存储目录， 块汇报的单位也由整个Datanode改为Datanode的存储目录， 所以getBlockReports()方法的返回值是一个Map对象， 这个Map保存了每个存储目录的DatanodeStorage对象（通过FsVolumeImp.toDataStorage()方法获得） 与这个存储目录保存的所有副本信息之间的映射getBlockReports()方法的实现非常简单， 就是遍历FsDatasetImpl.volumeMap字段中保存的所有副本， 将FINALIZED状态以及RBW、 RWR、RUR状态的副本放入存储目录对应的集合中， 然后返回。

@Override
      public Map<DatanodeStorage, BlockListAsLongs> getBlockReports(String bpid) {
    
        //构造返回值对象
        Map<DatanodeStorage, BlockListAsLongs> blockReportsMap =  new HashMap<DatanodeStorage, BlockListAsLongs>();
    
        Map<String, BlockListAsLongs.Builder> builders =  new HashMap<String, BlockListAsLongs.Builder>();
    
        List<FsVolumeImpl> curVolumes = null;
        try (AutoCloseableLock lock = datasetLock.acquire()) {
          // 0 : /tools/hadoop-3.2.1/data/hdfs/data
          curVolumes = volumes.getVolumes();
          for (FsVolumeSpi v : curVolumes) {
            // v:  FsVolumeImple : /tools/hadoop-3.2.1/data/hdfs/data
            //  builders : DS-23b73402-fa18-451d-930f-831de2e8686e -> {BlockListAsLongs$Builder@4462}
            builders.put(v.getStorageID(), BlockListAsLongs.builder(maxDataLength));
          }
          // volumeMap :  BP-451827885-192.168.8.156-1584099133244 -> {FoldedTreeSet@4644}  size = 5077
          Set<String> missingVolumesReported = new HashSet<>();
          for (ReplicaInfo b : volumeMap.replicas(bpid)) {
            // skip PROVIDED replicas.
            // 跳过外部存储
            if (b.getVolume().getStorageType() == StorageType.PROVIDED) {
              continue;
            }
            // 获取存储id  : DS-23b73402-fa18-451d-930f-831de2e8686e
            String volStorageID = b.getVolume().getStorageID();
    
            if (!builders.containsKey(volStorageID)) {
              if (!missingVolumesReported.contains(volStorageID)) {
                LOG.warn("Storage volume: " + volStorageID + " missing for the"
                    + " replica block: " + b + ". Probably being removed!");
                // 添加丢失卷信息
                missingVolumesReported.add(volStorageID);
              }
              continue;
            }
            LOG.info(" volStorageID: {} ,blockID : {} , bState  : {} ",volStorageID, b.getBlockId() , b.getState());
            switch(b.getState()) {
            case FINALIZED:
            case RBW:
            case RWR:
              // 写入FINALIZED,RBW , RWR 到 builders
              builders.get(volStorageID).add(b);
              break;
            case RUR:
              // 构建中的副本 写入builders
              ReplicaInfo orig = b.getOriginalReplica();
              builders.get(volStorageID).add(orig);
              break;
            case TEMPORARY:
              break;
            default:
              assert false : "Illegal ReplicaInfo state.";
            }
          }
        }
    
        for (FsVolumeImpl v : curVolumes) {
          // 写入所有的副本
          blockReportsMap.put(v.toDatanodeStorage(),
                              builders.get(v.getStorageID()).build());
        }
    
        return blockReportsMap;
      }

getCacheReport()方法与getBlockReports()方法一样， 用于在Datanode通过DatanodeProtocol.cacheReport()向Namenode进行缓存汇报时， 获取对应块池在当前Datanode上缓存的所有数据块副本的信息.

getCacheReport()方法调用了FsDatasetCache.getCachedBlocks()方法，getCachedBlocks()会遍历FsDatasetCache.mappableBlockMap字段中保存的所有缓存数据块， 将状态为CACHED的缓存数据块保存到一个集合中， 然后返回。

@Override // FsDatasetSpi
      public List<Long> getCacheReport(String bpid) {
        return cacheManager.getCachedBlocks(bpid);
      }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/3c7357522ab444f89ce0640974494fd2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 1]: https://img-blog.csdnimg.cn/20201114171313776.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 2]: https://img-blog.csdnimg.cn/20201114180219455.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0,size_16,color_FFFFFF,t_70#pic_center