外发加工网是不是骗局_搜狗提交网站收录入口_网站建设价格_搜索引擎查关键词排名的软件

文章目录

概述
问题描述
- 直接堆栈
- 排除不可能的原因
- 添加监控验证设想
- 为什么预留块越来越多
异常Case分析
- Case 1: 一个多副本异常块的写入过程
解决方案
- 快速的解决方案
- 快速解决以后问题依然存在
- Case 2: 单副本情况下的写入过程
- - 不同客户端写入同一文件的Block管理
  - 收到DataNode的块汇报以后，删除失效Block
  - scheduledBlocks是否会降低
- 从根本上解决问题
引用

概述

一个分布式系统发生问题，根据我的经验，这些问题可以根据性质不同被分成以下几类:

有明确异常堆栈且异常本身本身就是问题所在：比如，我们最熟悉的空指针异常。这种异常非常明确，并且我们只需要解决空指针，就从根本上解决了问题。
有明确异常堆栈但是堆栈本身只是系统问题的结果，原因在别处：比如，栈溢出异常或者由于内存不足导致对象分配失败的异常。这种异常有明确的错误信息，但是这个错误只是系统发生问题的最终结果，需要我们根据异常信息，进一步探究异常发生的原因。
性能问题：性能的好与坏本身如果没有横向对比，往往无法确认其是否是一个问题。比如，写操作的QPS在过去的几天增长了30%，然后HDFS的latency的p99在过去的几天增长了50%，这是否是一个问题？在很多情况下，如果存在一个横向对比(比如，存在另一个HDFS集群，硬件、机器性能与当前的HDFS不太一致)，我们往往很明确当前的性能是否需要优化，或者，即使不需要优化，当前系统的性能较差的大致原因有一个明确的结论(当前系统的QPS的确更大，当前系统的磁盘是HDD而另一个HDFS集群的磁盘是SSD等)
同上面提到的性能问题一样，还有一种问题类型，它没有明确异常堆栈，但是随着时间累积，统计性结果逐渐显现不正常状态：这种问题的重要特点，就是我们无法通过某一个具体的例子来复现问题，也往往无法从当前的不正常状态中直接找到一个可供诊断的例子。很多时候，我们只能硬着头皮看代码，或者对日志中的某些特征进行统计性分析，查找问题的发生的时间线与日志中某些特征的时间线的关联度。

本文遇到的问题就是第四类问题。
本文的目的，依然是通过我们 1). 发现问题、2). 快速诊断问题、3).根据诊断临时解决问题、4).事后根据更详细的代码分析查找根本原因和根本解决方案的几个基本流程，记录我们面对一个分布式系统的问题的工作方式，既是知识上的记录、总结和备忘，也是计数经验的分享。

我们的HDFS是很旧的版本2.6.0的CDH版本，本文描述的问题不一定在其他版本中存在。

问题描述

直接堆栈

我们的HDFS文件系统在长期运行过程中发现，用户的客户端偶尔出现所有DataNode不可用的状况:

org.apache.hadoop.ipc.RemoteException(java.io.IOException): File /tlb2/qe2-eco/aa-1min-output/tmp/2024/11/05/02/51/partition-60-part-0.parquet could only be replicated to 0 nodes instead of minReplication (=1). There are 6 datanode(s) running and no node(s) are excluded in this operation.
at org.apache.hadoop.hdfs.server.blockmanagement.BlockManager.chooseTarget4NewBlock(BlockManager.java:1720)
at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.getAdditionalBlock(FSNamesystem.java:3389)
at org.apache.hadoop.hdfs.server.namenode.NameNodeRpcServer.addBlock(NameNodeRpcServer.java:683)
at org.apache.hadoop.hdfs.server.namenode.AuthorizationProviderProxyClientProtocol.addBlock(AuthorizationProviderProxyClientProtocol.java:214)
at org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolServerSideTranslatorPB.addBlock(ClientNamenodeProtocolServerSideTranslatorPB.java:495)
at org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos$ClientNamenodeProtocol$2.callBlockingMethod(ClientNamenodeProtocolProtos.java)
at org.apache.hadoop.ipc.ProtobufRpcEngine$Server$ProtoBufRpcInvoker.call(ProtobufRpcEngine.java:617)
....
at org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolTranslatorPB.addBlock(ClientNamenodeProtocolTranslatorPB.java:500)
at java.base@22.0.2/java.lang.reflect.Method.invoke(Method.java:580)
...
at jdk.proxy4/jdk.proxy4.$Proxy48.addBlock(Unknown Source)
at org.apache.hadoop.hdfs.DFSOutputStream.addBlock(DFSOutputStream.java:1143)
at org.apache.hadoop.hdfs.DataStreamer.locateFollowingBlock(DataStreamer.java:2009)
at org.apache.hadoop.hdfs.DataStreamer.nextBlockOutputStream(DataStreamer.java:1811)
at org.apache.hadoop.hdfs.DataStreamer.run(DataStreamer.java:751)
....
|-backtrace (1)

可以看到，是HDFS在写文件的时候，所有的DataNode全部不可用。
虽然后堆栈，但是很明显，这个堆栈是一个最终结论的汇报，而这个最终结论本身的得出，依赖于HDFS的块放置策略。

排除不可能的原因

在我的另外一篇文章《HDFS的块汇报和块放置策略–从一次HDFS写文件故障开始》中详细讲解过HDFS的块放置策略，HDFS在为一个副本块选择节点的时候，显然只会选择存储空间足够的节点。但是，我们的HDFS集群是刚刚进行过扩容的HDFS集群，从NameNode暴露出来的metrics看到，整个集群的Storage使用率始终不到10%:
在这里插入图片描述

因此，我们当前遇到的问题与《HDFS的块汇报和块放置策略–从一次HDFS写文件故障开始》并不是同一个问题。我们不得不重新对问题进行分析。

在《HDFS的块汇报和块放置策略–从一次HDFS写文件故障开始》中我们讲解过DataNode无法被选择成为候选节点的几个基本原因:

这个节点是否是inService的状态：没有任何DataNode进入过decommission或者maintenance状态，因此该原因被排除。
这个节点是否是Stale节点：即这个节点是否还在向NameNode进行持续心跳。我们监控了DataNode向NameNode进行心跳的次数、增量块汇报的次数、所有DataNode的uptime(JVM持续存活的时间)，以及运行状态(GC Pause等)，所有DataNode没有发生过重启(更不可能同时被重启)：
每10分钟的Heartbeat的数量如下图所示:

