HDFS3源码分析之Namenode租约与错误恢复(二)

看完Client端的错误恢复和追加写, 也就是pipeline恢复的过程后, 再来看看Namenode端的相关异常处理, 不过这部分内容不多, 放最后简要说下

主要篇幅是介绍一个, 与写流程间相对独立, 但在整个HDFS和分布式文件系统中很重要的概念租约, 以及由NN主导的触发租约恢复的概念

文档正在完善ing, 主要是补图…

0x00. 租约の概念

租约的概念形象说就像读写锁里面带有过期时间(TTL)的“写锁”, 也就是说一个文件在写入的时候仅有一个载体能够操作它. 它实现比较简单, 但是不可或缺.

核心对象有两个:

• Lease Manager (包租婆): 它记录着所有准备写文件的client(借租的记录) [Server]
  1. Monitor (看门狗): 定时检查租约是否过期, 默认2s检查一次
  2. Lease (租约): 租约的本体(数据结构), 维护”债主名”, “最后续约时间” , “此债主借的所有文件”三个参数
• Lease Renewer(续约器): client使用它来维系与NN的租约关系, 超过30s就请求NN更新一下 [Client –> NN]

它有两个超时时间: (不可调整)

• 软超时(1min): 超过1min原客户端没有续租(租房合同到期), 那么此时其他客户端(房客)就可让此文件进入租约恢复, 然后获得它的租约(入住)
• 硬超时(1h): 租约软超时后如果一直没有客户端来写(入住)此文件, 那它不会自动释放, 而是一直等到超过1h后, 则由NN(房产局)拥有文件租约并进入租约恢复, 完成后释放租约.

这两个超时概念需要清晰理解, 一个文件如果触发硬超时, 那必然是某种原因(比如Client挂了)先触发了软超时, 然后超时之后仍然长时间没人尝试重新写它.

上面的关键在于LeaseManger和Lease的数据结构, 搞清楚它们的关系, 后面理解就容易许多, 下面引用一张美团Tech 画的挺好的图:

其中sortedLeases 是一个按时间先后排序的集合, 第一个元素就是未更新租约时间最长的.

然后租约本身的数据结构中, 维护了一个Set<INodeId>, 从而对应多个文件. 所以在后面移除租约的时候, 会把对应的所有文件都依次移除. 细节可参考源码

0x01. 租约の管理

搞清楚租约和包租婆之间的关系后, 这个部分内容比较简单, 主要就是针对租约的CURD(增删改查)操作的实现, 当然也是基础 (其中增删的代码就不说了)

A. 租约添加

添加租约的情况比较简单, 写/追加写文件, 会在租约管理器里添加该文件(INode)的映射记录, 暂略

B. 租约检查

检查租约分两类情况, 一是租约管理器每2s定时检查所有租约(自动), 另一种是特定操作在NN执行之前会先去检查租约是否正常(手动). 先看看定时检查逻辑:

1. 每2s查看一下按时间排序的Set集合, 看看第一个租约元素是否硬超时 (1h)
2. 如果超时, 尝试逐个恢复此租约下的所有文件
3. 如果恢复失败, 先跳当前文件, 继续对此租约下剩余文件进行租约恢复. (路径非法的文件会被直接移除租约)

