Repository: apache/hadoop
File: hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BPServiceActor.java (Java)

Code before:
            }
            dn.getMetrics().addHeartbeat(scheduler.monotonicNow() - startTime,
                getRpcMetricSuffix());

            // If the state of this NN has changed (eg STANDBY->ACTIVE)
            // then let the BPOfferService update itself.
            //
            // Important that this happens before processCommand below,
            // since the first heartbeat to a new active might have commands
            // that we should actually process.
            bpos.updateActorStatesFromHeartbeat(
                this, resp.getNameNodeHaState());
            HAServiceState stateFromResp = resp.getNameNodeHaState().getState();
            if (state != stateFromResp) {
              LOG.info("After receiving heartbeat response, updating state of namenode {} to {}",
                  this.getNameNodeAddress(), stateFromResp);
            }
            state = stateFromResp;

            if (state == HAServiceState.ACTIVE) {
              handleRollingUpgradeStatus(resp);
            }
            commandProcessingThread.enqueue(resp.getCommands());
            isSlownode = resp.getIsSlownode();
          }
        }
        if (!dn.areIBRDisabledForTests() &&
            (ibrManager.sendImmediately()|| sendHeartbeat)) {
          ibrManager.sendIBRs(bpNamenode, bpRegistration,
              bpos.getBlockPoolId(), getRpcMetricSuffix());
        }

        List<DatanodeCommand> cmds = null;
        boolean forceFullBr =
            scheduler.forceFullBlockReport.getAndSet(false);
        if (forceFullBr) {
          LOG.info("Forcing a full block report to " + nnAddr);
        }
        if ((fullBlockReportLeaseId != 0) || forceFullBr) {
          cmds = blockReport(fullBlockReportLeaseId);
          fullBlockReportLeaseId = 0;
        }
        commandProcessingThread.enqueue(cmds);

        if (!dn.areCacheReportsDisabledForTests()) {
          DatanodeCommand cmd = cacheReport();
          commandProcessingThread.enqueue(cmd);
        }

        if (sendHeartbeat) {
          dn.getMetrics().addHeartbeatTotal(

Review comment (performance):
Will this loop will impact the performance of DataNode when there are huge number commands from NameNode? I prefer to let NameNode send back only Access key without other commands when necessary. Would you like to improve it at NameNode side too? BTW, It does not block this PR progress.

Code after:
            }
            dn.getMetrics().addHeartbeat(scheduler.monotonicNow() - startTime,
                getRpcMetricSuffix());

            // If the state of this NN has changed (eg STANDBY->ACTIVE)
            // then let the BPOfferService update itself.
            //
            // Important that this happens before processCommand below,
            // since the first heartbeat to a new active might have commands
            // that we should actually process.
            bpos.updateActorStatesFromHeartbeat(
                this, resp.getNameNodeHaState());
            HAServiceState stateFromResp = resp.getNameNodeHaState().getState();
            if (state != stateFromResp) {
              LOG.info("After receiving heartbeat response, updating state of namenode {} to {}",
                  this.getNameNodeAddress(), stateFromResp);
            }
            state = stateFromResp;

            if (state == HAServiceState.ACTIVE) {
              handleRollingUpgradeStatus(resp);
            }
            DatanodeCommand[] cmds = resp.getCommands();
            if (cmds != null && cmds.length != 0) {
              int length = cmds.length;
              for (int i = length - 1; i >= 0; i--) {
                if (cmds[i] instanceof KeyUpdateCommand) {
                  commandProcessingThread.enqueueFirst(cmds[i]);
                  cmds[i] = null;
                  break;
                }
              }
              commandProcessingThread.enqueue(cmds);
            }
            isSlownode = resp.getIsSlownode();
          }
        }
        if (!dn.areIBRDisabledForTests() &&
            (ibrManager.sendImmediately()|| sendHeartbeat)) {
          ibrManager.sendIBRs(bpNamenode, bpRegistration,
              bpos.getBlockPoolId(), getRpcMetricSuffix());
        }

        List<DatanodeCommand> cmds = null;
        boolean forceFullBr =
            scheduler.forceFullBlockReport.getAndSet(false);
        if (forceFullBr) {
          LOG.info("Forcing a full block report to " + nnAddr);
        }
        if ((fullBlockReportLeaseId != 0) || forceFullBr) {
          cmds = blockReport(fullBlockReportLeaseId);