每10分钟的块汇报数量如下图所示:

DataNode距离最近一次启动的时间：

DataNode实时的CMS GC的时间和次数:

请注意，我们从心跳的图上看到在事故发生期间的确心跳减少。后来我们看到，心跳的发生是IBR(Incremental Block Report)带来的，假如DataNode因为发生某种问题而没有新的块写入，那么就没有IBR，因此心跳也不会发生。这意味着，我们从上面的dashboard中看到事故发生时候的IBR和Heartbeat减小，是问题发生的结果，而不是原因。下面的代码证明，只有当有新的块写入进来的时候，才会触发IBR:
```
  /** Send IBRs to namenode. */void sendIBRs(DatanodeProtocol namenode, DatanodeRegistration registration,String bpid, DataNodeMetrics metrics) throws IOException {// Generate a list of the pending reports for each storage under the lockfinal StorageReceivedDeletedBlocks[] reports = generateIBRs();if (reports.length == 0) { // 如果没有新的块写入，则不进行IBR// Nothing new to report.return;}....boolean success = false;final long startTime = monotonicNow();try {namenode.blockReceivedAndDeleted(registration, bpid, reports);success = true;} finally {metrics.addIncrementalBlockReport(monotonicNow() - startTime);....}}
```
节点负载过大：在我的文章《HDFS的块汇报和块放置策略–从一次HDFS写文件故障开始》中也讲过节点负载过大导致DataNode写入失败的具体细节，这里的负载过大指的是，这个节点的负载超过所有DataNode的平均负载的两倍。显然，这并不是我们这边发生的情况，因为
- 首先，我们这边发生问题的时候，所有节点都出了问题，没有任何一个节点可用了。
- 并且，从下面的dashboard可以看出，没有哪一个节点的负载显著高于系统的平均负载
超过了单个机架的最大允许副本数：这里指的是，对于某一个Block，允许放置在单个Rack上的最大的副本数量。我们整个集群没有配置多Rack，根据我在文章《HDFS的块汇报和块放置策略–从一次HDFS写文件故障开始》中对问题的讲解，如果集群总共只有一个Rack，那么单个机架允许的最大副本数就是这个Block的总副本数，即，对于这个Block，单机架的总副本数不会成为问题。
这个节点的可用空间是否足够：在《HDFS的块汇报和块放置策略–从一次HDFS写文件故障开始》中，我们已经介绍过，可用空间的介绍需要考虑这个节点已经被调度(scheduledBlocks)但是还没有最终完成写入的Block。我们从监控上已经看到磁盘的实际可用空间是远远足够的，但是，是否，在减去了被调度块(scheduledBlocks)以后，是否就不够了呢？

添加监控验证设想

为了验证上面的猜想，我们添加了监控。NameNode没有直接暴露这个Metrics，我们通过用 磁盘可用空间 - 已经被调度的Block * Block Size ) 来间接计算剩余空间，监控结果如下:

在这里插入图片描述

可以看到，在去掉已经预留的块空间以后，所有DataNode根据公式 磁盘可用空间 - 已经被调度的Block * Block Size ) 计算得到的可用空间的确已经不足了，并且该值在不足的时候的时间与我们系统发生问题的时间吻合。

为什么预留块越来越多

我们确认了，HDFS的块预留策略会为正在进行写操作的块预留空间。因此，我们似乎可以得出这样一个粗糙的结论：尽管磁盘的物理存储空间还剩余很多，但是，由于客户端的写操作正在进行，HDFS为正在进行写操作的块进行空间预留，因此，HDFS的DataNode认为实际可用的存储空间就少于当前磁盘的真实物理可用存储空间。
这种解释在某种程度上似乎有一定的合理性：尽管集群的物理可用空间还有很多，但是，由于集群的节点数较少，因此，在某种并发度条件下，分摊到每一个节点的并发读写就会很大，这样，导致平均到每一个节点的预留空间就会很多。
比如:
当前HDFS集群的并发访问是1000 QPS，集群每台机器上的可用磁盘存储空间是1TB，这意味着，每台机器可接受的最大预留块数量为1TB/128MB = 8192。基于这种假设，集群有10台节点和集群有100台节点，虽然都是有看似足够的剩余存储空间，但是由于预留存储的存在，在DataNode的选择这个角度，实际上两个集群可以承受的并发不同。

集群有10台节点，可以承受的最大并发为8192 * 10 = 81920
集群有100台节点，可以承受的最大并发为8192 * 100 = 819200

但是，这种说法似乎与我们看到的现象不完全吻合:

按照这种说法，似乎客户端的写并发只要加大，那么预留的block就会变多，因此问题就会发生。写并发减小，预留的block就会变少，此时问题就会消失。但是，当问题发生的时候，HDFS用户反馈其对HDFS的使用无明显变化。
同时，我们观察当前预留的块数量，发现往往HDFS集群运行一段时间以后，问题开始逐渐发生并且越来越频繁。从下图可以看到，ScheduledBlock的数量逐渐增大，而不是维持在一个动态稳定状态。
我们根据块预留数量，平均到每台机器上的并发的块的写操作，显然大到了不合理的程度：从上图可以看到，每台机器上块预留最大达到了11K，即同时有11K个块在往这个DataNode上写，没有到达最后的Finalyzed状态。这显然是不可能的。

此时，问题的根本原因，从客户端的并发访问带来的块预留导致HDFS层面可用空间不足，变成，块预留不应该只增不降。

异常Case分析

Case 1: 一个多副本异常块的写入过程

我们通过在DataNode端的ReplicaAlreadyExistsException，找到了一个最早发现问题的典型的Block blk_1285224506，分析了它的写入路径。由于问题发生的时候我们没有备份NameNode的日志，因此没有在NameNode端这个块的DEBUG级别的状态转换信息。

在NameNode端，客户端发起了addBlock()请求，NameNode为该请求选择了三个节点:rccd101.dev.com, rccd102.dev.com,rccd103.dev.com。我在《HDFS的块汇报和块放置策略–从一次HDFS写文件故障开始》中介绍过，客户端在NameNode的响应中，拿到了这个DataNode的list，并且只会同第一个DataNode建立streaming，进行写操作。