// sortedLeases是按租约更新时间排序的Set<Lease>集合, 排在最前的就是时间最早的
synchronized boolean checkLeases() {
    boolean needSync = false;
    long start = monotonicNow();

    // 只要有超过硬超时的租约, 且满足基本条件. 则尝试触发租约恢复
    while(!sortedLeases.isEmpty() && sortedLeases.first().expiredHardLimit() &&
          !isMaxLockHoldToReleaseLease(start)) {
      Lease leaseToCheck = sortedLeases.first();
      final List<Long> removing = new ArrayList<>();
        
      // 需要给最早的租约们一个备份, 因为internalReleaseLease()方法
      // 会删除空文件对应的租约, 这可能会触发并发修改的异常 (which?)
      Collection<Long> files = leaseToCheck.getFiles();
      Long[] leaseINodeIds = files.toArray(new Long[files.size()]);
      FSDirectory fsd = fsnamesystem.getFSDirectory();
      String p = null;
      String newHolder = getInternalLeaseHolder(); // 生成新租户名:"HDFS_NameNode-时间戳"
      for(Long id : leaseINodeIds) { // 遍历超时租约持有者拥有的所有文件(正常来说, 它们应该都超时了)
        try {
          INodesInPath iip = INodesInPath.fromINode(fsd.getInode(id));
          p = iip.getPath();
          validatePath(p);
          final INodeFile lastINode = iip.getLastINode().asFile();
          if (fsnamesystem.isFileDeleted(lastINode)) { // 如果文件已被删除
            removeLease(lastINode.getId()); // 直接移除租约
            continue;
          }
            
          boolean completed = false;
          try { // 返回true表示租约恢复成功, 文件正常关闭
            completed = fsnamesystem.internalReleaseLease(leaseToCheck, p, iip, newHolder);
          } catch (IOException e) { // 如果租约恢复异常, 先跳过此文件, 之后再重试
            LOG.warn("Cannot release the path xx in lease xx. Retry later.");
            continue;
          }
            
          // 如果发生了租约恢复, 需要在editLog中同步记录
          if (!needSync && !completed) needSync = true;
        } catch (IOException e) { // 如果path不合法抛出异常, 则最后一次性移除租约
          removing.add(id);
        }
        if (isMaxLockHoldToReleaseLease(start)) break;
      } // end loop files

      // 依次移除path非法的文件租约
      for(Long id : removing) removeLease(leaseToCheck, id);
    } // end while
    return needSync;
  }

C. 租约删除

知道一个文件什么时候租约会被删除, 或者释放是很重要的, 因为这涉及到没有正常释放的进入租约恢复的情况

• 完成文件(complet file)后
• 删除文件 (包括重命名)
• 文件或路径不存在导致租约恢复失败后

D. 租约恢复(*)

不管是软超时导致的租约持有者变更, 还是硬超时导致的持有者变成Namenode自身, 租约恢复简单说是是对租约的修改, 它在以下几个地方可能会触发:

1. 硬超时后, Namenode检测到进入错误恢复, 所有权先变成NN自身, 恢复完成再移除
2. 软超时后, 其他客户端调用写/追加写此文件, 会触发租约恢复使文件写权转移
3. 直接RPC调用recoverLease() , 强制进行租约恢复释放租约, 此时无视软超时限制. (比如DFSAdmin释放租约命令)
4. 直接调用truncate-RPC剔除文件部分数据时 (FShell的新命令)

其中第一种情况是NN后台自动触发, 属于一类; 其他的情况都是Client手动触发, 属于另一类, 先来看看这一类的通用判断:

// 主动发起的租约恢复, 对象是以文件为单位
boolean recoverLeaseInternal(params.., String holder, boolean force) {
    INodeFile file = iip.getLastINode().asFile();
    // 如果文件处于UC态, 那么正常来说它一定在租约管理器中, 可以找到原有的租约持有者
    if (file.isUnderConstruction()) { 
      Lease lease = leaseManager.getLease(holder); // 通过当前发起者, 尝试找到对应的租约
      
      // 如果当前申请租约恢复的客户端和原有是一个, 则认为重复申请, 抛出异常
      if (!force && lease != null) {
        Lease leaseFile = leaseManager.getLease(file);
        if (leaseFile != null && leaseFile.equals(lease)) 
          throw new AlreadyBeingCreatedException("xx is already the current lease holder."));
      }

      // 如果是其他客户端发起, 则先找到文件原有的租约持有者
      FileUnderConstructionFeature uc = file.getFileUnderConstructionFeature();
      String clientName = uc.getClientName();
      lease = leaseManager.getLease(clientName);
      if (lease == null)
        throw new AlreadyBeingCreatedException("the file is UC but no leases found."));

      if (force) { // 强制发起租约恢复是无视软超时的
        return internalReleaseLease(lease, src, iip, holder);
      } else {
        if (lease.expiredSoftLimit()) { // 如果原持有租约软超时过期, 触发租约恢复
          if (internalReleaseLease(lease, src, iip, null)) {
            return true;
          } else {
            throw new RecoveryInProgressException("lease recovery is in progress.."));
          }
        } else {
          final BlockInfo lastBlock = file.getLastBlock();
          if (lastBlock != null && lastBlock.getBlockUCState() == UNDER_RECOVERY) {
            throw new RecoveryInProgressException("another recovery is in progress xx");
          } else {
            throw new AlreadyBeingCreatedException("Lease already created xx");
          }
        }
      }
    } else { // 文件不处于UC状态, 那么认为已正常关闭, 直接跳过所有租约恢复判断
      return true;
     }
  }