我们从第一个DataNode rccd101.dev.com 日志中，看到了来自客户端的写操作请求:

2024-12-10 07:14:21,635 INFO .DataNode: Receiving BP-1772323693-10.30.122.121-1692931980521:blk_1285224506_211490316 src: /client-01.dev.com:17936 dest: /rccd101.dev.com:50010
2024-12-10 07:14:46,402 INFO .DataNode.clienttrace: src: /client-01.dev.com:17936, dest: /rccd101.dev.com:50010, bytes: 202610, op: HDFS_WRITE, cliID: DFSClient_NONMAPREDUCE_-1172730298_1, offset: 0, srvID: 71e1b3ab-2c2d-4abe-b6fc-3d1b5819e690, blockid: BP-1772323693-10.30.122.121-1692931980521:blk_1285224506_211490316, duration: 6498422530
2024-12-10 07:14:46,402 INFO .DataNode: PacketResponder: BP-1772323693-10.30.122.121-1692931980521:blk_1285224506_211490316, type=HAS_DOWNSTREAM_IN_PIPELINE terminating

从日志可以看到:

客户端的id是DFSClient_NONMAPREDUCE_-1172730298_1，这个Block id为blk_1285224506_211490316,大小为202610B;
写操作在2024-12-10 07:14:46,402结束
由于rccd101.dev.com是第一个DataNode，因此它会将对应的数据接力给下游，因此HAS_DOWNSTREAM_IN_PIPELINE。
参考代码，这里的terminating是指所有的写入操作完成了。

然后，我们在第二个节点rccd102.dev.com中看到了来自rccd101.dev.com的写入数据流：

2024-12-10 07:14:21,635 INFO .DataNode: Receiving BP-1772323693-10.30.122.121-1692931980521:blk_1285224506_211490316 src: /rccd101.dev.com:48954 dest: /rccd102.dev.com:50010
2024-12-10 07:14:46,401 INFO .DataNode.clienttrace: src: /rccd101.dev.com:48954, dest: /rccd102.dev.com:50010, bytes: 202610, op: HDFS_WRITE, cliID: DFSClient_NONMAPREDUCE_-1172730298_1, offset: 0, srvID: 93e08455-8172-45a1-b998-ab9468954f29, blockid: BP-1772323693-10.30.122.121-1692931980521:blk_1285224506_211490316, duration: 7182675412
2024-12-10 07:14:46,401 INFO .DataNode: PacketResponder: BP-1772323693-10.30.122.121-1692931980521:blk_1285224506_211490316, type=HAS_DOWNSTREAM_IN_PIPELINE terminating

其相关含义这里不做解释，与节点rccd101.dev.com的日志相同。这里我们可以看到，尽管写入到Pipeline中的第一个节点rccd101.dev.com的是用户所在的HDFS客户端，写入到第Pipeline的二个节点rccd102.dev.com是Pipeline的第一个节点rccd101.dev.com，但是对于被写入的节点来看，没有任何区别。

然后，我们在第三个节点rccd103.dev.com中看到了来自Pipeline中的第二个节点rccd102.dev.com的写入:

2024-12-10 07:14:39,218 INFO .DataNode: Receiving BP-1772323693-10.30.122.121-1692931980521:blk_1285224506_211490316 src: /rccd102.dev.com:51586 dest: /rccd103.dev.com:50010
:2024-12-10 07:14:46,401 INFO .DataNode.clienttrace: src: /rccd102.dev.com:51586, dest: /rccd103.dev.com:50010, bytes: 202610, op: HDFS_WRITE, cliID: DFSClient_NONMAPREDUCE_-1172730298_1, offset: 0, srvID: f5d24f10-9b8d-4ce3-bdd6-7d79a7102fa2, blockid: BP-1772323693-10.30.122.121-1692931980521:blk_1285224506_211490316, duration: 976576184
2024-12-10 07:14:46,401 INFO .DataNode: PacketResponder: BP-1772323693-10.30.122.121-1692931980521:blk_1285224506_211490316, type=LAST_IN_PIPELINE, downstreams=0:[] terminating

从日志type=LAST_IN_PIPELINE, downstreams=0:[] terminating可以看到,这是Pipeline中的最后一个节点，因此没有任何的下游节点。

随后，我们又在第一个DataNode rccd101.dev.com 节点，看到了下面的日志:

2024-12-10 07:14:50,216 INFO .DataNode: DatanodeRegistration(rccd101.dev.com, datanodeUuid=71e1b3ab-2c2d-4abe-b6fc-3d1b5819e690, infoPort=50075, infoSecurePort=0, ipcPort=50020, storageInfo=lv=-56;cid=cluster12;nsid=2036993546;c=0) Starting thread to transfer BP-1772323693-10.30.122.121-1692931980521:blk_1285224506_211490316 to rccd103.dev.com:50010
2024-12-10 07:14:50,252 WARN DataNode: DatanodeRegistration(rccd103.dev.com, datanodeUuid=71e1b3ab-2c2d-4abe-b6fc-3d1b5819e690, infoPort=50075, infoSecurePort=0, ipcPort=50020, storageInfo=lv=-56;cid=cluster12;nsid=2036993546;c=0):Failed to transfer BP-1772323693-10.30.122.121-1692931980521:blk_1285224506_211490316 to rccd103.dev.com:50010 got
java.net.SocketException: Original Exception : java.io.IOException: Connection reset by peerat sun.nio.ch.FileChannelImpl.transferTo0(Native Method)at sun.nio.ch.FileChannelImpl.transferToDirectlyInternal(FileChannelImpl.java:428)at sun.nio.ch.FileChannelImpl.transferToDirectly(FileChannelImpl.java:493)at sun.nio.ch.FileChannelImpl.transferTo(FileChannelImpl.java:608)at org.apache.hadoop.net.SocketOutputStream.transferToFully(SocketOutputStream.java:223)at org.apache.hadoop.hdfs.server.datanode.BlockSender.sendPacket(BlockSender.java:593)at org.apache.hadoop.hdfs.server.datanode.BlockSender.doSendBlock(BlockSender.java:777)at org.apache.hadoop.hdfs.server.datanode.BlockSender.sendBlock(BlockSender.java:724)at org.apache.hadoop.hdfs.server.datanode.DataNode$DataTransfer.run(DataNode.java:2331)at java.lang.Thread.run(Thread.java:748)
Caused by: java.io.IOException: Connection reset by peer... 10 more