整体上的, 对租约的增删改查操作就说完了, 下面开始说其中的细节, 也就是主角—-租约恢复.

0x02. 租约の恢复

租约恢复是租约里相对最复杂的部分, 它是一个文件的租约没有被正常续约后超时触发的, 而续约超时, 多半说明写入操作没有正常完成(中断了), 也意味着

1. 文件本身(INode)处于中间状态
2. 文件对应3DN上的副本状态可能不一致 , 实际长度不统一
3. 租约没有被正常释放, 无法继续新的写操作

所以如果要继续对这个文件进行操作, 得先让文件和对应的块副本恢复到一个一致的状态, 然后再释放租约, 整个过程就称之为租约恢复了.

租约恢复通过判断文件块的状态, 大概分为两种情况:

• 一种是NN可以自己解决的, 可以认为只是异常处理 (简单)
• 另一种是需要DN参与进来的, 这是真正意义上的租约恢复 (复杂)

1. 租约恢复(前置)

流程图待补

1. NN的租约检查线程发现租约异常的文件. 开始逐个遍历检查
2. 检查文件的最后两个block的状态
   • 如果处于committed, 尝试直接complete它, 然后完成文件, 跳过后续步骤
   • 如果最后一个block处于UC态, 需要找到对应的DN列表, 然后选择主恢复DN, 带着新的块时间戳发给主DN
   • 如果处于非法状态, 或者其他异常, 则跳过当前文件恢复, 之后再重新单独尝试
3. 主DN接收到新的时间戳后, 会向剩余DN发送RPC请求每个DN上的副本信息
4. 然后根据所有DN返回的信息, 计算出一个状态最优的, 发RPC让所有DN统一
5. 完成后主DN给NN回复这次块恢复的结果
   • 正常, NN更新块的信息, 然后移除自己持有的租约
   • 异常, 说明此次恢复失败, 然后怎么感知呢? 恢复失败的文件如何处理呢? (这里有疑问)

上面说的两种租约恢复情况, 都在NN的一个方法中进行判断, 简单情况NN直接处理, 其他需要DN参与, 代码如下:

boolean internalReleaseLease(param.., String recoveryLeaseHolder) {

    final INodeFile pendingFile = iip.getLastINode().asFile();
    int nrBlocks = pendingFile.numBlocks();
    BlockInfo[] blocks = pendingFile.getBlocks();

    int nrCompleteBlocks;
    BlockInfo curBlock = null;
    for(nrCompleteBlocks = 0; nrCompleteBlocks < nrBlocks; nrCompleteBlocks++) {
      curBlock = blocks[nrCompleteBlocks];
      if(!curBlock.isComplete()) break;
    }

    // 1. 最佳情况, 文件的所有块全complete, 直接将文件完成, 返回恢复成功
    if(nrCompleteBlocks == nrBlocks) {
      finalizeINodeFileUnderConstruction(src, pendingFile, iip.getLatestSnapshotId(), false);
      return true;
    }

    // 2.1 倒数第2个块不是complete, 那必须为是committed状态, 否则抛出异常
    if(twoBlocksNotComplete(nrCompleteBlocks)) throw new IOException("xxx");

    // 2.2 如果倒数1,2个块至少处于committed状态, 接着判断
    final BlockInfo lastBlock = pendingFile.getLastBlock();
    BlockUCState lastBlockState = lastBlock.getBlockUCState();
    BlockInfo penultimateBlock = pendingFile.getPenultimateBlock();

    // 倒数第二个块如果存在, 判断是否满足最小副本数(默认1)
    boolean penultimateBlockMinStorage = penultimateBlock == null ||
                                         blockManager.hasMinStorage(penultimateBlock);

    switch(lastBlockState) {
    case COMPLETE: // 异常情况
      assert false : "Already checked that the last block is incomplete";
      break;
    case COMMITTED: // 2.3 如果倒数1,2个块副本数满足最小要求, 直接完成文件 (同全complete状态)
      if(penultimateBlockMinStorage && blockManager.hasMinStorage(lastBlock)) {
        finalizeINodeFileUnderConstruction(params..., false);
        return true;
      }
      // 2.4 如果倒数1,2个块没满足最小副本要求, 为了避免出现死循环, 直接抛出异常
      // 比如DN上没有满足条件的副本?
      throw new AlreadyBeingCreatedException(message);
    case UNDER_CONSTRUCTION:
    case UNDER_RECOVERY:
      BlockUnderConstructionFeature uc = lastBlock.getUnderConstructionFeature();
      BlockInfo recoveryBlock = uc.getTruncateBlock(); // 是否是truncate的块
      boolean truncateRecovery = recoveryBlock != null;
      boolean copyOnTruncate = truncateRecovery &&
                               recoveryBlock.getBlockId() != lastBlock.getBlockId();
      assert !copyOnTruncate ||
             recoveryBlock.getBlockId() < lastBlock.getBlockId() &&
             recoveryBlock.getGenerationStamp() < lastBlock.getGenerationStamp() &&
             recoveryBlock.getNumBytes() > lastBlock.getNumBytes() :
             "wrong recoveryBlock";

      // 如果lastBlock的位置未知, 则从BlockManager设置
      if (uc.getNumExpectedLocations() == 0) uc.setExpectedLocations(lastBlock...);
      // 3. 如果最后一个块长度是0, 认为是客户端申请block后还没写数据就挂了, 直接删除此块完成文件
      if (uc.getNumExpectedLocations() == 0 && lastBlock.getNumBytes() == 0) {
        pendingFile.removeLastBlock(lastBlock);
        finalizeINodeFileUnderConstruction(src, pendingFile,iip.getLatestSnapshotId(), false);
        return true;
      }
            
      // 如果最后一个块没有处于恢复状态或已经恢复超时
      // 那么准备开始进入真正的租约恢复, 先把它加入pendingRecovery队列中
      if (blockManager.addBlockRecoveryAttempt(lastBlock)) {
        // 生成一个数值更大的新时间戳
        long blockRecoveryId = nextGenerationStamp(blockManager.isLegacyBlock(lastBlock));
        if(copyOnTruncate) { // truncate相关, 暂略
          lastBlock.setGenerationStamp(blockRecoveryId);
        } else if(truncateRecovery) {
          recoveryBlock.setGenerationStamp(blockRecoveryId);
        }
      // 4. 如果最后一个块写入了数据, 处于UC/UR状态,进入正式的错误恢复 (并续约NN的持有)
      uc.initializeBlockRecovery(lastBlock, blockRecoveryId, true);
      lease = reassignLease(lease, src, recoveryLeaseHolder, pendingFile);
      leaseManager.renewLease(lease);
      // 租约恢复时期不能关闭文件, 需等待完成
      break;
    } // end switch case

    return false; // 返回false, 则代表正在进行租约恢复
  }

2. 租约恢复(正式)

这对应上面第二种情况, 需要DN参与的流程, 用时序图来看会很清晰, 如下所示: (图待画, 这是旧的先占个位)

sequenceDiagram
    participant A as Client
    participant B as DFSOutputStream
    participant C as DataStreamer
    participant D as NameNode
    %%participant E as DataNode

    A->>+B: close()
    B->>B: flushBuffer()
    alt packet非空
    B->>+B: 剩余数据包入队()    
    B->>-C: waitAndQueuePacket()
    activate C
    end
    B->>+B: 生成空尾包()
    B->>-C: 发尾包, 等待ack收尾
    deactivate C
    B->>D: RPC请求: complete()
    Note right of D: 此时NN释放租约
    B->>+B: closeThreads()
    B->>C: 关闭线程与socket
    B->>-B: 客户端停止更新租约
    B-->>-A: 写完成

然后来看看, NN部分就做两个事, 一是选出主恢复DN, 二是准备好环境信息随心跳包发送过去:

public void initializeBlockRecovery(params..) {
    // 0.1 设置当前block为UR态. 传入新的block时间戳
    setBlockUCState(UNDER_RECOVERY);
    blockRecoveryId = recoveryId;
    if (!startRecovery) return;

    if (replicas.length == 0) { //异常:文件没有找到副本, 标记主DN为-1, 退出
      primaryNodeIndex = -1;
      return;
    }
    
    // 0.2 为选取主恢复DN做准备
    boolean allLiveReplicasTriedAsPrimary = true;
    for (ReplicaUnderConstruction replica : replicas) {
      if (replica.isAlive()) { // 是否有心跳包
        allLiveReplicasTriedAsPrimary = allLiveReplicasTriedAsPrimary &&
                                        replica.getChosenAsPrimary();
      }
    }
    if (allLiveReplicasTriedAsPrimary) {
      for (ReplicaUnderConstruction replica : replicas) 
        replica.setChosenAsPrimary(false);
    }
    
    // 1.在这选取主恢复DN, 条件是最新给NN进行了心跳汇报的DN
    long mostRecentLastUpdate = 0;
    ReplicaUnderConstruction primary = null;
    primaryNodeIndex = -1;
    for (int i = 0; i < replicas.length; i++) {  // 跳过已被选做主恢复DN的节点
      if (!(replicas[i].isAlive() && !replicas[i].getChosenAsPrimary())) continue;

      final ReplicaUnderConstruction ruc = replicas[i];
      final long lastUpdate = ruc.getExpectedStorageLocation()
                                 .getDatanodeDescriptor().getLastUpdateMonotonic();
      if (lastUpdate > mostRecentLastUpdate) {
        primaryNodeIndex = i;
        primary = ruc;
        mostRecentLastUpdate = lastUpdate;
      }
    }
    // 2. 选好了主DN后, 随着"心跳包"下发指令给主DN, 并不是单独发RPC的模式
    if (primary != null) {
      primary.getExpectedStorageLocation().getDatanodeDescriptor()
                                          .addBlockToBeRecovered(blockInfo);
      primary.setChosenAsPrimary(true);
    }
  }

然后最后主DN做完了租约恢复后, 通过发送commitBlockSynchronization的RPC确认DN块恢复完成, 然后NN正式更新块信息和文件信息, 释放租约, 完成此次租约恢复

void commitBlockSynchronization(params...) {
    final String src;
    writeLock();
    boolean copyTruncate = false;
    BlockInfo truncatedBlock = null;
    try {
      // 一系列的检查, 包括safemode/写权限/文件是否被删?/提交的block是否存在等等..
      final BlockInfo storedBlock = getStoredBlock(ExtendedBlock.getLocalBlock(oldBlock));
      if (storedBlock == null) {
        if (deleteblock) {
          // This may be a retry attempt so ignore the failure to locate the block.
          return;
        } else {
          throw new IOException("Block (=" + oldBlock + ") not found");
        }
      }
      final long oldGenerationStamp = storedBlock.getGenerationStamp();
      final long oldNumBytes = storedBlock.getNumBytes();
      //
      // The implementation of delete operation (see @deleteInternal method)
      // first removes the file paths from namespace, and delays the removal
      // of blocks to later time for better performance. When
      // commitBlockSynchronization (this method) is called in between, the
      // blockCollection of storedBlock could have been assigned to null by
      // the delete operation, throw IOException here instead of NPE; if the
      // file path is already removed from namespace by the delete operation,
      // throw FileNotFoundException here, so not to proceed to the end of
      // this method to add a CloseOp to the edit log for an already deleted
      // file (See HDFS-6825).
      //
      if (storedBlock.isDeleted()) {
        throw new IOException("The blockCollection of " + storedBlock
            + " is null, likely because the file owning this block was"
            + " deleted and the block removal is delayed");
      }
      final INodeFile iFile = getBlockCollection(storedBlock);
      src = iFile.getFullPathName();
      if (isFileDeleted(iFile))
        throw new FileNotFoundException("likely due to delayed block removal");

      if ((!iFile.isUnderConstruction() || storedBlock.isComplete()) &&
          iFile.getLastBlock().isComplete()) { // 异常
        return;
      }

      truncatedBlock = iFile.getLastBlock();
      final long recoveryId = truncatedBlock.getUnderConstructionFeature()
                                            .getBlockRecoveryId();
      copyTruncate = truncatedBlock.getBlockId() != storedBlock.getBlockId();
      if(recoveryId != newgenerationstamp) {
        throw new IOException("The recovery id " + newgenerationstamp
                              + " does not match current recovery id "
                              + recoveryId + " for block " + oldBlock);
      }

      // 一直到这, 进入正常的流程提交
      if (deleteblock) { // DN携带删除此block的请求, 则删除块
        Block blockToDel = ExtendedBlock.getLocalBlock(oldBlock);
        boolean remove = iFile.removeLastBlock(blockToDel) != null;
        if (remove) {
          blockManager.removeBlock(storedBlock);
        }
      } else { // 普通情况, 更新块信息, 使得NN记录的block信息与DN副本一致
        if(!copyTruncate) {
          storedBlock.setGenerationStamp(newgenerationstamp);
          storedBlock.setNumBytes(newlength);
        }

        // Find the target DatanodeStorageInfos. If not found because of invalid
        // or empty DatanodeID/StorageID, the slot of same offset in dsInfos is
        // null
        final DatanodeStorageInfo[] dsInfos = blockManager.getDatanodeManager().
                                              getDatanodeStorageInfos(newDNInfo...);

        if (closeFile && dsInfos != null) {
          // the file is getting closed. Insert block locations into blockManager.
          // Otherwise fsck will report these blocks as MISSING, especially if the
          // blocksReceived from Datanodes take a long time to arrive.
          for (int i = 0; i < dsInfos.length; i++) {
            if (dsInfos[i] != null) {
              if(copyTruncate) {
                dsInfos[i].addBlock(truncatedBlock, truncatedBlock);
              } else {
                Block bi = new Block(storedBlock);
                if (storedBlock.isStriped()) {
                  bi.setBlockId(bi.getBlockId() + i);
                }
                dsInfos[i].addBlock(storedBlock, bi);
              }
            }
          }
        }

        // add pipeline locations into the INodeUnderConstruction
        if(copyTruncate) {
          iFile.convertLastBlockToUC(truncatedBlock, dsInfos);
        } else {
          iFile.convertLastBlockToUC(storedBlock, dsInfos);
          if (closeFile) {
            blockManager.markBlockReplicasAsCorrupt(oldBlock.getLocalBlock(),
                storedBlock, oldGenerationStamp, oldNumBytes,
                dsInfos);
          }
        }
      }

      if (closeFile) { // 根据主DN携带参数判断是否立刻关闭文件
        if(copyTruncate) {
          closeFileCommitBlocks(src, iFile, truncatedBlock);
          if(!iFile.isBlockInLatestSnapshot(storedBlock)) {
            blockManager.removeBlock(storedBlock);
          }
        } else {
          closeFileCommitBlocks(src, iFile, storedBlock);
        }
      } else {
        // If this commit does not want to close the file, persist blocks
        FSDirWriteFileOp.persistBlocks(dir, src, iFile, false);
      }
      blockManager.successfulBlockRecovery(storedBlock);
    } finally {
      writeUnlock("commitBlockSynchronization");
    }
    getEditLog().logSync();
  }

0x03. 客户端の租约

由于租约这块需要整体理解, 所以并没有在第一篇Client篇单独讲客户端写入社会的续约租约相关操作, 一并放在此说.

对客户端而言, 其实要做的就是获得租约, 然后定期去续约, 如果超时没有续约, 那就可能触发租约过期, 搞清楚这两个概念, 再来看看具体实现

这是客户端LeaseRenewer这的核心代码. 主要就是续约操作, 比较简单.

// 首先类中提供了租约的CURD操作,租约是一个map映射

// 比如写新文件的时候
private void beginFileLease(final long inodeId, final DFSOutputStream out) {
    synchronized (filesBeingWritten) {
      putFileBeingWritten(inodeId, out);
      getLeaseRenewer().put(this); // 设置租约.开启后台检查进程
    }
  }

public synchronized void put(final DFSClient dfsc) {
    if (dfsc.isClientRunning()) {
      if (!isRunning() || isRenewerExpired()) {
        final int id = ++currentId;
        daemon = new Daemon(new Runnable() {
          @Override
          public void run() {
            try {
              LeaseRenewer.this.run(id); // 调用run()方法
            } catch(InterruptedException e) {
            } finally {
              synchronized(LeaseRenewer.this) {
                Factory.INSTANCE.remove(LeaseRenewer.this);
              }
            }
          } //...省略toString
        });
        daemon.start();
      }
      emptyTime = Long.MAX_VALUE;
    }
  }

// renew的核心代码
private void run(final int id) throws InterruptedException {
    for(long lastRenewed = Time.monotonicNow();
        !Thread.interrupted(); Thread.sleep(getSleepPeriod())) {
      final long elapsed = Time.monotonicNow() - lastRenewed;
      if (elapsed >= getRenewalTime()) {
        try {
          renew();
          lastRenewed = Time.monotonicNow();
        } catch (SocketTimeoutException ie) {
          List<DFSClient> dfsclientsCopy;
          synchronized (this) {
            DFSClientFaultInjector.get().delayWhenRenewLeaseTimeout();
            dfsclientsCopy = new ArrayList<>(dfsclients);
            dfsclients.clear();
            emptyTime = 0;
            Factory.INSTANCE.remove(LeaseRenewer.this);
          }
          for (DFSClient dfsClient : dfsclientsCopy) {
            dfsClient.closeAllFilesBeingWritten(true);
          }
          break;
        } catch (IOException ie) {....}
      }

        // clinet挂了
        if (!clientsRunning() && emptyTime == Long.MAX_VALUE) {
          emptyTime = Time.monotonicNow();
        }
      }
    }
  }