我们从代码里面可以看到，这个特殊日志说明，source datanode rccd101.dev.com似乎在直接往target datanode rccd103.dev.com写入该Block并且写入失败。可问题是，刚刚不是已经看到，这个write pipeline中rccd101.dev.com的下游节点是rccd102.dev.com吗？

这个失败日志在rccd103.dev.com接收方也有对应的错误日志:

2024-12-10 07:14:50,245 INFO .DataNode: opWriteBlock BP-1772323693-10.30.122.121-1692931980521:blk_1285224506_211490316 received exception .ReplicaAlreadyExistsException: Block BP-1772323693-10.30.122.121-1692931980521:blk_1285224506_211490316 already exists in state FINALIZED and thus cannot be created.
2024-12-10 07:14:50,245 INFO .DataNode: rccd103.dev.com:50010:DataXceiver error processing WRITE_BLOCK operation  src: /rccd101.dev.com:44712 dst: /rccd103.dev.com:50010; .ReplicaAlreadyExistsException: Block BP-1772323693-10.30.122.121-1692931980521:blk_1285224506_211490316 already exists in state FINALIZED and thus cannot be created.

我们对比日志，看到Starting thread to transfer日志比较特殊，在正常写入的Pipeline中这个日志并不存在，因此查看代码:

@VisibleForTestingvoid transferBlock(ExtendedBlock block, DatanodeInfo[] xferTargets,StorageType[] xferTargetStorageTypes) throws IOException {BPOfferService bpos = getBPOSForBlock(block);DatanodeRegistration bpReg = getDNRegistrationForBP(block.getBlockPoolId());....if (numTargets > 0) {....LOG.info(bpReg + " Starting thread to transfer " + block + " to " + xfersBuilder);                       new Daemon(new DataTransfer(xferTargets, xferTargetStorageTypes, block,BlockConstructionStage.PIPELINE_SETUP_CREATE, "")).start();}}