0x04. Namenode的错误恢复

因为相对租约恢复来说, 上一篇追加写和错误恢复中, 涉及到Namenode这部分的内容相对简单许多, 就放这里一起简单说一嘴, 后续有需要再补详细.

1. abandonBlock

同样实际代码调用在FSDirWriteFileOp类中, 逻辑很简单, 就不贴代码和流程图了, 只是将当前空的block从INode关联和BlockMap两个映射里删掉, 就没了.

2. updateBlockForPipeline

这个RPC是为当前block生成一个新的时间戳/Token, 逻辑也很简单, 如果blockId是随机生成就给随机, 如果是顺序增加则时间戳在原有基础上+1, 然后通过BlockManager生成一个新的Block对象, 返回给客户端

3. updatePipeline

这个RPC和上面的分配新的时间戳搭配使用, 它会检查分配的新时间戳是否合理(比旧的大). 然后在blockManager中替换掉旧的Block.

4. getAdditionalDatanode

核心是调用BlockManager的chooseTarget4AdditionalDatanode()方法, 然后底下也是调用addBlock时的选择DN的逻辑, 在这不重复说了.

5. append

相比而言, 追加写在Namenode这端做的操作要多一些, 并且因为它涉及到block状态的变更, 所以导致问题的可能性也大一些.

0x05. 疑问

租约这里的疑难点基本都在异常情况下, 也就是租约恢复仍然失败的情况, 整个流程可能无法形成闭环, 所以之后需要确认几个大点.

1. 租约恢复的死循环

我们上面讲述的租约恢复流程, 基本是说的普通情况, 但是举个最简单的例子:

在租约恢复大流程的时候, 如果失败抛出异常, 外层的方法catch住后直接跳过了当前文件继续对当前租约对应的其他文件进行租约恢复, 之后这个恢复失败被跳过的文件还会重新进入错误恢复的过程, 那问题就来了.

• 这个租约恢复失败的文件原本就是排序Set的第一个, 也就是每次都会先把它读出来, 它一直恢复失败重新恢复, 后面的文件不就都无法正常操作了?
• 从而引出另一个问题, 租约恢复的线程是只有1个么? 这个过程又加了锁, 那大集群这样一个个排队等也很慢吧.
• 如果是, 但是仍然报错, 比如副本数一直无法达到最低要求, 那不还是会触发死循环么? (早期HDFS版本就有这个问题, 且没有很好解决)

2. 租约被强制移除后的文件状态

不同于``HDFS3.1的做法,HDFS2.X`版本的做法里, 一个文件如果租约恢复异常, 则会被直接删除租约, 这样似乎至少不会导致上述的死循环发生

但是又会引出一个新的棘手问题, 一个处于中间状态的文件, 它的块可能处于UB状态, 此时直接duang一下删除它的租约后, 这个文件能被正常读写么?

如果能, 岂不是读写的时候DN状态也不一致…而且读多少呢? 还是说文件本身也会在读写异常时被删掉? 那似乎也不太合理… 而且既然高版本取消了这个做法, 说明应该是有所考虑的, 可以参考一下相关JIRA问题..

补充一下HDFS官方blog的引用, 就可以看出租约恢复和写异常中断就有一系列已被确认和未被修复的问题

下一篇是追加写/错误恢复, 以及租约部分对应的Datanode篇, 也是收尾篇了, 内容同样也很多.

---

参考资料:

1. HDFS3.1.2官方源码
2. Hadoop 2.X HDFS源码剖析-Chap.3.4
3. Cloudera-HDFS官方blog-part2
4. Namenode内存结构