可以看到，这个打印Starting thread to transfer ..日志的块传输请求直接来自于NameNode的调度，而不是一个正常的Write Pipeline中对客户端的写操作的接力传输。实际上，这个块传输来自于Active NameNode的DatanodeProtocol.DNA_TRANSFER命令:

  private boolean processCommandFromActive(DatanodeCommand cmd,BPServiceActor actor) throws IOException {final BlockCommand bcmd = cmd instanceof BlockCommand? (BlockCommand)cmd: null;final BlockIdCommand blockIdCmd = cmd instanceof BlockIdCommand ? (BlockIdCommand)cmd: null;switch(cmd.getAction()) {case DatanodeProtocol.DNA_TRANSFER:// Send a copy of a block to another datanodedn.transferBlocks(bcmd.getBlockPoolId(), bcmd.getBlocks(),bcmd.getTargets(), bcmd.getTargetStorageTypes());

我们看一下在NameNode端，什么时候会发出该命令呢？

在NameNode端，有一个ReplicationMonitor的Daemon Thread持续运行，不断监控所有的Block的副本数量，如果发现有Block的副本数不足，则尝试进行副本复制的调度，即，将向某个持有副本的DataNode发送命令，将副本复制到其他DataNode上。我们看到的DatanodeProtocol.DNA_TRANSFER命令，就是来自这个过程。我们随后在代码中可以看到，这个调度过程也会对目标DataNode(rccd103.dev.com)的scheduledBlocks加1，预期结果是，如果完成传输，会收到目标DataNode的Block Report，从而再将目标DataNode的scheduledBlocks减去1。
这个过程发生再以下关键方法中:

/*** Periodically calls computeReplicationWork().*/private class ReplicationMonitor implements Runnable {@Overridepublic void run() {while (namesystem.isRunning()) {// Process replication work only when active NN is out of safe mode.if (namesystem.isPopulatingReplQueues()) {// 如果有neededReplications，则挂到DataNode上面去，然后加入到pendingReplication中computeDatanodeWork();processPendingReplications(); // 从pendingReplication中取出已经timeout的block，检查副本数，如果副本依然不足，再次添加到neededReplication中rescanPostponedMisreplicatedBlocks();}Thread.sleep(replicationRecheckInterval);}

这里涉及到NameNode的以下数据结构：

一个全局唯一的UnderReplicatedBlocks neededReplication，用来负责存储副本数不足的Block信息。这些副本数不足可能来自于很多种情况:
- 在客户端写完文件，并调用complete方法结束写操作的时候，NameNode会检查这个文件的副本数量，如果副本数量不足，会将这个Block交给neededReplication，等待后续检查;
- 我们在Decommission一个节点的时候，显然，开始Decommission以后，这个Decommissioning DataNode上的所有Block都会处于副本不足状态，此时也都会添加给neededReplication来管理，等待后续处理
ReplicationMonitor的computeDatanodeWork()方法会不断从neededReplication中取出Block，通过相应的调度策略确定这些副本不足的Block的调度目标，构造对应的ReplicationWork，挂载到NameNode端的对应DataNode的DataNodeDescriptor上。这样，下一次，一旦DataNode的心跳过来，就可以取出这个调度任务响应给DataNode，即对应的DatanodeProtocol.DNA_TRANSFER命令，进行副本的复制操作。
- 从neededReplication中取出并生成了调度任务的块，会同时添加到一个全局对象PendingReplicationBlocks pendingReplications中。添加到pendingReplications的意思是：这些Block已经被派发了RepliationWork，我们过一段时间来检测这些Block的副本数是否已经恢复到正常状态，即ReplicationWork是否已经运行成功，如果成功，则可以从pendingReplications中删除，如果不成功，则貌似需要再次为其派发ReplicationWork，因此需要重新添加到UnderReplicatedBlocks neededReplication中。
- 需要注意，本文发生的预留块的统计计数，除了正常的用户写操作的时候会增加，在这里也会增加。而且，HDFS只有这两个地方会增加scheduledBlocks。我们可以查看下面代码：
```
 int computeReplicationWorkForBlocks(List<List<BlockInfo>> blocksToReplicate) {int requiredReplication, numEffectiveReplicas;.....// 把这个block添加到这个DataNode的toBeReplicated上面，一旦DataNode的心跳过来，这个DataNode// 就会收到对应的ReplicationWork，从而进行Replication的复制// Add block to the to be replicated listrw.srcNode.addBlockToBeReplicated(block, targets);scheduledWork++;// 增加scheduledBlock的计数器，由于Block是即将写到目标block，因此对应的scheduledBlocks计数器是添加到目标DataNode上的DatanodeStorageInfo.incrementBlocksScheduled(targets);// 把这个Block添加到pendingreplications中，这里的pending是因为此时NameNode还愿意等一等pendingReplications.increment(block,DatanodeStorageInfo.toDatanodeDescriptors(targets));blockLog.debug("BLOCK* block {} is moved from neededReplications to "+ "pendingReplications", block);.....}}} finally {namesystem.writeUnlock();}
```

ReplicationMonitor的processPendingReplications()方法会从pendingReplications中取出自从上次进行ReplicationWork调度以后已经超时的Block，检查这些Block的副本状态，如果依然不足，则这些Block会被重新放回neededReplication中，下一轮ReplicationMonitor会通过方法computeDatanodeWork()再次处理neededReplication中的块：

 private void processPendingReplications() {BlockInfo[] timedOutItems = pendingReplications.getTimedOutBlocks();for (int i = 0; i < timedOutItems.length; i++) {BlockInfo bi = blocksMap.getStoredBlock(timedOutItems[i]);NumberReplicas num = countNodes(timedOutItems[i]);if (isNeededReplication(bi, num)) {neededReplications.add(bi, num.liveReplicas(),num.readOnlyReplicas(), num.outOfServiceReplicas(),getReplication(bi));}}}

rescanPostponedMisreplicatedBlocks()方法比较特殊。一个Block的副本数超过预期数量，有两种情况:
- 这个Block的所有副本块都是正常状态，此时可以立刻调度一个InvalidateBlock到任意一个副本块，从而将这个Block的副本数量降低到正常状态;
- 但是，也有这种情况，这个Block含有Stale的副本块，这里Stale的含义是，如果一个NameNode(比如这个NameNode是刚刚从standby切换到active状态的NameNode)还从没有收到一个DataNode的heartbeat和block report，那么这个DataNode就被认为Stale状态，它上面所有的副本块都被认为是Stale。处于Stale状态的DataNode也许正在进行副本的删除操作, 我们知道，DataNode的删除操作是异步的，因此删除完成的时间不可预估,而且,如果是比如在DataNode需要下架一个Storage的时候，有大量的Block需要删除，而其删除操作仅仅是用一个4个线程的线程池完成的，非常缓慢，关于DataNode端的块删除，详情参考我的另一篇文章 HDFS从Trash中删除大量小文件导致的读写受阻问题。
- 因此，这时候，尽管NameNode已经知道这个Block的副本数量超过预期，但是，如果这个Block有的副本所在的DataNode还是stale状态，那么NameNode在收到对应Stale Node的Heartbeat和block report以前，不会也无法采取降低副本的操作，因为有些状态还不确定:
```
 private MisReplicationResult processMisReplicatedBlock(BlockInfo block) {BlockCollection bc = block.getBlockCollection();.....if (numCurrentReplica > expectedReplication) {if (num.replicasOnStaleNodes() > 0) {return MisReplicationResult.POSTPONE;}// over-replicated blockprocessOverReplicatedBlock(block, expectedReplication, null, null);return MisReplicationResult.OVER_REPLICATED;}return MisReplicationResult.OK;}
```
  下面的代码显式在 NameNode端将一个DataNode标记为Stale以及解除stale标记的过程：
```
public class DatanodeStorageInfo {....//在Failover刚刚发生的时候，设置为false，直到收到了DataNode的块汇报private boolean heartbeatedSinceFailover = false;/*** 在启动Failover时，有些DataNode可能正在进行上一个Active NameNode的一些删除操作。新的NameNode在接收到这个DataNode的块汇报以前，只能将这个DataNode标记位stale状态 详见 HDFS-1972。*/private boolean blockContentsStale = true;...boolean areBlockContentsStale() {return blockContentsStale;}void markStaleAfterFailover() {heartbeatedSinceFailover = false;blockContentsStale = true; // 刚刚failover的时候，标记这个DataNode为Stale状态}void receivedHeartbeat(StorageReport report) {updateState(report);heartbeatedSinceFailover = true; // 在Failover以后受到了heartbeat}void receivedBlockReport() {if (heartbeatedSinceFailover) { // 如果已经收到过heartbea，那么取消这个DataNode的stale状态blockContentsStale = false; }}
```
知道了以上背景，我们就可以理解方法rescanPostponedMisreplicatedBlocks()：该方法就是被ReplicationMonitor反复调度，反复扫描当前的postponed Misreplicated Blocks，对其状态进行重新鉴定，比如，如果这里的Block已经受到了对应的Stale DataNode的BlockReport，因此可以解除postpone状态，由于副本超过预期数量，因此可以进行降副本的操作了。

所以，可以看到，ReplicationMonitor会对副本数量不足的副本进行拷贝，以将副本增加到预期数量，这就是通过DatanodeProtocol.DNA_TRANSFER命令进行的。关于副本数量不足的原因，由于这种Case我们没有保存NameNode端的具体日志，因此无从得知这个具体过程，但是从我们后来的Case 2可以看到，用户端在写文件的时候存在一个文件的多次重复写操作，这种情况下，NameNode如果无法正确处理这种异常情况，可能就会发起对某一个Block进行错误的ReplicationWork调度,调度的目标节点其实在之前已经在正常的Pipeline中完成了Block的写操作，因此这个NameNode 触发的ReplicationWork在目标DataNode被抛出blk_1285224506_211490316 already exists in state FINALIZED and thus cannot be created异常。在这种异常发生的时候，NameNode端算在这个DataNode的ScheduledBlock数量再也无法降下来。

所以，基于上述分析，我们明白:

客户端的写入的方式的确存在问题，由于客户端存在对同一个文件的多次写入
导致DataNode端会抛出blk_1285224506_211490316 already exists in state FINALIZED异常，导致NameNode端的scheduledBlocks metrics在增加后无法降下来，并越来越多
NameNode的块调度策略在确认某一个DataNode的可用空间的时候，会用当前磁盘的剩余空间减去正在被调度的块总大小scheduledBlocks，因此，如果scheduledBlocks的统计有问题，那么就会造成DataNode被错误地认为不可用。。。

解决方案

快速的解决方案

当时，在我们只是确认时scheduledBlocks的存在导致NameNode认为DataNode的可用空间不足因此无法调度的时候，并没有意识到这是一个bug，只是认为scheduledBlocks过大只是写入并发过大而我们HDFS的DataNode的数量过小大致的，因此，我们当时的解决方案是:

修改客户端的配置文件，将文件副本数降低到1
修改客户端的配置文件，将副本大小从128MB降低到32MB
这样：
在公式NameNode认为的可用空间 = DataNode磁盘可用空间 - 已经被调度的Block * Block Size ) 中，NameNode认为的可用空间就会大大增加。
由于不存在副本复制问题，因此，在上文的Case 1中由于副本复制导致的块写入错误，以及由于该错误引起的scheduledBlocks无法降低的问题就不存在了。

快速解决以后问题依然存在

下图显式我们在采取了以上措施以后，基于磁盘可用空间 - 已经被调度的Block * Block Size ) 的计算方式得到的剩余空间的下降速度变慢很多，但是，它的大小依然在不断降低，而不是一个动态平衡状态。
在这里插入图片描述

所以，我们继续调查单副本情况下的异常。

Case 2: 单副本情况下的写入过程

在发现单副本情况下依然存在问题以后，我们依然针对DataNode端的某一个发生异常的Block进行了日志分析。并且，这一次，我们有了NameNode端的日志。

在NameNode端，客户端写入文件/prod/partition-6-part-0.parquet， NameNode为该文件分配Block blk_1299911045，并将该Block调度到DataNode rccd101-10c.dev.com, 并且，DataNode的确收到了客户端的write stream：

------ NameNode ---------
2024-12-12 16:54:21,969 INFO org.apache.hadoop.hdfs.StateChange: BLOCK* allocateBlock: /prod/partition-6-part-0.parquet. BP-1772323693-10.30.122.121-1692931980521 blk_1299911045_226176860{blockUCState=UNDER_CONSTRUCTION, primaryNodeIndex=-1, replicas=[ReplicaUnderConstruction[[DISK]DS-9b8c71a9-f051-4377-8ef2-a191e5b8399e:NORMAL:rccd101-10c.dev.com:50010|RBW]]}

------ DataNode rccd101-10c.dev.com ---------
2024-12-12 16:54:23,504 INFO org.apache.hadoop.hdfs.server.datanode.DataNode: Receiving BP-1772323693-10.30.122.121-1692931980521:blk_1299911045_226176860 src: /10.9.2.117:5219 dest: /rccd101-10c.dev.com:50010

随后，写入同一文件的另外一个客户端同样写入该文件/prod/partition-6-part-0.parquet，NameNode为该文件分配Block blk_1299911585，并将该Block调度到DataNode rccd102-7c.dev.com：

 ------ NameNode ---------2024-12-12 16:54:22,791 INFO org.apache.hadoop.hdfs.StateChange: BLOCK* allocateBlock: /prod/partition-6-part-0.parquet. BP-1772323693-10.30.122.121-1692931980521 blk_1299911585_226177400{blockUCState=UNDER_CONSTRUCTION, primaryNodeIndex=-1, replicas=[ReplicaUnderConstruction[[DISK]DS-1690e125-6910-43e9-84c4-a969f2104daf:NORMAL:rccd102-7c.dev.com:50010|RBW]]}

------ DataNode rccd102-7c.dev.com ---------
2024-12-12 16:54:22,979 INFO org.apache.hadoop.hdfs.server.datanode.DataNode: Receiving BP-1772323693-10.30.122.121-1692931980521:blk_1299911585_226177400 src: /client-138.dev.com:48692 dest: /rccd102-7c.dev.com:50010

NameNode将第一次写入该文件的Block添加到InvalidatedBlock中，随后，对应的DataNode rccd101-10c.dev.com 就收到了来自NameNode的删除命令。我们下文会讲到，NameNode的下面的日志是在收到来自DataNode的BlockReport以后发现该Block已经不存在，因此将该Block添加到DataNode rccd101-10c.dev.com 的InvalidatesBlock中去：

------- NameNode -------
2024-12-12 16:54:23,547 DEBUG BlockStateChange: BLOCK* addBlock: block blk_1299911045_226176860 on node rccd101-10c.dev.com:50010 size 33554432 does not belong to any file
2024-12-12 16:54:23,547 INFO BlockStateChange: BLOCK* InvalidateBlocks: add blk_1299911045_226176860 to rccd101-10c.dev.com:50010  //  添加到InvalidateBlocks，等待心跳调度过来，就把删除命令发送给rccd101-10c
2024-12-12 16:54:23,547 DEBUG BlockStateChange: BLOCK* block RECEIVING_BLOCK: blk_1299911045_226176860 is received from rccd101-10c.dev.com:50010
2024-12-12 16:54:23,811 INFO BlockStateChange: BLOCK* BlockManager: ask rccd101-10c.dev.com:50010 to delete [blk_1299910292_226176107, blk_1299911045_226176860, blk_1299911447_226177262, blk_1299911420_226177235]

------- DataNode rccd101-10c.dev.com -------
2024-12-12 16:54:24,563 INFO org.apache.hadoop.hdfs.server.datanode.fsdataset.impl.FsDatasetAsyncDiskService: Scheduling blk_1299911045_226176860 file /conviva/data/ssd1/dfs/dn/current/BP-1772323693-10.30.122.121-1692931980521/current/rbw/blk_1299911045 for deletion // 收到来自NameNode的删除请求
2024-12-12 16:54:24,564 INFO org.apache.hadoop.hdfs.server.datanode.fsdataset.impl.FsDatasetAsyncDiskService: Deleted BP-1772323693-10.30.122.121-1692931980521 blk_1299911045_226176860 file /conviva/data/ssd1/dfs/dn/current/BP-1772323693-10.30.122.121-1692931980521/current/rbw/blk_1299911045

在rccd101-10c.dev.com删除了Block blk_1299911045以后，居然收到了前客户端的finalizeBlock()的请求，显然，由于对应的Block已经删除，这时候会抛出异常:

------- DataNode rccd101-10c.dev.com -------
2024-12-12 16:54:25,102 WARN org.apache.hadoop.hdfs.server.datanode.DataNode: IOException in BlockReceiver.run():
org.apache.hadoop.hdfs.server.datanode.ReplicaNotFoundException: Cannot append to a non-existent replica BP-1772323693-10.30.122.121-1692931980521:blk_1299911045_226176860at org.apache.hadoop.hdfs.server.datanode.fsdataset.impl.FsDatasetImpl.getReplicaInfo(FsDatasetImpl.java:773)at org.apache.hadoop.hdfs.server.datanode.fsdataset.impl.FsDatasetImpl.finalizeBlock(FsDatasetImpl.java:1652)at org.apache.hadoop.hdfs.server.datanode.BlockReceiver$PacketResponder.finalizeBlock(BlockReceiver.java:1422)at org.apache.hadoop.hdfs.server.datanode.BlockReceiver$PacketResponder.run(BlockReceiver.java:1379)at java.lang.Thread.run(Thread.java:748)

而写入同一文件的第二个客户端，则在另外一个DataNoderccd102-7c.dev.com上写入成功:

2024-12-12 16:54:25,088 DEBUG BlockStateChange: BLOCK* block RECEIVED_BLOCK: blk_1299911585_226177400 is received from rccd102-7c.dev.com:50010

------ DataNode rccd102-7c.dev.com ---------
2024-12-12 16:54:25,089 INFO org.apache.hadoop.hdfs.server.datanode.DataNode.clienttrace: src: /client-119.dev.com:48692, dest: /rccd102-7c.dev.com:50010, bytes: 3391691, op: HDFS_WRITE, cliID: DFSClient_NONMAPREDUCE_1134367480_1, offset: 0, srvID: 7a83a27a-badc-4dbd-a3e1-f17e2d1dc799, blockid: BP-1772323693-10.30.122.121-1692931980521:blk_1299911585_226177400, duration: 2094405017
2024-12-12 16:54:25,089 INFO org.apache.hadoop.hdfs.server.datanode.DataNode: PacketResponder: BP-1772323693-10.30.122.121-1692931980521:blk_1299911585_226177400, type=LAST_IN_PIPELINE, downstreams=0:[] terminating

针对以上过程，我们需要确认的问题主要包括:

NameNode为客户端分配了Block blk_1299911045 ，然后第二个客户端申请写入同一个文件，NameNode没有任何报错，也为其分配了一个Block blk_1299911585，那么为第一个客户端分配的Block blk_1299911045会怎么处理？
我们从NameNode的日志可以看到，这个Block blk_1299911045 随后被调度删除，为什么会这样？
这个被调度然后被删除的Block blk_1299911045，在对应的DataNode端发生了异常Cannot append to a non-existent replica，这个异常是否会导致scheduledBlocks不降低？

不同客户端写入同一文件的Block管理

客户端在写入文件的时候，会先调用ClientNamenodeProtocolServerSideTranslatorPB.create()接口创建文件，只有创建文件成功，才会再调用ClientNamenodeProtocolServerSideTranslatorPB.addBlock()接口去申请Block。
客户端在创建DFSClient的时候，默认情况下是Overwrite模式，即，如果文件已经存在，那么会无条件将之前的文件覆盖掉:

  /*** Create an FSDataOutputStream at the indicated Path.* Files are overwritten by default.* @param f the file to create*/public FSDataOutputStream create(Path f) throws IOException {return create(f, true); // 第二个参数代表是否使用overwrite模式}public FSDataOutputStream create(Path f, boolean overwrite)throws IOException {return create(f, overwrite, getConf().getInt("io.file.buffer.size", 4096),getDefaultReplication(f),getDefaultBlockSize(f));}

在下层，如果使用了覆盖模式，那么，会在创建文件的RPC请求中添加CreateFlag.CREATE, CreateFlag.OVERWRITE两个CreateFlag:

  @Overridepublic FSDataOutputStream create(Path f, FsPermission permission,boolean overwrite, int bufferSize, short replication, long blockSize,Progressable progress) throws IOException {return this.create(f, permission,overwrite ? EnumSet.of(CreateFlag.CREATE, CreateFlag.OVERWRITE): EnumSet.of(CreateFlag.CREATE), bufferSize, replication,blockSize, progress, null);}

NameNode端收到创建请求并最终调用到BlocksMapUpdateInfo startFileInternal()中进行文件覆盖的调用堆栈如下所示:
在这里插入图片描述
从下面的方法可以看到，如果是overwrite模式，那么，第二个客户端如果尝试创建同一个文件，NameNode会将前一个客户端写入的文件和对应的Block删除，包括，从FSDirectory删除已经创建的文件，以及从BlocksMap中删除对应的Block：

private BlocksMapUpdateInfo startFileInternal(FSPermissionChecker pc, String src, .....String clientMachine, boolean create, boolean overwrite, boolean createParent  .....){ ....if (myFile == null) { // 如果这个文件不存在if (!create) { // 对于不存在的问题，必须要求是create 模式throw new FileNotFoundException("Can't overwrite non-existent " +src + " for client " + clientMachine);}} else { // 如果这个文件已经存在if (overwrite) { //  如果客户端是Override模式toRemoveBlocks = new BlocksMapUpdateInfo();List<INode> toRemoveINodes = new ChunkedArrayList<>();List<Long> toRemoveUCFiles = new ChunkedArrayList<>();// 从 FSDirectory中删除这个目录long ret = dir.delete(src, toRemoveBlocks, toRemoveINodes,toRemoveUCFiles, now());if (ret >= 0) { // 目录删除成功，那么开始删除对应的Block信息，包括inode中的block信息和blockmap中的block信息incrDeletedFileCount(ret); removePathAndBlocks(src, null, toRemoveUCFiles, toRemoveINodes,true);}} else {// If lease soft limit time is expired, recover the leaserecoverLeaseInternal(myFile, src, holder, clientMachine, false);throw new FileAlreadyExistsException(src + " for client " +clientMachine + " already exists");}}}

这时候，如果已经被删除的Block的DataNode发送过来这个DataNode的RECEIVING_BLOCK块汇报，NameNode试图在BlocksMap中查找该Block且找不到该Block，就会开始走Block的删除流程。下文讲解。

收到DataNode的块汇报以后，删除失效Block

在同一文件的第二个客户端开始写入，由于第一个客户端写入时分配的Block被覆盖，因此，这个已经被NameNode删除的Block锁在的DataNode发送过来RECEIVING_BLOCK的块汇报以后，NameNode找不到这个Block，并通过调用addToInvalidates()方法将该Block添加到InvalidateBlock中:

  private void processAndHandleReportedBlock(DatanodeStorageInfo storageInfo, Block block,ReplicaState reportedState, DatanodeDescriptor delHintNode)throws IOException {.....// find block by blockIdBlockInfo storedBlock = getStoredBlock(block);if(storedBlock == null) {// If blocksMap does not contain reported block id,// the replica should be removed from the data-node.blockLog.debug("BLOCK* addBlock: block {} on node {} size {} does not " +"belong to any file", block, node, block.getNumBytes());addToInvalidates(new Block(block), node);return;}

addToInvalidates()方法最终调用InvalidateBlocks类的add()方法，将对应的Block挂载到这个DataNode的invalidate blocks中，并在心跳的响应中将这批Block发送给DataNode进行删除:

  synchronized void add(final Block block, final DatanodeInfo datanode,final boolean log) {LightWeightHashSet<Block> set = node2blocks.get(datanode);if (set == null) {set = new LightWeightHashSet<Block>();node2blocks.put(datanode, set);}if (set.add(block)) {numBlocks++;if (log) {NameNode.blockStateChangeLog.info("BLOCK* {}: add {} to {}",getClass().getSimpleName(), block, datanode);}}}

对应上面的日志:

------- NameNode -------
2024-12-12 16:54:23,547 DEBUG BlockStateChange: BLOCK* addBlock: block blk_1299911045_226176860 on node rccd101-10c.dev.com:50010 size 33554432 does not belong to any file
2024-12-12 16:54:23,547 INFO BlockStateChange: BLOCK* InvalidateBlocks: add blk_1299911045_226176860 to rccd101-10c.dev.com:50010  //  添加到InvalidateBlocks，等待心跳调度过来，就把删除命令发送给rccd101-10c
2024-12-12 16:54:23,547 DEBUG BlockStateChange: BLOCK* block RECEIVING_BLOCK: blk_1299911045_226176860 is received from rccd101-10c.dev.com:50010
2024-12-12 16:54:23,811 INFO BlockStateChange: BLOCK* BlockManager: ask rccd101-10c.dev.com:50010 to delete [blk_1299910292_226176107, blk_1299911045_226176860, blk_1299911447_226177262, blk_1299911420_226177235]

DataNode在心跳响应中会收到删除的命令DatanodeProtocol.DNA_INVALIDATE，然后会对这个Block进行删除:

private boolean processCommandFromActive(DatanodeCommand cmd,BPServiceActor actor) throws IOException {switch(cmd.getAction()) {case DatanodeProtocol.DNA_INVALIDATE://// Some local block(s) are obsolete and can be // safely garbage-collected.//Block toDelete[] = bcmd.getBlocks();try {// using global fsdatasetdn.getFSDataset().invalidate(bcmd.getBlockPoolId(), toDelete);} catch(IOException e) {// Exceptions caught here are not expected to be disk-related.throw e;}

这对应了我们在DataNode端看到的日志:

    ------- DataNode rccd101-10c.dev.com -------2024-12-12 16:54:24,563 INFO org.apache.hadoop.hdfs.server.datanode.fsdataset.impl.FsDatasetAsyncDiskService: Scheduling blk_1299911045_226176860 file /conviva/data/ssd1/dfs/dn/current/BP-1772323693-10.30.122.121-1692931980521/current/rbw/blk_1299911045 for deletion // 收到来自NameNode的删除请求

scheduledBlocks是否会降低

上文两种Case刚好对应了NameNode会为某一个DataNode增加对应的scheduledBlock的情况:

Case 1中，是NameNode为一个副本不足的Block进行调度的时候发生的，这时候，会增加目标DataNode的scheduledBlock数量：

@VisibleForTestingint computeReplicationWorkForBlocks(List<List<BlockInfo>> blocksToReplicate) {...// 把这个block添加到这个DataNode的toBeReplicated上面，一旦DataNode的心跳过来，这个DataNode// 就会收到对应的ReplicationWork，从而进行Replication的复制// Add block to the to be replicated listrw.srcNode.addBlockToBeReplicated(block, targets);scheduledWork++;DatanodeStorageInfo.incrementBlocksScheduled(targets);// Move the block-replication into a "pending" state.// The reason we use 'pending' is so we can retry// replications that fail after an appropriate amount of time.// 把这个Block添加到pendingreplications中，这里的pending是因为此时NameNode还愿意等一等pendingReplications.increment(block,DatanodeStorageInfo.toDatanodeDescriptors(targets));blockLog.debug("BLOCK* block {} is moved from neededReplications to "+ "pendingReplications", block);// remove from neededReplicationsif(numEffectiveReplicas + targets.length >= requiredReplication) {neededReplications.remove(block, priority); // remove from neededReplications}}}} finally {namesystem.writeUnlock();}2. Case 2中的情况，在正常的Write Pipeline中，在BlockPlacementPolicy确定了一个块的放置目标位置以后，会增加目标DataNode的scheduledBlock数量：

在为一个文件添加Block的时候，都会调用FSNamesystem.saveAllocatedBlock()。从该方法中可以看到，这时候会将对应的目标DataNode的blocksScheduled值加1:

  BlockInfo saveAllocatedBlock(String src, INodesInPath inodes,Block newBlock, DatanodeStorageInfo[] targets)throws IOException {assert hasWriteLock();BlockInfo b = dir.addBlock(src, inodes, newBlock, targets);NameNode.stateChangeLog.info("BLOCK* allocateBlock: " + src + ". "+ getBlockPoolId() + " " + b);DatanodeStorageInfo.incrementBlocksScheduled(targets);return b;}

那么对应的，scheduledBlock什么时候会降低呢？只有一种情况，就是收到了来自DataNode的Incremental Block Report，并且，这个Block的状态是RECEIVED_BLOCK: