You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by we...@apache.org on 2019/12/11 01:16:59 UTC
[hadoop] branch trunk updated: HDFS-14854. Create improved
decommission monitor implementation. Contributed by Stephen O'Donnell.
This is an automated email from the ASF dual-hosted git repository.
weichiu pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/hadoop.git
The following commit(s) were added to refs/heads/trunk by this push:
new c93cb67 HDFS-14854. Create improved decommission monitor implementation. Contributed by Stephen O'Donnell.
c93cb67 is described below
commit c93cb6790e0f1c64efd03d859f907a0522010894
Author: Stephen O'Donnell <so...@cloudera.com>
AuthorDate: Tue Dec 10 17:15:30 2019 -0800
HDFS-14854. Create improved decommission monitor implementation. Contributed by Stephen O'Donnell.
Reviewed-by: Inigo Goiri <in...@apache.org>
Signed-off-by: Wei-Chiu Chuang <we...@apache.org>
---
.../java/org/apache/hadoop/hdfs/DFSConfigKeys.java | 15 +
.../DatanodeAdminBackoffMonitor.java | 818 +++++++++++++++++++++
.../DatanodeAdminDefaultMonitor.java | 446 +++++++++++
.../blockmanagement/DatanodeAdminManager.java | 449 +----------
.../blockmanagement/DatanodeAdminMonitorBase.java | 154 ++++
.../DatanodeAdminMonitorInterface.java | 39 +
.../src/main/resources/hdfs-default.xml | 35 +
.../hdfs/TestDecommissionWithBackoffMonitor.java | 51 ++
.../hadoop/hdfs/TestDecommissionWithStriped.java | 7 +-
.../TestDecommissionWithStripedBackoffMonitor.java | 40 +
.../server/namenode/TestDecommissioningStatus.java | 63 +-
...estDecommissioningStatusWithBackoffMonitor.java | 151 ++++
12 files changed, 1826 insertions(+), 442 deletions(-)
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
index cedf873..b5fd3b1 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
@@ -808,6 +808,21 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
public static final int DFS_NAMENODE_DECOMMISSION_BLOCKS_PER_INTERVAL_DEFAULT = 500000;
public static final String DFS_NAMENODE_DECOMMISSION_MAX_CONCURRENT_TRACKED_NODES = "dfs.namenode.decommission.max.concurrent.tracked.nodes";
public static final int DFS_NAMENODE_DECOMMISSION_MAX_CONCURRENT_TRACKED_NODES_DEFAULT = 100;
+ public static final String DFS_NAMENODE_DECOMMISSION_MONITOR_CLASS
+ = "dfs.namenode.decommission.monitor.class";
+ public static final String
+ DFS_NAMENODE_DECOMMISSION_MONITOR_CLASS_DEFAULT =
+ "org.apache.hadoop.hdfs.server.blockmanagement."+
+ "DatanodeAdminDefaultMonitor";
+ public static final String
+ DFS_NAMENODE_DECOMMISSION_BACKOFF_MONITOR_PENDING_LIMIT
+ = "dfs.namenode.decommission.backoff.monitor.pending.limit";
+ public static final int
+ DFS_NAMENODE_DECOMMISSION_BACKOFF_MONITOR_PENDING_LIMIT_DEFAULT = 10000;
+ public static final String DFS_NAMENODE_DECOMMISSION_BACKOFF_MONITOR_PENDING_BLOCKS_PER_LOCK =
+ "dfs.namenode.decommission.backoff.monitor.pending.blocks.per.lock";
+ public static final int DFS_NAMENODE_DECOMMISSION_BACKOFF_MONITOR_PENDING_BLOCKS_PER_LOCK_DEFAULT
+ = 1000;
public static final String DFS_NAMENODE_HANDLER_COUNT_KEY = "dfs.namenode.handler.count";
public static final int DFS_NAMENODE_HANDLER_COUNT_DEFAULT = 10;
public static final String DFS_NAMENODE_LIFELINE_HANDLER_RATIO_KEY =
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeAdminBackoffMonitor.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeAdminBackoffMonitor.java
new file mode 100644
index 0000000..af2c12f
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeAdminBackoffMonitor.java
@@ -0,0 +1,818 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.blockmanagement;
+
+import com.google.common.collect.Iterables;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
+import org.apache.hadoop.hdfs.server.namenode.INode;
+import org.apache.hadoop.hdfs.server.namenode.INodeFile;
+import org.apache.hadoop.hdfs.server.namenode.INodeId;
+import org.apache.hadoop.hdfs.util.LightWeightHashSet;
+import org.apache.hadoop.hdfs.util.LightWeightLinkedSet;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.HashMap;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.List;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.ArrayDeque;
+import java.util.Queue;
+
+/**
+ * This class implements the logic to track decommissioning and entering
+ * maintenance nodes, ensure all their blocks are adequately replicated
+ * before they are moved to the decommissioned or maintenance state.
+ *
+ * This monitor avoids flooding the replication queue with all pending blocks
+ * and instead feeds them to the queue as the prior set complete replication.
+ *
+ * HDFS-14854 contains details about the overall design of this class.
+ *
+ */
+public class DatanodeAdminBackoffMonitor extends DatanodeAdminMonitorBase
+ implements DatanodeAdminMonitorInterface {
+ /**
+ * Map containing the DECOMMISSION_INPROGRESS or ENTERING_MAINTENANCE
+ * datanodes that are being tracked so they can be be marked as
+ * DECOMMISSIONED or IN_MAINTENANCE. Even after the node is marked as
+ * IN_MAINTENANCE, the node remains in the map until
+ * maintenance expires checked during a monitor tick.
+ * <p/>
+ * This holds a set of references to the under-replicated blocks on the DN
+ * at the time the DN is added to the map, i.e. the blocks that are
+ * preventing the node from being marked as decommissioned. During a monitor
+ * tick, this list is pruned as blocks becomes replicated.
+ * <p/>
+ * Note also that the reference to the list of under-replicated blocks
+ * will be null on initial add
+ * <p/>
+ * However, this map can become out-of-date since it is not updated by block
+ * reports or other events. Before being finally marking as decommissioned,
+ * another check is done with the actual block map.
+ */
+ private HashMap<DatanodeDescriptor, HashMap<BlockInfo, Integer>>
+ outOfServiceNodeBlocks = new HashMap<>();
+
+ /**
+ * Any nodes where decommission or maintenance has been cancelled are added
+ * to this queue for later processing.
+ */
+ private final Queue<DatanodeDescriptor> cancelledNodes = new ArrayDeque<>();
+
+ /**
+ * The numbe of blocks to process when moving blocks to pendingReplication
+ * before releasing and reclaiming the namenode lock.
+ */
+ private int blocksPerLock;
+
+ /**
+ * The number of blocks that have been checked on this tick.
+ */
+ private int numBlocksChecked = 0;
+ /**
+ * The maximum number of blocks to hold in PendingRep at any time.
+ */
+ private int pendingRepLimit;
+
+ /**
+ * The list of blocks which have been placed onto the replication queue
+ * and are waiting to be sufficiently replicated.
+ */
+ private final Map<DatanodeDescriptor, List<BlockInfo>>
+ pendingRep = new HashMap<>();
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(DatanodeAdminBackoffMonitor.class);
+
+ DatanodeAdminBackoffMonitor() {
+ }
+
+
+ @Override
+ protected void processConf() {
+ this.pendingRepLimit = conf.getInt(
+ DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_BACKOFF_MONITOR_PENDING_LIMIT,
+ DFSConfigKeys.
+ DFS_NAMENODE_DECOMMISSION_BACKOFF_MONITOR_PENDING_LIMIT_DEFAULT);
+ if (this.pendingRepLimit < 1) {
+ LOG.error("{} is set to an invalid value, it must be greater than "+
+ "zero. Defaulting to {}",
+ DFSConfigKeys.
+ DFS_NAMENODE_DECOMMISSION_BACKOFF_MONITOR_PENDING_LIMIT,
+ DFSConfigKeys.
+ DFS_NAMENODE_DECOMMISSION_BACKOFF_MONITOR_PENDING_LIMIT_DEFAULT
+ );
+ this.pendingRepLimit = DFSConfigKeys.
+ DFS_NAMENODE_DECOMMISSION_BACKOFF_MONITOR_PENDING_LIMIT_DEFAULT;
+ }
+ this.blocksPerLock = conf.getInt(
+ DFSConfigKeys.
+ DFS_NAMENODE_DECOMMISSION_BACKOFF_MONITOR_PENDING_BLOCKS_PER_LOCK,
+ DFSConfigKeys.
+ DFS_NAMENODE_DECOMMISSION_BACKOFF_MONITOR_PENDING_BLOCKS_PER_LOCK_DEFAULT
+ );
+ if (blocksPerLock <= 0) {
+ LOG.error("{} is set to an invalid value, it must be greater than "+
+ "zero. Defaulting to {}",
+ DFSConfigKeys.
+ DFS_NAMENODE_DECOMMISSION_BACKOFF_MONITOR_PENDING_BLOCKS_PER_LOCK,
+ DFSConfigKeys.
+ DFS_NAMENODE_DECOMMISSION_BACKOFF_MONITOR_PENDING_BLOCKS_PER_LOCK_DEFAULT);
+ blocksPerLock =
+ DFSConfigKeys.
+ DFS_NAMENODE_DECOMMISSION_BACKOFF_MONITOR_PENDING_BLOCKS_PER_LOCK_DEFAULT;
+ }
+ LOG.info("Initialized the Backoff Decommission and Maintenance Monitor");
+ }
+
+ /**
+ * Queue a node to be removed from tracking. This method must be called
+ * under the namenode write lock.
+ * @param dn The datanode to stop tracking for decommission.
+ */
+ @Override
+ public void stopTrackingNode(DatanodeDescriptor dn) {
+ pendingNodes.remove(dn);
+ cancelledNodes.add(dn);
+ }
+
+ @Override
+ public int getTrackedNodeCount() {
+ return outOfServiceNodeBlocks.size();
+ }
+
+ @Override
+ public int getNumNodesChecked() {
+ // We always check all nodes on each tick
+ return outOfServiceNodeBlocks.size();
+ }
+
+ @Override
+ public void run() {
+ LOG.debug("DatanodeAdminMonitorV2 is running.");
+ if (!namesystem.isRunning()) {
+ LOG.info("Namesystem is not running, skipping " +
+ "decommissioning/maintenance checks.");
+ return;
+ }
+ // Reset the checked count at beginning of each iteration
+ numBlocksChecked = 0;
+ // Check decommission or maintenance progress.
+ try {
+ namesystem.writeLock();
+ try {
+ /**
+ * Other threads can modify the pendingNode list and the cancelled
+ * node list, so we must process them under the NN write lock to
+ * prevent any concurrent modifications.
+ *
+ * Always process the cancelled list before the pending list, as
+ * it is possible for a node to be cancelled, and then quickly added
+ * back again. If we process these the other way around, the added
+ * node will be removed from tracking by the pending cancel.
+ */
+ processCancelledNodes();
+ processPendingNodes();
+ } finally {
+ namesystem.writeUnlock();
+ }
+ // After processing the above, various parts of the check() method will
+ // take and drop the read / write lock as needed. Aside from the
+ // cancelled and pending lists, nothing outside of the monitor thread
+ // modifies anything inside this class, so many things can be done
+ // without any lock.
+ check();
+ } catch (Exception e) {
+ LOG.warn("DatanodeAdminMonitor caught exception when processing node.",
+ e);
+ }
+ if (numBlocksChecked + outOfServiceNodeBlocks.size() > 0) {
+ LOG.info("Checked {} blocks this tick. {} nodes are now " +
+ "in maintenance or transitioning state. {} nodes pending. {} " +
+ "nodes waiting to be cancelled.",
+ numBlocksChecked, outOfServiceNodeBlocks.size(), pendingNodes.size(),
+ cancelledNodes.size());
+ }
+ }
+
+ /**
+ * Move any pending nodes into outOfServiceNodeBlocks to initiate the
+ * decommission or maintenance mode process.
+ *
+ * This method must be executed under the namenode write lock to prevent
+ * the pendingNodes list from being modified externally.
+ */
+ private void processPendingNodes() {
+ while (!pendingNodes.isEmpty() &&
+ (maxConcurrentTrackedNodes == 0 ||
+ outOfServiceNodeBlocks.size() < maxConcurrentTrackedNodes)) {
+ outOfServiceNodeBlocks.put(pendingNodes.poll(), null);
+ }
+ }
+
+ /**
+ * Process any nodes which have had their decommission or maintenance mode
+ * cancelled by an administrator.
+ *
+ * This method must be executed under the
+ * write lock to prevent the cancelledNodes list being modified externally.
+ */
+ private void processCancelledNodes() {
+ while(!cancelledNodes.isEmpty()) {
+ DatanodeDescriptor dn = cancelledNodes.poll();
+ outOfServiceNodeBlocks.remove(dn);
+ pendingRep.remove(dn);
+ }
+ }
+
+ /**
+ * This method performs each of the steps to track a node from
+ * decommissioning or entering maintenance to the end state.
+ *
+ * First, any newly added nodes are scanned.
+ *
+ * Then any expired maintenance nodes are handled.
+ *
+ * Next the pendingRep map is scanned and all blocks which are now
+ * sufficiently replicated are removed
+ *
+ * Then new blocks are moved to pendingRep
+ *
+ * Finally we check if any nodes have completed the replication process and
+ * if so move them to their final states.
+ *
+ * This methods which this method calls will take and release the namenode
+ * read and write lock several times.
+ *
+ */
+ private void check() {
+ final List<DatanodeDescriptor> toRemove = new ArrayList<>();
+
+ if (outOfServiceNodeBlocks.size() == 0) {
+ // No nodes currently being tracked so simply return
+ return;
+ }
+
+ // Check if there are any pending nodes to process, ie those where the
+ // storage has not been scanned yet. For all which are pending, scan
+ // the storage and load the under-replicated block list into
+ // outOfServiceNodeBlocks. As this does not modify any external structures
+ // it can be done under the namenode *read* lock, and the lock can be
+ // dropped between each storage on each node.
+ //
+ // TODO - This is an expensive call, depending on how many nodes are
+ // to be processed, but it requires only the read lock and it will
+ // be dropped and re-taken frequently. We may want to throttle this
+ // to process only a few nodes per iteration.
+ outOfServiceNodeBlocks.keySet()
+ .stream()
+ .filter(n -> outOfServiceNodeBlocks.get(n) == null)
+ .forEach(n -> scanDatanodeStorage(n, true));
+
+ processMaintenanceNodes();
+ // First check the pending replication list and remove any blocks
+ // which are now replicated OK. This list is constrained in size so this
+ // call should not be overly expensive.
+ processPendingReplication();
+
+ // Now move a limited number of blocks to pending
+ moveBlocksToPending();
+
+ // Check if any nodes have reached zero blocks and also update the stats
+ // exposed via JMX for all nodes still being processed.
+ checkForCompletedNodes(toRemove);
+
+ // Finally move the nodes to their final state if they are ready.
+ processCompletedNodes(toRemove);
+ }
+
+ /**
+ * Checks for any nodes which are in maintenance and if maintenance has
+ * expired, the node will be moved back to in_service (or dead) as required.
+ */
+ private void processMaintenanceNodes() {
+ // Check for any maintenance state nodes which need to be expired
+ namesystem.writeLock();
+ try {
+ for (DatanodeDescriptor dn : outOfServiceNodeBlocks.keySet()) {
+ if (dn.isMaintenance() && dn.maintenanceExpired()) {
+ // If maintenance expires, stop tracking it. This can be an
+ // expensive call, as it may need to invalidate blocks. Therefore
+ // we can yield and retake the write lock after each node
+ //
+ // The call to stopMaintenance makes a call to stopTrackingNode()
+ // which added the node to the cancelled list. Therefore expired
+ // maintenance nodes do not need to be added to the toRemove list.
+ dnAdmin.stopMaintenance(dn);
+ namesystem.writeUnlock();
+ namesystem.writeLock();
+ }
+ }
+ } finally {
+ namesystem.writeUnlock();
+ }
+ }
+
+ /**
+ * Loop over all nodes in the passed toRemove list and move the node to
+ * the required end state. This will also remove any entries from
+ * outOfServiceNodeBlocks and pendingRep for the node if required.
+ *
+ * @param toRemove The list of nodes to process for completion.
+ */
+ private void processCompletedNodes(List<DatanodeDescriptor> toRemove) {
+ if (toRemove.size() == 0) {
+ // If there are no nodes to process simply return and avoid
+ // taking the write lock at all.
+ return;
+ }
+ namesystem.writeLock();
+ try {
+ for (DatanodeDescriptor dn : toRemove) {
+ final boolean isHealthy =
+ blockManager.isNodeHealthyForDecommissionOrMaintenance(dn);
+ if (isHealthy) {
+ if (dn.isDecommissionInProgress()) {
+ dnAdmin.setDecommissioned(dn);
+ outOfServiceNodeBlocks.remove(dn);
+ pendingRep.remove(dn);
+ } else if (dn.isEnteringMaintenance()) {
+ // IN_MAINTENANCE node remains in the outOfServiceNodeBlocks to
+ // to track maintenance expiration.
+ dnAdmin.setInMaintenance(dn);
+ pendingRep.remove(dn);
+ } else if (dn.isInService()) {
+ // Decom / maint was cancelled and the node is yet to be processed
+ // from cancelledNodes
+ LOG.info("Node {} completed decommission and maintenance " +
+ "but has been moved back to in service", dn);
+ pendingRep.remove(dn);
+ outOfServiceNodeBlocks.remove(dn);
+ continue;
+ } else {
+ // Should not happen
+ LOG.error("Node {} is in an unexpected state {} and has been "+
+ "removed from tracking for decommission or maintenance",
+ dn, dn.getAdminState());
+ pendingRep.remove(dn);
+ outOfServiceNodeBlocks.remove(dn);
+ continue;
+ }
+ LOG.info("Node {} is sufficiently replicated and healthy, "
+ + "marked as {}.", dn, dn.getAdminState());
+ } else {
+ LOG.info("Node {} isn't healthy."
+ + " It needs to replicate {} more blocks."
+ + " {} is still in progress.", dn,
+ getPendingCountForNode(dn), dn.getAdminState());
+ }
+ }
+ } finally {
+ namesystem.writeUnlock();
+ }
+ }
+
+ /**
+ * Loop over all nodes and check for any which have zero unprocessed or
+ * pending blocks. If the node has zero blocks pending, the storage is
+ * rescanned to ensure no transient blocks were missed on the first pass.
+ *
+ * If, after rescan the number of blocks pending replication is zero, the
+ * node is added to the passed removeList which will later be processed to
+ * complete the decommission or entering maintenance process.
+ *
+ * @param removeList Nodes which have zero pending blocks are added to this
+ * list.
+ */
+ private void checkForCompletedNodes(List<DatanodeDescriptor> removeList) {
+ for (DatanodeDescriptor dn : outOfServiceNodeBlocks.keySet()) {
+ // If the node is already in maintenance, we don't need to perform
+ // any further checks on it.
+ if (dn.isInMaintenance()) {
+ LOG.debug("Node {} is currently in maintenance", dn);
+ continue;
+ } else if (!dn.isInService()) {
+ // A node could be inService if decom or maint has been cancelled, but
+ // the cancelled list is yet to be processed. We don't need to check
+ // inService nodes here
+ int outstandingBlocks = getPendingCountForNode(dn);
+ if (outstandingBlocks == 0) {
+ scanDatanodeStorage(dn, false);
+ outstandingBlocks = getPendingCountForNode(dn);
+ }
+ LOG.info("Node {} has {} blocks yet to process", dn, outstandingBlocks);
+ if (outstandingBlocks == 0) {
+ removeList.add(dn);
+ }
+ }
+ }
+ }
+
+ /**
+ * Returns the number of block pending for the given node by adding those
+ * blocks in pendingRep and outOfServiceNodeBlocks.
+ *
+ * @param dn The datanode to return the count for
+ * @return The total block count, or zero if none are pending
+ */
+ private int getPendingCountForNode(DatanodeDescriptor dn) {
+ int count = 0;
+ HashMap<BlockInfo, Integer> blocks = outOfServiceNodeBlocks.get(dn);
+ if (blocks != null) {
+ count += blocks.size();
+ }
+ List<BlockInfo> pendingBlocks = pendingRep.get(dn);
+ if (pendingBlocks != null) {
+ count += pendingBlocks.size();
+ }
+ return count;
+ }
+
+ /**
+ * Iterate across all nodes in outOfServiceNodeBlocks which have blocks yet
+ * to be processed.
+ *
+ * The block is removed from outOfServiceNodeBlocks and if it needs
+ * replication it is added to the pendingRep map and also to the
+ * BlockManager replication queue.
+ *
+ * Any block that does not need replication is discarded.
+ *
+ * The method will return when there are the pendingRep map has
+ * pendingRepLimit blocks or there are no further blocks to process.
+ */
+ private void moveBlocksToPending() {
+ int blocksProcessed = 0;
+ int pendingCount = getPendingCount();
+ int yetToBeProcessed = getYetToBeProcessedCount();
+
+ if (pendingCount == 0 && yetToBeProcessed == 0) {
+ // There are no blocks to process so just return
+ LOG.debug("There are no pending or blocks yet to be processed");
+ return;
+ }
+
+ namesystem.writeLock();
+ try {
+ long repQueueSize = blockManager.getLowRedundancyBlocksCount();
+
+ LOG.info("There are {} blocks pending replication and the limit is "+
+ "{}. A further {} blocks are waiting to be processed. "+
+ "The replication queue currently has {} blocks",
+ pendingCount, pendingRepLimit, yetToBeProcessed, repQueueSize);
+
+ if (pendingCount >= pendingRepLimit) {
+ // Only add more blocks to the replication queue if we don't already
+ // have too many pending
+ return;
+ }
+
+ // Create a "Block Iterator" for each node decommissioning or entering
+ // maintenance. These iterators will be used "round robined" to add blocks
+ // to the replication queue and PendingRep
+ HashMap<DatanodeDescriptor, Iterator<BlockInfo>>
+ iterators = new HashMap<>();
+ for (Map.Entry<DatanodeDescriptor, HashMap<BlockInfo, Integer>> e
+ : outOfServiceNodeBlocks.entrySet()) {
+ iterators.put(e.getKey(), e.getValue().keySet().iterator());
+ }
+
+ // Now loop until we fill the pendingRep map with pendingRepLimit blocks
+ // or run out of blocks to add.
+ Iterator<DatanodeDescriptor> nodeIter =
+ Iterables.cycle(iterators.keySet()).iterator();
+ while (nodeIter.hasNext()) {
+ // Cycle through each node with blocks which still need processed
+ DatanodeDescriptor dn = nodeIter.next();
+ Iterator<BlockInfo> blockIt = iterators.get(dn);
+ while (blockIt.hasNext()) {
+ // Process the blocks for the node until we find one that needs
+ // replication
+ if (blocksProcessed >= blocksPerLock) {
+ blocksProcessed = 0;
+ namesystem.writeUnlock();
+ namesystem.writeLock();
+ }
+ blocksProcessed++;
+ if (nextBlockAddedToPending(blockIt, dn)) {
+ // Exit the inner "block" loop so an iterator for the next datanode
+ // is used for the next block.
+ pendingCount++;
+ break;
+ }
+ }
+ if (!blockIt.hasNext()) {
+ // remove the iterator as there are no blocks left in it
+ nodeIter.remove();
+ }
+ if (pendingCount >= pendingRepLimit) {
+ // We have scheduled the limit of blocks for replication, so do
+ // not add any more
+ break;
+ }
+ }
+ } finally {
+ namesystem.writeUnlock();
+ }
+ LOG.debug("{} blocks are now pending replication", pendingCount);
+ }
+
+ /**
+ * Takes and removes the next block from the given iterator and checks if it
+ * needs additional replicas. If it does, it will be scheduled for
+ * reconstruction and added to the pendingRep map.
+ * @param it The iterator to take the next block from
+ * @param dn The datanodeDescriptor the iterator applies to
+ * @return True if the block needs replication, otherwise false
+ */
+ private boolean nextBlockAddedToPending(Iterator<BlockInfo> it,
+ DatanodeDescriptor dn) {
+ BlockInfo block = it.next();
+ it.remove();
+ numBlocksChecked++;
+ if (!isBlockReplicatedOk(dn, block, true, null)) {
+ pendingRep.computeIfAbsent(dn, k -> new LinkedList<>()).add(block);
+ return true;
+ }
+ return false;
+ }
+
+ private int getPendingCount() {
+ if (pendingRep.size() == 0) {
+ return 0;
+ }
+ return pendingRep.values()
+ .stream()
+ .map(a -> a.size())
+ .reduce(0, (a, b) -> a + b);
+ }
+
+ private int getYetToBeProcessedCount() {
+ if (outOfServiceNodeBlocks.size() == 0) {
+ return 0;
+ }
+ return outOfServiceNodeBlocks.values()
+ .stream()
+ .map(a -> a.size())
+ .reduce(0, (a, b) -> a + b);
+ }
+
+ /**
+ * Scan all the blocks held on a datanodes. For a node being decommissioned
+ * we assume that the majority of blocks on the node will need to have new
+ * replicas made, and therefore we do not check if they are under replicated
+ * here and instead add them to the list of blocks to track.
+ *
+ * For a node being moved into maintenance, we assume most blocks will be
+ * replicated OK and hence we do check their under-replicated status here,
+ * hopefully reducing the number of blocks to track.
+ *
+ * On a re-scan (initalScan = false) we assume the node has been processed
+ * already, and hence there should be few under-replicated blocks, so we
+ * check the under-replicated status before adding the blocks to the
+ * tracking list.
+ *
+ * This means that for a node being decomission there should be a large
+ * number of blocks to process later but for maintenance, a smaller number.
+ *
+ * As this method does not schedule any blocks for reconstuction, this
+ * scan can be performed under the namenode readlock, and the lock is
+ * dropped and reaquired for each storage on the DN.
+ *
+ * @param dn - The datanode to process
+ * @param initialScan - True is this is the first time scanning the node
+ * or false if it is a rescan.
+ */
+ private void scanDatanodeStorage(DatanodeDescriptor dn,
+ Boolean initialScan) {
+ HashMap<BlockInfo, Integer> blockList = outOfServiceNodeBlocks.get(dn);
+ if (blockList == null) {
+ blockList = new HashMap<>();
+ outOfServiceNodeBlocks.put(dn, blockList);
+ }
+
+ DatanodeStorageInfo[] storage;
+ namesystem.readLock();
+ try {
+ storage = dn.getStorageInfos();
+ } finally {
+ namesystem.readUnlock();
+ }
+
+ for (DatanodeStorageInfo s : storage) {
+ namesystem.readLock();
+ try {
+ // As the lock is dropped and re-taken between each storage, we need
+ // to check the storage is still present before processing it, as it
+ // may have been removed.
+ if (dn.getStorageInfo(s.getStorageID()) == null) {
+ continue;
+ }
+ Iterator<BlockInfo> it = s.getBlockIterator();
+ while (it.hasNext()) {
+ BlockInfo b = it.next();
+ if (!initialScan || dn.isEnteringMaintenance()) {
+ // this is a rescan, so most blocks should be replicated now,
+ // or this node is going into maintenance. On a healthy
+ // cluster using racks or upgrade domain, a node should be
+ // able to go into maintenance without replicating many blocks
+ // so we will check them immediately.
+ if (!isBlockReplicatedOk(dn, b, false, null)) {
+ blockList.put(b, null);
+ }
+ } else {
+ blockList.put(b, null);
+ }
+ numBlocksChecked++;
+ }
+ } finally {
+ namesystem.readUnlock();
+ }
+ }
+ }
+
+ /**
+ * Process the list of pendingReplication Blocks. These are the blocks
+ * which have been moved from outOfServiceNodeBlocks, confirmed to be
+ * under-replicated and were added to the blockManager replication
+ * queue.
+ *
+ * Any blocks which have been confirmed to be replicated sufficiently are
+ * removed from the list.
+ *
+ * The datanode stats are also updated in this method, updating the total
+ * pending block count, the number of blocks in PendingRep which are in
+ * open files and the number of blocks in PendingRep which are only on
+ * out of service nodes.
+ *
+ * As this method makes changes to the replication queue, it acquires the
+ * namenode write lock while it runs.
+ */
+ private void processPendingReplication() {
+ namesystem.writeLock();
+ try {
+ for (Iterator<Map.Entry<DatanodeDescriptor, List<BlockInfo>>>
+ entIt = pendingRep.entrySet().iterator(); entIt.hasNext();) {
+ Map.Entry<DatanodeDescriptor, List<BlockInfo>> entry = entIt.next();
+ DatanodeDescriptor dn = entry.getKey();
+ List<BlockInfo> blocks = entry.getValue();
+ if (blocks == null) {
+ // should not be able to happen
+ entIt.remove();
+ continue;
+ }
+ Iterator<BlockInfo> blockIt = blocks.iterator();
+ BlockStats suspectBlocks = new BlockStats();
+ while(blockIt.hasNext()) {
+ BlockInfo b = blockIt.next();
+ if (isBlockReplicatedOk(dn, b, true, suspectBlocks)) {
+ blockIt.remove();
+ }
+ numBlocksChecked++;
+ }
+ if (blocks.size() == 0) {
+ entIt.remove();
+ }
+ // Update metrics for this datanode.
+ dn.getLeavingServiceStatus().set(
+ suspectBlocks.getOpenFileCount(),
+ suspectBlocks.getOpenFiles(),
+ getPendingCountForNode(dn),
+ suspectBlocks.getOutOfServiceBlockCount());
+ }
+ } finally {
+ namesystem.writeUnlock();
+ }
+ }
+
+ /**
+ * Checks if a block is sufficiently replicated and optionally schedules
+ * it for reconstruction if it is not.
+ *
+ * If a BlockStats object is passed, this method will also update it if the
+ * block is part of an open file or only on outOfService nodes.
+ *
+ * @param datanode The datanode the block belongs to
+ * @param block The block to check
+ * @param scheduleReconStruction Whether to add the block to the replication
+ * queue if it is not sufficiently replicated.
+ * Passing true will add it to the replication
+ * queue, and false will not.
+ * @param suspectBlocks If non-null check if the block is part of an open
+ * file or only on out of service nodes and update the
+ * passed object accordingly.
+ * @return
+ */
+ private boolean isBlockReplicatedOk(DatanodeDescriptor datanode,
+ BlockInfo block, Boolean scheduleReconStruction,
+ BlockStats suspectBlocks) {
+ if (blockManager.blocksMap.getStoredBlock(block) == null) {
+ LOG.trace("Removing unknown block {}", block);
+ return true;
+ }
+
+ long bcId = block.getBlockCollectionId();
+ if (bcId == INodeId.INVALID_INODE_ID) {
+ // Orphan block, will be invalidated eventually. Skip.
+ return false;
+ }
+
+ final BlockCollection bc = blockManager.getBlockCollection(block);
+ final NumberReplicas num = blockManager.countNodes(block);
+ final int liveReplicas = num.liveReplicas();
+
+ // Schedule low redundancy blocks for reconstruction
+ // if not already pending.
+ boolean isDecommission = datanode.isDecommissionInProgress();
+ boolean isMaintenance = datanode.isEnteringMaintenance();
+
+ boolean neededReconstruction = isDecommission ?
+ blockManager.isNeededReconstruction(block, num) :
+ blockManager.isNeededReconstructionForMaintenance(block, num);
+ if (neededReconstruction && scheduleReconStruction) {
+ if (!blockManager.neededReconstruction.contains(block) &&
+ blockManager.pendingReconstruction.getNumReplicas(block) == 0 &&
+ blockManager.isPopulatingReplQueues()) {
+ // Process these blocks only when active NN is out of safe mode.
+ blockManager.neededReconstruction.add(block,
+ liveReplicas, num.readOnlyReplicas(),
+ num.outOfServiceReplicas(),
+ blockManager.getExpectedRedundancyNum(block));
+ }
+ }
+
+ if (suspectBlocks != null) {
+ // Only if we pass a BlockStats object should we do these
+ // checks, as they should only be checked when processing PendingRep.
+ if (bc.isUnderConstruction()) {
+ INode ucFile = namesystem.getFSDirectory().getInode(bc.getId());
+ if (!(ucFile instanceof INodeFile) ||
+ !ucFile.asFile().isUnderConstruction()) {
+ LOG.warn("File {} is not under construction. Skipping add to " +
+ "low redundancy open files!", ucFile.getLocalName());
+ } else {
+ suspectBlocks.addOpenFile(ucFile.getId());
+ }
+ }
+ if ((liveReplicas == 0) && (num.outOfServiceReplicas() > 0)) {
+ suspectBlocks.incrementOutOfServiceBlocks();
+ }
+ }
+
+ // Even if the block is without sufficient redundancy,
+ // it might not block decommission/maintenance if it
+ // has sufficient redundancy.
+ if (dnAdmin.isSufficient(block, bc, num, isDecommission, isMaintenance)) {
+ return true;
+ }
+ return false;
+ }
+
+ static class BlockStats {
+ private LightWeightHashSet<Long> openFiles =
+ new LightWeightLinkedSet<>();
+ private int openFileBlockCount = 0;
+ private int outOfServiceBlockCount = 0;
+
+ public void addOpenFile(long id) {
+ // Several blocks can be part of the same file so track how
+ // many adds we get, as the same file could be added several times
+ // for different blocks.
+ openFileBlockCount++;
+ openFiles.add(id);
+ }
+
+ public void incrementOutOfServiceBlocks() {
+ outOfServiceBlockCount++;
+ }
+
+ public LightWeightHashSet<Long> getOpenFiles() {
+ return openFiles;
+ }
+
+ public int getOpenFileCount() {
+ return openFileBlockCount;
+ }
+
+ public int getOutOfServiceBlockCount() {
+ return outOfServiceBlockCount;
+ }
+ }
+}
\ No newline at end of file
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeAdminDefaultMonitor.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeAdminDefaultMonitor.java
new file mode 100644
index 0000000..a5650d1
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeAdminDefaultMonitor.java
@@ -0,0 +1,446 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.blockmanagement;
+
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
+import org.apache.hadoop.hdfs.protocol.DatanodeID;
+import org.apache.hadoop.hdfs.server.namenode.INode;
+import org.apache.hadoop.hdfs.server.namenode.INodeFile;
+import org.apache.hadoop.hdfs.server.namenode.INodeId;
+import org.apache.hadoop.hdfs.util.CyclicIteration;
+import org.apache.hadoop.hdfs.util.LightWeightHashSet;
+import org.apache.hadoop.hdfs.util.LightWeightLinkedSet;
+import org.apache.hadoop.util.ChunkedArrayList;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.AbstractList;
+import java.util.TreeMap;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.List;
+import java.util.Iterator;
+
+/**
+ * Checks to see if datanodes have finished DECOMMISSION_INPROGRESS or
+ * ENTERING_MAINTENANCE state.
+ * <p/>
+ * Since this is done while holding the namesystem lock,
+ * the amount of work per monitor tick is limited.
+ */
+
+public class DatanodeAdminDefaultMonitor extends DatanodeAdminMonitorBase
+ implements DatanodeAdminMonitorInterface {
+
+ /**
+ * Map containing the DECOMMISSION_INPROGRESS or ENTERING_MAINTENANCE
+ * datanodes that are being tracked so they can be be marked as
+ * DECOMMISSIONED or IN_MAINTENANCE. Even after the node is marked as
+ * IN_MAINTENANCE, the node remains in the map until
+ * maintenance expires checked during a monitor tick.
+ * <p/>
+ * This holds a set of references to the under-replicated blocks on the DN
+ * at the time the DN is added to the map, i.e. the blocks that are
+ * preventing the node from being marked as decommissioned. During a monitor
+ * tick, this list is pruned as blocks becomes replicated.
+ * <p/>
+ * Note also that the reference to the list of under-replicated blocks
+ * will be null on initial add
+ * <p/>
+ * However, this map can become out-of-date since it is not updated by block
+ * reports or other events. Before being finally marking as decommissioned,
+ * another check is done with the actual block map.
+ */
+ private final TreeMap<DatanodeDescriptor, AbstractList<BlockInfo>>
+ outOfServiceNodeBlocks;
+
+ /**
+ * The maximum number of blocks to check per tick.
+ */
+ private int numBlocksPerCheck;
+
+ /**
+ * The number of blocks that have been checked on this tick.
+ */
+ private int numBlocksChecked = 0;
+ /**
+ * The number of blocks checked after (re)holding lock.
+ */
+ private int numBlocksCheckedPerLock = 0;
+ /**
+ * The number of nodes that have been checked on this tick. Used for
+ * statistics.
+ */
+ private int numNodesChecked = 0;
+ /**
+ * The last datanode in outOfServiceNodeBlocks that we've processed.
+ */
+ private DatanodeDescriptor iterkey = new DatanodeDescriptor(
+ new DatanodeID("", "", "", 0, 0, 0, 0));
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(DatanodeAdminDefaultMonitor.class);
+
+ DatanodeAdminDefaultMonitor() {
+ this.outOfServiceNodeBlocks = new TreeMap<>();
+ }
+
+ @Override
+ protected void processConf() {
+ numBlocksPerCheck = conf.getInt(
+ DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_BLOCKS_PER_INTERVAL_KEY,
+ DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_BLOCKS_PER_INTERVAL_DEFAULT);
+ if (numBlocksPerCheck <= 0) {
+ LOG.error("{} must be greater than zero. Defaulting to {}",
+ DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_BLOCKS_PER_INTERVAL_KEY,
+ DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_BLOCKS_PER_INTERVAL_DEFAULT);
+ numBlocksPerCheck =
+ DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_BLOCKS_PER_INTERVAL_DEFAULT;
+ }
+ LOG.info("Initialized the Default Decommission and Maintenance monitor");
+ }
+
+ private boolean exceededNumBlocksPerCheck() {
+ LOG.trace("Processed {} blocks so far this tick", numBlocksChecked);
+ return numBlocksChecked >= numBlocksPerCheck;
+ }
+
+ @Override
+ public void stopTrackingNode(DatanodeDescriptor dn) {
+ pendingNodes.remove(dn);
+ outOfServiceNodeBlocks.remove(dn);
+ }
+
+ @Override
+ public int getTrackedNodeCount() {
+ return outOfServiceNodeBlocks.size();
+ }
+
+ @Override
+ public int getNumNodesChecked() {
+ return numNodesChecked;
+ }
+
+ @Override
+ public void run() {
+ LOG.debug("DatanodeAdminMonitor is running.");
+ if (!namesystem.isRunning()) {
+ LOG.info("Namesystem is not running, skipping " +
+ "decommissioning/maintenance checks.");
+ return;
+ }
+ // Reset the checked count at beginning of each iteration
+ numBlocksChecked = 0;
+ numBlocksCheckedPerLock = 0;
+ numNodesChecked = 0;
+ // Check decommission or maintenance progress.
+ namesystem.writeLock();
+ try {
+ processPendingNodes();
+ check();
+ } catch (Exception e) {
+ LOG.warn("DatanodeAdminMonitor caught exception when processing node.",
+ e);
+ } finally {
+ namesystem.writeUnlock();
+ }
+ if (numBlocksChecked + numNodesChecked > 0) {
+ LOG.info("Checked {} blocks and {} nodes this tick. {} nodes are now " +
+ "in maintenance or transitioning state. {} nodes pending.",
+ numBlocksChecked, numNodesChecked, outOfServiceNodeBlocks.size(),
+ pendingNodes.size());
+ }
+ }
+
+ /**
+ * Pop datanodes off the pending list and into decomNodeBlocks,
+ * subject to the maxConcurrentTrackedNodes limit.
+ */
+ private void processPendingNodes() {
+ while (!pendingNodes.isEmpty() &&
+ (maxConcurrentTrackedNodes == 0 ||
+ outOfServiceNodeBlocks.size() < maxConcurrentTrackedNodes)) {
+ outOfServiceNodeBlocks.put(pendingNodes.poll(), null);
+ }
+ }
+
+ private void check() {
+ final Iterator<Map.Entry<DatanodeDescriptor, AbstractList<BlockInfo>>>
+ it = new CyclicIteration<>(outOfServiceNodeBlocks,
+ iterkey).iterator();
+ final List<DatanodeDescriptor> toRemove = new ArrayList<>();
+
+ while (it.hasNext() && !exceededNumBlocksPerCheck() && namesystem
+ .isRunning()) {
+ numNodesChecked++;
+ final Map.Entry<DatanodeDescriptor, AbstractList<BlockInfo>>
+ entry = it.next();
+ final DatanodeDescriptor dn = entry.getKey();
+ try {
+ AbstractList<BlockInfo> blocks = entry.getValue();
+ boolean fullScan = false;
+ if (dn.isMaintenance() && dn.maintenanceExpired()) {
+ // If maintenance expires, stop tracking it.
+ dnAdmin.stopMaintenance(dn);
+ toRemove.add(dn);
+ continue;
+ }
+ if (dn.isInMaintenance()) {
+ // The dn is IN_MAINTENANCE and the maintenance hasn't expired yet.
+ continue;
+ }
+ if (blocks == null) {
+ // This is a newly added datanode, run through its list to schedule
+ // under-replicated blocks for replication and collect the blocks
+ // that are insufficiently replicated for further tracking
+ LOG.debug("Newly-added node {}, doing full scan to find " +
+ "insufficiently-replicated blocks.", dn);
+ blocks = handleInsufficientlyStored(dn);
+ outOfServiceNodeBlocks.put(dn, blocks);
+ fullScan = true;
+ } else {
+ // This is a known datanode, check if its # of insufficiently
+ // replicated blocks has dropped to zero and if it can move
+ // to the next state.
+ LOG.debug("Processing {} node {}", dn.getAdminState(), dn);
+ pruneReliableBlocks(dn, blocks);
+ }
+ if (blocks.size() == 0) {
+ if (!fullScan) {
+ // If we didn't just do a full scan, need to re-check with the
+ // full block map.
+ //
+ // We've replicated all the known insufficiently replicated
+ // blocks. Re-check with the full block map before finally
+ // marking the datanode as DECOMMISSIONED or IN_MAINTENANCE.
+ LOG.debug("Node {} has finished replicating current set of "
+ + "blocks, checking with the full block map.", dn);
+ blocks = handleInsufficientlyStored(dn);
+ outOfServiceNodeBlocks.put(dn, blocks);
+ }
+ // If the full scan is clean AND the node liveness is okay,
+ // we can finally mark as DECOMMISSIONED or IN_MAINTENANCE.
+ final boolean isHealthy =
+ blockManager.isNodeHealthyForDecommissionOrMaintenance(dn);
+ if (blocks.size() == 0 && isHealthy) {
+ if (dn.isDecommissionInProgress()) {
+ dnAdmin.setDecommissioned(dn);
+ toRemove.add(dn);
+ } else if (dn.isEnteringMaintenance()) {
+ // IN_MAINTENANCE node remains in the outOfServiceNodeBlocks to
+ // to track maintenance expiration.
+ dnAdmin.setInMaintenance(dn);
+ } else {
+ Preconditions.checkState(false,
+ "Node %s is in an invalid state! "
+ + "Invalid state: %s %s blocks are on this dn.",
+ dn, dn.getAdminState(), blocks.size());
+ }
+ LOG.debug("Node {} is sufficiently replicated and healthy, "
+ + "marked as {}.", dn, dn.getAdminState());
+ } else {
+ LOG.info("Node {} {} healthy."
+ + " It needs to replicate {} more blocks."
+ + " {} is still in progress.", dn,
+ isHealthy ? "is": "isn't", blocks.size(), dn.getAdminState());
+ }
+ } else {
+ LOG.info("Node {} still has {} blocks to replicate "
+ + "before it is a candidate to finish {}.",
+ dn, blocks.size(), dn.getAdminState());
+ }
+ } catch (Exception e) {
+ // Log and postpone to process node when meet exception since it is in
+ // an invalid state.
+ LOG.warn("DatanodeAdminMonitor caught exception when processing node "
+ + "{}.", dn, e);
+ pendingNodes.add(dn);
+ toRemove.add(dn);
+ } finally {
+ iterkey = dn;
+ }
+ }
+ // Remove the datanodes that are DECOMMISSIONED or in service after
+ // maintenance expiration.
+ for (DatanodeDescriptor dn : toRemove) {
+ Preconditions.checkState(dn.isDecommissioned() || dn.isInService(),
+ "Removing node %s that is not yet decommissioned or in service!",
+ dn);
+ outOfServiceNodeBlocks.remove(dn);
+ }
+ }
+
+ /**
+ * Removes reliable blocks from the block list of a datanode.
+ */
+ private void pruneReliableBlocks(final DatanodeDescriptor datanode,
+ AbstractList<BlockInfo> blocks) {
+ processBlocksInternal(datanode, blocks.iterator(), null, true);
+ }
+
+ /**
+ * Returns a list of blocks on a datanode that are insufficiently
+ * replicated or require recovery, i.e. requiring recovery and
+ * should prevent decommission or maintenance.
+ * <p/>
+ * As part of this, it also schedules replication/recovery work.
+ *
+ * @return List of blocks requiring recovery
+ */
+ private AbstractList<BlockInfo> handleInsufficientlyStored(
+ final DatanodeDescriptor datanode) {
+ AbstractList<BlockInfo> insufficient = new ChunkedArrayList<>();
+ processBlocksInternal(datanode, datanode.getBlockIterator(),
+ insufficient, false);
+ return insufficient;
+ }
+
+ /**
+ * Used while checking if DECOMMISSION_INPROGRESS datanodes can be
+ * marked as DECOMMISSIONED or ENTERING_MAINTENANCE datanodes can be
+ * marked as IN_MAINTENANCE. Combines shared logic of pruneReliableBlocks
+ * and handleInsufficientlyStored.
+ *
+ * @param datanode Datanode
+ * @param it Iterator over the blocks on the
+ * datanode
+ * @param insufficientList Return parameter. If it's not null,
+ * will contain the insufficiently
+ * replicated-blocks from the list.
+ * @param pruneReliableBlocks whether to remove blocks reliable
+ * enough from the iterator
+ */
+ private void processBlocksInternal(
+ final DatanodeDescriptor datanode,
+ final Iterator<BlockInfo> it,
+ final List<BlockInfo> insufficientList,
+ boolean pruneReliableBlocks) {
+ boolean firstReplicationLog = true;
+ // Low redundancy in UC Blocks only
+ int lowRedundancyBlocksInOpenFiles = 0;
+ LightWeightHashSet<Long> lowRedundancyOpenFiles =
+ new LightWeightLinkedSet<>();
+ // All low redundancy blocks. Includes lowRedundancyOpenFiles.
+ int lowRedundancyBlocks = 0;
+ // All maintenance and decommission replicas.
+ int outOfServiceOnlyReplicas = 0;
+ while (it.hasNext()) {
+ if (insufficientList == null
+ && numBlocksCheckedPerLock >= numBlocksPerCheck) {
+ // During fullscan insufficientlyReplicated will NOT be null, iterator
+ // will be DN's iterator. So should not yield lock, otherwise
+ // ConcurrentModificationException could occur.
+ // Once the fullscan done, iterator will be a copy. So can yield the
+ // lock.
+ // Yielding is required in case of block number is greater than the
+ // configured per-iteration-limit.
+ namesystem.writeUnlock();
+ try {
+ LOG.debug("Yielded lock during decommission/maintenance check");
+ Thread.sleep(0, 500);
+ } catch (InterruptedException ignored) {
+ return;
+ }
+ // reset
+ numBlocksCheckedPerLock = 0;
+ namesystem.writeLock();
+ }
+ numBlocksChecked++;
+ numBlocksCheckedPerLock++;
+ final BlockInfo block = it.next();
+ // Remove the block from the list if it's no longer in the block map,
+ // e.g. the containing file has been deleted
+ if (blockManager.blocksMap.getStoredBlock(block) == null) {
+ LOG.trace("Removing unknown block {}", block);
+ it.remove();
+ continue;
+ }
+
+ long bcId = block.getBlockCollectionId();
+ if (bcId == INodeId.INVALID_INODE_ID) {
+ // Orphan block, will be invalidated eventually. Skip.
+ continue;
+ }
+
+ final BlockCollection bc = blockManager.getBlockCollection(block);
+ final NumberReplicas num = blockManager.countNodes(block);
+ final int liveReplicas = num.liveReplicas();
+
+ // Schedule low redundancy blocks for reconstruction
+ // if not already pending.
+ boolean isDecommission = datanode.isDecommissionInProgress();
+ boolean isMaintenance = datanode.isEnteringMaintenance();
+ boolean neededReconstruction = isDecommission ?
+ blockManager.isNeededReconstruction(block, num) :
+ blockManager.isNeededReconstructionForMaintenance(block, num);
+ if (neededReconstruction) {
+ if (!blockManager.neededReconstruction.contains(block) &&
+ blockManager.pendingReconstruction.getNumReplicas(block) == 0 &&
+ blockManager.isPopulatingReplQueues()) {
+ // Process these blocks only when active NN is out of safe mode.
+ blockManager.neededReconstruction.add(block,
+ liveReplicas, num.readOnlyReplicas(),
+ num.outOfServiceReplicas(),
+ blockManager.getExpectedRedundancyNum(block));
+ }
+ }
+
+ // Even if the block is without sufficient redundancy,
+ // it might not block decommission/maintenance if it
+ // has sufficient redundancy.
+ if (dnAdmin.isSufficient(block, bc, num, isDecommission, isMaintenance)) {
+ if (pruneReliableBlocks) {
+ it.remove();
+ }
+ continue;
+ }
+
+ // We've found a block without sufficient redundancy.
+ if (insufficientList != null) {
+ insufficientList.add(block);
+ }
+ // Log if this is our first time through
+ if (firstReplicationLog) {
+ dnAdmin.logBlockReplicationInfo(block, bc, datanode, num,
+ blockManager.blocksMap.getStorages(block));
+ firstReplicationLog = false;
+ }
+ // Update various counts
+ lowRedundancyBlocks++;
+ if (bc.isUnderConstruction()) {
+ INode ucFile = namesystem.getFSDirectory().getInode(bc.getId());
+ if (!(ucFile instanceof INodeFile) ||
+ !ucFile.asFile().isUnderConstruction()) {
+ LOG.warn("File {} is not under construction. Skipping add to " +
+ "low redundancy open files!", ucFile.getLocalName());
+ } else {
+ lowRedundancyBlocksInOpenFiles++;
+ lowRedundancyOpenFiles.add(ucFile.getId());
+ }
+ }
+ if ((liveReplicas == 0) && (num.outOfServiceReplicas() > 0)) {
+ outOfServiceOnlyReplicas++;
+ }
+ }
+
+ datanode.getLeavingServiceStatus().set(lowRedundancyBlocksInOpenFiles,
+ lowRedundancyOpenFiles, lowRedundancyBlocks,
+ outOfServiceOnlyReplicas);
+ }
+}
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeAdminManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeAdminManager.java
index aaf1745..0771c28 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeAdminManager.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeAdminManager.java
@@ -20,37 +20,20 @@ package org.apache.hadoop.hdfs.server.blockmanagement;
import static com.google.common.base.Preconditions.checkArgument;
import static org.apache.hadoop.util.Time.monotonicNow;
-import java.util.AbstractList;
-import java.util.ArrayDeque;
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
import java.util.Queue;
-import java.util.TreeMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
-
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdfs.DFSConfigKeys;
-import org.apache.hadoop.hdfs.protocol.DatanodeID;
-import org.apache.hadoop.hdfs.server.namenode.INode;
-import org.apache.hadoop.hdfs.server.namenode.INodeFile;
-import org.apache.hadoop.hdfs.server.namenode.INodeId;
import org.apache.hadoop.hdfs.server.namenode.NameNode;
import org.apache.hadoop.hdfs.server.namenode.Namesystem;
-import org.apache.hadoop.hdfs.util.CyclicIteration;
-import org.apache.hadoop.hdfs.util.LightWeightHashSet;
-import org.apache.hadoop.hdfs.util.LightWeightLinkedSet;
-import org.apache.hadoop.util.ChunkedArrayList;
+import org.apache.hadoop.util.ReflectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
/**
@@ -100,35 +83,7 @@ public class DatanodeAdminManager {
private final HeartbeatManager hbManager;
private final ScheduledExecutorService executor;
- /**
- * Map containing the DECOMMISSION_INPROGRESS or ENTERING_MAINTENANCE
- * datanodes that are being tracked so they can be be marked as
- * DECOMMISSIONED or IN_MAINTENANCE. Even after the node is marked as
- * IN_MAINTENANCE, the node remains in the map until
- * maintenance expires checked during a monitor tick.
- * <p/>
- * This holds a set of references to the under-replicated blocks on the DN at
- * the time the DN is added to the map, i.e. the blocks that are preventing
- * the node from being marked as decommissioned. During a monitor tick, this
- * list is pruned as blocks becomes replicated.
- * <p/>
- * Note also that the reference to the list of under-replicated blocks
- * will be null on initial add
- * <p/>
- * However, this map can become out-of-date since it is not updated by block
- * reports or other events. Before being finally marking as decommissioned,
- * another check is done with the actual block map.
- */
- private final TreeMap<DatanodeDescriptor, AbstractList<BlockInfo>>
- outOfServiceNodeBlocks;
-
- /**
- * Tracking a node in outOfServiceNodeBlocks consumes additional memory. To
- * limit the impact on NN memory consumption, we limit the number of nodes in
- * outOfServiceNodeBlocks. Additional nodes wait in pendingNodes.
- */
- private final Queue<DatanodeDescriptor> pendingNodes;
- private Monitor monitor = null;
+ private DatanodeAdminMonitorInterface monitor = null;
DatanodeAdminManager(final Namesystem namesystem,
final BlockManager blockManager, final HeartbeatManager hbManager) {
@@ -139,8 +94,6 @@ public class DatanodeAdminManager {
executor = Executors.newScheduledThreadPool(1,
new ThreadFactoryBuilder().setNameFormat("DatanodeAdminMonitor-%d")
.setDaemon(true).build());
- outOfServiceNodeBlocks = new TreeMap<>();
- pendingNodes = new ArrayDeque<>();
}
/**
@@ -181,7 +134,20 @@ public class DatanodeAdminManager {
"value for "
+ DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_MAX_CONCURRENT_TRACKED_NODES);
- monitor = new Monitor(blocksPerInterval, maxConcurrentTrackedNodes);
+ Class cls = null;
+ try {
+ cls = conf.getClass(
+ DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_MONITOR_CLASS,
+ DatanodeAdminDefaultMonitor.class);
+ monitor =
+ (DatanodeAdminMonitorInterface)ReflectionUtils.newInstance(cls, conf);
+ monitor.setBlockManager(blockManager);
+ monitor.setNameSystem(namesystem);
+ monitor.setDatanodeAdminManager(this);
+ } catch (Exception e) {
+ throw new RuntimeException("Unable to create the Decommission monitor " +
+ "from "+cls, e);
+ }
executor.scheduleAtFixedRate(monitor, intervalSecs, intervalSecs,
TimeUnit.SECONDS);
@@ -217,7 +183,7 @@ public class DatanodeAdminManager {
node, storage, storage.numBlocks());
}
node.getLeavingServiceStatus().setStartTime(monotonicNow());
- pendingNodes.add(node);
+ monitor.startTrackingNode(node);
}
} else {
LOG.trace("startDecommission: Node {} in {}, nothing to do.",
@@ -240,8 +206,7 @@ public class DatanodeAdminManager {
blockManager.processExtraRedundancyBlocksOnInService(node);
}
// Remove from tracking in DatanodeAdminManager
- pendingNodes.remove(node);
- outOfServiceNodeBlocks.remove(node);
+ monitor.stopTrackingNode(node);
} else {
LOG.trace("stopDecommission: Node {} in {}, nothing to do.",
node, node.getAdminState());
@@ -271,7 +236,7 @@ public class DatanodeAdminManager {
}
// Track the node regardless whether it is ENTERING_MAINTENANCE or
// IN_MAINTENANCE to support maintenance expiration.
- pendingNodes.add(node);
+ monitor.startTrackingNode(node);
} else {
LOG.trace("startMaintenance: Node {} in {}, nothing to do.",
node, node.getAdminState());
@@ -319,20 +284,19 @@ public class DatanodeAdminManager {
}
// Remove from tracking in DatanodeAdminManager
- pendingNodes.remove(node);
- outOfServiceNodeBlocks.remove(node);
+ monitor.stopTrackingNode(node);
} else {
LOG.trace("stopMaintenance: Node {} in {}, nothing to do.",
node, node.getAdminState());
}
}
- private void setDecommissioned(DatanodeDescriptor dn) {
+ protected void setDecommissioned(DatanodeDescriptor dn) {
dn.setDecommissioned();
LOG.info("Decommissioning complete for node {}", dn);
}
- private void setInMaintenance(DatanodeDescriptor dn) {
+ protected void setInMaintenance(DatanodeDescriptor dn) {
dn.setInMaintenance();
LOG.info("Node {} has entered maintenance mode.", dn);
}
@@ -344,7 +308,7 @@ public class DatanodeAdminManager {
* always necessary, hence "sufficient".
* @return true if sufficient, else false.
*/
- private boolean isSufficient(BlockInfo block, BlockCollection bc,
+ protected boolean isSufficient(BlockInfo block, BlockCollection bc,
NumberReplicas numberReplicas,
boolean isDecommission,
boolean isMaintenance) {
@@ -388,7 +352,7 @@ public class DatanodeAdminManager {
return false;
}
- private void logBlockReplicationInfo(BlockInfo block,
+ protected void logBlockReplicationInfo(BlockInfo block,
BlockCollection bc,
DatanodeDescriptor srcNode, NumberReplicas num,
Iterable<DatanodeStorageInfo> storages) {
@@ -423,380 +387,27 @@ public class DatanodeAdminManager {
@VisibleForTesting
public int getNumPendingNodes() {
- return pendingNodes.size();
+ return monitor.getPendingNodeCount();
}
@VisibleForTesting
public int getNumTrackedNodes() {
- return outOfServiceNodeBlocks.size();
+ return monitor.getTrackedNodeCount();
}
@VisibleForTesting
public int getNumNodesChecked() {
- return monitor.numNodesChecked;
+ return monitor.getNumNodesChecked();
}
@VisibleForTesting
public Queue<DatanodeDescriptor> getPendingNodes() {
- return pendingNodes;
- }
-
- /**
- * Checks to see if datanodes have finished DECOMMISSION_INPROGRESS or
- * ENTERING_MAINTENANCE state.
- * <p/>
- * Since this is done while holding the namesystem lock,
- * the amount of work per monitor tick is limited.
- */
- private class Monitor implements Runnable {
- /**
- * The maximum number of blocks to check per tick.
- */
- private final int numBlocksPerCheck;
- /**
- * The maximum number of nodes to track in outOfServiceNodeBlocks.
- * A value of 0 means no limit.
- */
- private final int maxConcurrentTrackedNodes;
- /**
- * The number of blocks that have been checked on this tick.
- */
- private int numBlocksChecked = 0;
- /**
- * The number of blocks checked after (re)holding lock.
- */
- private int numBlocksCheckedPerLock = 0;
- /**
- * The number of nodes that have been checked on this tick. Used for
- * statistics.
- */
- private int numNodesChecked = 0;
- /**
- * The last datanode in outOfServiceNodeBlocks that we've processed.
- */
- private DatanodeDescriptor iterkey = new DatanodeDescriptor(
- new DatanodeID("", "", "", 0, 0, 0, 0));
-
- Monitor(int numBlocksPerCheck, int maxConcurrentTrackedNodes) {
- this.numBlocksPerCheck = numBlocksPerCheck;
- this.maxConcurrentTrackedNodes = maxConcurrentTrackedNodes;
- }
-
- private boolean exceededNumBlocksPerCheck() {
- LOG.trace("Processed {} blocks so far this tick", numBlocksChecked);
- return numBlocksChecked >= numBlocksPerCheck;
- }
-
- @Override
- public void run() {
- LOG.debug("DatanodeAdminMonitor is running.");
- if (!namesystem.isRunning()) {
- LOG.info("Namesystem is not running, skipping " +
- "decommissioning/maintenance checks.");
- return;
- }
- // Reset the checked count at beginning of each iteration
- numBlocksChecked = 0;
- numBlocksCheckedPerLock = 0;
- numNodesChecked = 0;
- // Check decommission or maintenance progress.
- namesystem.writeLock();
- try {
- processPendingNodes();
- check();
- } catch (Exception e) {
- LOG.warn("DatanodeAdminMonitor caught exception when processing node.",
- e);
- } finally {
- namesystem.writeUnlock();
- }
- if (numBlocksChecked + numNodesChecked > 0) {
- LOG.info("Checked {} blocks and {} nodes this tick. {} nodes are now " +
- "in maintenance or transitioning state. {} nodes pending.",
- numBlocksChecked, numNodesChecked, outOfServiceNodeBlocks.size(),
- pendingNodes.size());
- }
- }
-
- /**
- * Pop datanodes off the pending list and into decomNodeBlocks,
- * subject to the maxConcurrentTrackedNodes limit.
- */
- private void processPendingNodes() {
- while (!pendingNodes.isEmpty() &&
- (maxConcurrentTrackedNodes == 0 ||
- outOfServiceNodeBlocks.size() < maxConcurrentTrackedNodes)) {
- outOfServiceNodeBlocks.put(pendingNodes.poll(), null);
- }
- }
-
- private void check() {
- final Iterator<Map.Entry<DatanodeDescriptor, AbstractList<BlockInfo>>>
- it = new CyclicIteration<>(outOfServiceNodeBlocks,
- iterkey).iterator();
- final List<DatanodeDescriptor> toRemove = new ArrayList<>();
-
- while (it.hasNext() && !exceededNumBlocksPerCheck() && namesystem
- .isRunning()) {
- numNodesChecked++;
- final Map.Entry<DatanodeDescriptor, AbstractList<BlockInfo>>
- entry = it.next();
- final DatanodeDescriptor dn = entry.getKey();
- try {
- AbstractList<BlockInfo> blocks = entry.getValue();
- boolean fullScan = false;
- if (dn.isMaintenance() && dn.maintenanceExpired()) {
- // If maintenance expires, stop tracking it.
- stopMaintenance(dn);
- toRemove.add(dn);
- continue;
- }
- if (dn.isInMaintenance()) {
- // The dn is IN_MAINTENANCE and the maintenance hasn't expired yet.
- continue;
- }
- if (blocks == null) {
- // This is a newly added datanode, run through its list to schedule
- // under-replicated blocks for replication and collect the blocks
- // that are insufficiently replicated for further tracking
- LOG.debug("Newly-added node {}, doing full scan to find " +
- "insufficiently-replicated blocks.", dn);
- blocks = handleInsufficientlyStored(dn);
- outOfServiceNodeBlocks.put(dn, blocks);
- fullScan = true;
- } else {
- // This is a known datanode, check if its # of insufficiently
- // replicated blocks has dropped to zero and if it can move
- // to the next state.
- LOG.debug("Processing {} node {}", dn.getAdminState(), dn);
- pruneReliableBlocks(dn, blocks);
- }
- if (blocks.size() == 0) {
- if (!fullScan) {
- // If we didn't just do a full scan, need to re-check with the
- // full block map.
- //
- // We've replicated all the known insufficiently replicated
- // blocks. Re-check with the full block map before finally
- // marking the datanode as DECOMMISSIONED or IN_MAINTENANCE.
- LOG.debug("Node {} has finished replicating current set of "
- + "blocks, checking with the full block map.", dn);
- blocks = handleInsufficientlyStored(dn);
- outOfServiceNodeBlocks.put(dn, blocks);
- }
- // If the full scan is clean AND the node liveness is okay,
- // we can finally mark as DECOMMISSIONED or IN_MAINTENANCE.
- final boolean isHealthy =
- blockManager.isNodeHealthyForDecommissionOrMaintenance(dn);
- if (blocks.size() == 0 && isHealthy) {
- if (dn.isDecommissionInProgress()) {
- setDecommissioned(dn);
- toRemove.add(dn);
- } else if (dn.isEnteringMaintenance()) {
- // IN_MAINTENANCE node remains in the outOfServiceNodeBlocks to
- // to track maintenance expiration.
- setInMaintenance(dn);
- } else {
- Preconditions.checkState(false,
- "Node %s is in an invalid state! "
- + "Invalid state: %s %s blocks are on this dn.",
- dn, dn.getAdminState(), blocks.size());
- }
- LOG.debug("Node {} is sufficiently replicated and healthy, "
- + "marked as {}.", dn, dn.getAdminState());
- } else {
- LOG.info("Node {} {} healthy."
- + " It needs to replicate {} more blocks."
- + " {} is still in progress.", dn,
- isHealthy ? "is": "isn't", blocks.size(), dn.getAdminState());
- }
- } else {
- LOG.info("Node {} still has {} blocks to replicate "
- + "before it is a candidate to finish {}.",
- dn, blocks.size(), dn.getAdminState());
- }
- } catch (Exception e) {
- // Log and postpone to process node when meet exception since it is in
- // an invalid state.
- LOG.warn("DatanodeAdminMonitor caught exception when processing node "
- + "{}.", dn, e);
- pendingNodes.add(dn);
- toRemove.add(dn);
- } finally {
- iterkey = dn;
- }
- }
- // Remove the datanodes that are DECOMMISSIONED or in service after
- // maintenance expiration.
- for (DatanodeDescriptor dn : toRemove) {
- Preconditions.checkState(dn.isDecommissioned() || dn.isInService(),
- "Removing node %s that is not yet decommissioned or in service!",
- dn);
- outOfServiceNodeBlocks.remove(dn);
- }
- }
-
- /**
- * Removes reliable blocks from the block list of a datanode.
- */
- private void pruneReliableBlocks(final DatanodeDescriptor datanode,
- AbstractList<BlockInfo> blocks) {
- processBlocksInternal(datanode, blocks.iterator(), null, true);
- }
-
- /**
- * Returns a list of blocks on a datanode that are insufficiently
- * replicated or require recovery, i.e. requiring recovery and
- * should prevent decommission or maintenance.
- * <p/>
- * As part of this, it also schedules replication/recovery work.
- *
- * @return List of blocks requiring recovery
- */
- private AbstractList<BlockInfo> handleInsufficientlyStored(
- final DatanodeDescriptor datanode) {
- AbstractList<BlockInfo> insufficient = new ChunkedArrayList<>();
- processBlocksInternal(datanode, datanode.getBlockIterator(),
- insufficient, false);
- return insufficient;
- }
-
- /**
- * Used while checking if DECOMMISSION_INPROGRESS datanodes can be
- * marked as DECOMMISSIONED or ENTERING_MAINTENANCE datanodes can be
- * marked as IN_MAINTENANCE. Combines shared logic of pruneReliableBlocks
- * and handleInsufficientlyStored.
- *
- * @param datanode Datanode
- * @param it Iterator over the blocks on the
- * datanode
- * @param insufficientList Return parameter. If it's not null,
- * will contain the insufficiently
- * replicated-blocks from the list.
- * @param pruneReliableBlocks whether to remove blocks reliable
- * enough from the iterator
- */
- private void processBlocksInternal(
- final DatanodeDescriptor datanode,
- final Iterator<BlockInfo> it,
- final List<BlockInfo> insufficientList,
- boolean pruneReliableBlocks) {
- boolean firstReplicationLog = true;
- // Low redundancy in UC Blocks only
- int lowRedundancyBlocksInOpenFiles = 0;
- LightWeightHashSet<Long> lowRedundancyOpenFiles =
- new LightWeightLinkedSet<>();
- // All low redundancy blocks. Includes lowRedundancyOpenFiles.
- int lowRedundancyBlocks = 0;
- // All maintenance and decommission replicas.
- int outOfServiceOnlyReplicas = 0;
- while (it.hasNext()) {
- if (insufficientList == null
- && numBlocksCheckedPerLock >= numBlocksPerCheck) {
- // During fullscan insufficientlyReplicated will NOT be null, iterator
- // will be DN's iterator. So should not yield lock, otherwise
- // ConcurrentModificationException could occur.
- // Once the fullscan done, iterator will be a copy. So can yield the
- // lock.
- // Yielding is required in case of block number is greater than the
- // configured per-iteration-limit.
- namesystem.writeUnlock();
- try {
- LOG.debug("Yielded lock during decommission/maintenance check");
- Thread.sleep(0, 500);
- } catch (InterruptedException ignored) {
- return;
- }
- // reset
- numBlocksCheckedPerLock = 0;
- namesystem.writeLock();
- }
- numBlocksChecked++;
- numBlocksCheckedPerLock++;
- final BlockInfo block = it.next();
- // Remove the block from the list if it's no longer in the block map,
- // e.g. the containing file has been deleted
- if (blockManager.blocksMap.getStoredBlock(block) == null) {
- LOG.trace("Removing unknown block {}", block);
- it.remove();
- continue;
- }
-
- long bcId = block.getBlockCollectionId();
- if (bcId == INodeId.INVALID_INODE_ID) {
- // Orphan block, will be invalidated eventually. Skip.
- continue;
- }
-
- final BlockCollection bc = blockManager.getBlockCollection(block);
- final NumberReplicas num = blockManager.countNodes(block);
- final int liveReplicas = num.liveReplicas();
-
- // Schedule low redundancy blocks for reconstruction
- // if not already pending.
- boolean isDecommission = datanode.isDecommissionInProgress();
- boolean isMaintenance = datanode.isEnteringMaintenance();
- boolean neededReconstruction = isDecommission ?
- blockManager.isNeededReconstruction(block, num) :
- blockManager.isNeededReconstructionForMaintenance(block, num);
- if (neededReconstruction) {
- if (!blockManager.neededReconstruction.contains(block) &&
- blockManager.pendingReconstruction.getNumReplicas(block) == 0 &&
- blockManager.isPopulatingReplQueues()) {
- // Process these blocks only when active NN is out of safe mode.
- blockManager.neededReconstruction.add(block,
- liveReplicas, num.readOnlyReplicas(),
- num.outOfServiceReplicas(),
- blockManager.getExpectedRedundancyNum(block));
- }
- }
-
- // Even if the block is without sufficient redundancy,
- // it might not block decommission/maintenance if it
- // has sufficient redundancy.
- if (isSufficient(block, bc, num, isDecommission, isMaintenance)) {
- if (pruneReliableBlocks) {
- it.remove();
- }
- continue;
- }
-
- // We've found a block without sufficient redundancy.
- if (insufficientList != null) {
- insufficientList.add(block);
- }
- // Log if this is our first time through
- if (firstReplicationLog) {
- logBlockReplicationInfo(block, bc, datanode, num,
- blockManager.blocksMap.getStorages(block));
- firstReplicationLog = false;
- }
- // Update various counts
- lowRedundancyBlocks++;
- if (bc.isUnderConstruction()) {
- INode ucFile = namesystem.getFSDirectory().getInode(bc.getId());
- if (!(ucFile instanceof INodeFile) ||
- !ucFile.asFile().isUnderConstruction()) {
- LOG.warn("File {} is not under construction. Skipping add to " +
- "low redundancy open files!", ucFile.getLocalName());
- } else {
- lowRedundancyBlocksInOpenFiles++;
- lowRedundancyOpenFiles.add(ucFile.getId());
- }
- }
- if ((liveReplicas == 0) && (num.outOfServiceReplicas() > 0)) {
- outOfServiceOnlyReplicas++;
- }
- }
-
- datanode.getLeavingServiceStatus().set(lowRedundancyBlocksInOpenFiles,
- lowRedundancyOpenFiles, lowRedundancyBlocks,
- outOfServiceOnlyReplicas);
- }
+ return monitor.getPendingNodes();
}
@VisibleForTesting
void runMonitorForTest() throws ExecutionException, InterruptedException {
executor.submit(monitor).get();
}
-}
+
+}
\ No newline at end of file
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeAdminMonitorBase.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeAdminMonitorBase.java
new file mode 100644
index 0000000..9eee241
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeAdminMonitorBase.java
@@ -0,0 +1,154 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.blockmanagement;
+
+import org.apache.hadoop.conf.Configurable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
+import org.apache.hadoop.hdfs.server.namenode.Namesystem;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayDeque;
+import java.util.Queue;
+
+/**
+ * This abstract class provides some base methods which are inherited by
+ * the DatanodeAdmin BackOff and Default Monitors, which control decommission
+ * and maintenance mode.
+ */
+public abstract class DatanodeAdminMonitorBase
+ implements DatanodeAdminMonitorInterface, Configurable {
+
+ protected BlockManager blockManager;
+ protected Namesystem namesystem;
+ protected DatanodeAdminManager dnAdmin;
+ protected Configuration conf;
+
+ protected final Queue<DatanodeDescriptor> pendingNodes = new ArrayDeque<>();
+
+ /**
+ * The maximum number of nodes to track in outOfServiceNodeBlocks.
+ * A value of 0 means no limit.
+ */
+ protected int maxConcurrentTrackedNodes;
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(DatanodeAdminMonitorBase.class);
+
+ /**
+ * Set the cluster namesystem.
+ *
+ * @param ns The namesystem for the cluster
+ */
+ @Override
+ public void setNameSystem(Namesystem ns) {
+ this.namesystem = ns;
+ }
+
+ /**
+ * Set the blockmanager for the cluster.
+ *
+ * @param bm The cluster BlockManager
+ */
+ @Override
+ public void setBlockManager(BlockManager bm) {
+ this.blockManager = bm;
+ }
+
+ /**
+ * Set the DatanodeAdminManager instance in use in the namenode.
+ *
+ * @param admin The current DatanodeAdminManager
+ */
+ @Override
+ public void setDatanodeAdminManager(DatanodeAdminManager admin) {
+ this.dnAdmin = admin;
+ }
+
+ /**
+ * Used by the Configurable interface, which is used by ReflectionUtils
+ * to create an instance of the monitor class. This method will be called to
+ * pass the Configuration to the new object.
+ *
+ * @param conf configuration to be used
+ */
+ @Override
+ public void setConf(Configuration conf) {
+ this.conf = conf;
+ this.maxConcurrentTrackedNodes = conf.getInt(
+ DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_MAX_CONCURRENT_TRACKED_NODES,
+ DFSConfigKeys
+ .DFS_NAMENODE_DECOMMISSION_MAX_CONCURRENT_TRACKED_NODES_DEFAULT);
+ if (this.maxConcurrentTrackedNodes < 0) {
+ LOG.error("{} is set to an invalid value, it must be zero or greater. "+
+ "Defaulting to {}",
+ DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_MAX_CONCURRENT_TRACKED_NODES,
+ DFSConfigKeys
+ .DFS_NAMENODE_DECOMMISSION_MAX_CONCURRENT_TRACKED_NODES_DEFAULT);
+ this.maxConcurrentTrackedNodes =
+ DFSConfigKeys
+ .DFS_NAMENODE_DECOMMISSION_MAX_CONCURRENT_TRACKED_NODES_DEFAULT;
+ }
+ processConf();
+ }
+
+ /**
+ * Get the current Configuration stored in this object.
+ *
+ * @return Configuration used when the object was created
+ */
+ @Override
+ public Configuration getConf() {
+ return this.conf;
+ }
+
+ /**
+ * Abstract method which must be implemented by the sub-classes to process
+ * set various instance variables from the Configuration passed at object
+ * creation time.
+ */
+ protected abstract void processConf();
+
+ /**
+ * Start tracking a node for decommission or maintenance. The given Datanode
+ * will be queued for later processing in pendingNodes. This method must be
+ * called under the namenode write lock.
+ * @param dn The datanode to start tracking
+ */
+ @Override
+ public void startTrackingNode(DatanodeDescriptor dn) {
+ pendingNodes.add(dn);
+ }
+
+ /**
+ * Get the number of datanodes nodes in the pending queue. Ie the count of
+ * nodes waiting to decommission but have not yet started the process.
+ *
+ * @return The count of pending nodes
+ */
+ @Override
+ public int getPendingNodeCount() {
+ return pendingNodes.size();
+ }
+
+ @Override
+ public Queue<DatanodeDescriptor> getPendingNodes() {
+ return pendingNodes;
+ }
+}
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeAdminMonitorInterface.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeAdminMonitorInterface.java
new file mode 100644
index 0000000..f34c005
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeAdminMonitorInterface.java
@@ -0,0 +1,39 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.blockmanagement;
+
+import org.apache.hadoop.hdfs.server.namenode.Namesystem;
+import java.util.Queue;
+
+/**
+ * Interface used to implement a decommission and maintenance monitor class,
+ * which is instantiated by the DatanodeAdminManager class.
+ */
+
+public interface DatanodeAdminMonitorInterface extends Runnable {
+ void stopTrackingNode(DatanodeDescriptor dn);
+ void startTrackingNode(DatanodeDescriptor dn);
+ int getPendingNodeCount();
+ int getTrackedNodeCount();
+ int getNumNodesChecked();
+ Queue<DatanodeDescriptor> getPendingNodes();
+
+ void setBlockManager(BlockManager bm);
+ void setDatanodeAdminManager(DatanodeAdminManager dnm);
+ void setNameSystem(Namesystem ns);
+}
\ No newline at end of file
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml b/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml
index 640e4ab..867828e 100755
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml
@@ -1093,6 +1093,41 @@
</property>
<property>
+ <name>dfs.namenode.decommission.monitor.class</name>
+ <value>org.apache.hadoop.hdfs.server.blockmanagement.DatanodeAdminDefaultMonitor</value>
+ <description>
+ Determines the implementation used for the decommission manager. The only
+ valid options are:
+
+ org.apache.hadoop.hdfs.server.blockmanagement.DatanodeAdminDefaultMonitor
+ org.apache.hadoop.hdfs.server.blockmanagement.DatanodeAdminBackoffMonitor
+
+ </description>
+</property>
+
+<property>
+ <name>dfs.namenode.decommission.backoff.monitor.pending.limit</name>
+ <value>10000</value>
+ <description>
+ When the Backoff monitor is enabled, determines the maximum number of blocks
+ related to decommission and maintenance operations that can be loaded
+ into the replication queue at any given time. Every
+ dfs.namenode.decommission.interval seconds, the list is checked to see if
+ the blocks have become fully replicated and then further blocks are added
+ to reach the limit defined in this parameter.
+ </description>
+</property>
+
+<property>
+ <name>dfs.namenode.decommission.backoff.monitor.pending.blocks.per.lock</name>
+ <value>1000</value>
+ <description>
+ When loading blocks into the replication queue, release the namenode write
+ lock after the defined number of blocks have been processed.
+ </description>
+</property>
+
+<property>
<name>dfs.namenode.redundancy.interval.seconds</name>
<value>3s</value>
<description>The periodicity in seconds with which the namenode computes
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDecommissionWithBackoffMonitor.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDecommissionWithBackoffMonitor.java
new file mode 100644
index 0000000..9c37a19
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDecommissionWithBackoffMonitor.java
@@ -0,0 +1,51 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.server.blockmanagement
+ .DatanodeAdminBackoffMonitor;
+import org.apache.hadoop.hdfs.server.blockmanagement
+ .DatanodeAdminMonitorInterface;
+import org.junit.Test;
+
+import java.io.IOException;
+
+/**
+ * This class tests decommission using the alternative backoff monitor. It
+ * works by sub-classing the original decommission tests and then setting the
+ * config to enable the alternative monitor version.
+ */
+
+public class TestDecommissionWithBackoffMonitor extends TestDecommission {
+
+ @Override
+ public void setup() throws IOException {
+ super.setup();
+ Configuration conf = getConf();
+ conf.setClass(DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_MONITOR_CLASS,
+ DatanodeAdminBackoffMonitor.class, DatanodeAdminMonitorInterface.class);
+ }
+
+ @Override
+ @Test
+ public void testBlocksPerInterval() {
+ // This test is not valid in the decommission monitor V2 so
+ // effectively commenting it out by overriding and having it do nothing.
+ }
+}
\ No newline at end of file
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDecommissionWithStriped.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDecommissionWithStriped.java
index eb23365..be3abab 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDecommissionWithStriped.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDecommissionWithStriped.java
@@ -109,10 +109,13 @@ public class TestDecommissionWithStriped {
private BlockManager bm;
private DFSClient client;
+ protected Configuration createConfiguration() {
+ return new HdfsConfiguration();
+ }
+
@Before
public void setup() throws IOException {
- conf = new HdfsConfiguration();
-
+ conf = createConfiguration();
// Set up the hosts/exclude files.
localFileSys = FileSystem.getLocal(conf);
Path workingDir = localFileSys.getWorkingDirectory();
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDecommissionWithStripedBackoffMonitor.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDecommissionWithStripedBackoffMonitor.java
new file mode 100644
index 0000000..d381673
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDecommissionWithStripedBackoffMonitor.java
@@ -0,0 +1,40 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.server.blockmanagement
+ .DatanodeAdminBackoffMonitor;
+import org.apache.hadoop.hdfs.server.blockmanagement
+ .DatanodeAdminMonitorInterface;
+
+/**
+ * Class to run all the stripped decommission tests with the
+ * DatanodeAdminBackoffMonitor.
+ */
+public class TestDecommissionWithStripedBackoffMonitor
+ extends TestDecommissionWithStriped{
+
+ @Override
+ protected Configuration createConfiguration() {
+ Configuration conf = new Configuration();
+ conf.setClass(DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_MONITOR_CLASS,
+ DatanodeAdminBackoffMonitor.class, DatanodeAdminMonitorInterface.class);
+ return conf;
+ }
+}
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestDecommissioningStatus.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestDecommissioningStatus.java
index cfebff7..ad99c11 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestDecommissioningStatus.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestDecommissioningStatus.java
@@ -65,20 +65,31 @@ import org.junit.Test;
* This class tests the decommissioning of nodes.
*/
public class TestDecommissioningStatus {
- private static final long seed = 0xDEADBEEFL;
- private static final int blockSize = 8192;
- private static final int fileSize = 16384;
- private static final int numDatanodes = 2;
- private static MiniDFSCluster cluster;
- private static FileSystem fileSys;
- private static HostsFileWriter hostsFileWriter;
- private static Configuration conf;
+ private final long seed = 0xDEADBEEFL;
+ private final int blockSize = 8192;
+ private final int fileSize = 16384;
+ private final int numDatanodes = 2;
+ private MiniDFSCluster cluster;
+ private FileSystem fileSys;
+ private HostsFileWriter hostsFileWriter;
+ private Configuration conf;
private Logger LOG;
final ArrayList<String> decommissionedNodes = new ArrayList<String>(numDatanodes);
-
- @Before
- public void setUp() throws Exception {
+
+ protected MiniDFSCluster getCluster() {
+ return cluster;
+ }
+
+ protected FileSystem getFileSys() {
+ return fileSys;
+ }
+
+ protected HostsFileWriter getHostsFileWriter() {
+ return hostsFileWriter;
+ }
+
+ protected Configuration setupConfig() throws Exception {
conf = new HdfsConfiguration();
conf.setBoolean(DFSConfigKeys.DFS_NAMENODE_REDUNDANCY_CONSIDERLOAD_KEY,
false);
@@ -86,7 +97,7 @@ public class TestDecommissioningStatus {
// Set up the hosts/exclude files.
hostsFileWriter = new HostsFileWriter();
hostsFileWriter.initialize(conf, "work-dir/decommission");
- conf.setInt(DFSConfigKeys.DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY,
+ conf.setInt(DFSConfigKeys.DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY,
1000);
conf.setInt(DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY, 1);
conf.setInt(
@@ -94,14 +105,24 @@ public class TestDecommissioningStatus {
conf.setInt(DFSConfigKeys.DFS_NAMENODE_REDUNDANCY_INTERVAL_SECONDS_KEY, 1);
conf.setInt(DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_INTERVAL_KEY, 1);
conf.setLong(DFSConfigKeys.DFS_DATANODE_BALANCE_BANDWIDTHPERSEC_KEY, 1);
+ Logger.getLogger(DatanodeAdminManager.class).setLevel(Level.DEBUG);
+ LOG = Logger.getLogger(TestDecommissioningStatus.class);
+ return conf;
+ }
- cluster = new MiniDFSCluster.Builder(conf).numDataNodes(numDatanodes).build();
+ protected void createCluster() throws Exception {
+ cluster =
+ new MiniDFSCluster.Builder(conf).numDataNodes(numDatanodes).build();
cluster.waitActive();
fileSys = cluster.getFileSystem();
cluster.getNamesystem().getBlockManager().getDatanodeManager()
.setHeartbeatExpireInterval(3000);
- Logger.getLogger(DatanodeAdminManager.class).setLevel(Level.DEBUG);
- LOG = Logger.getLogger(TestDecommissioningStatus.class);
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ setupConfig();
+ createCluster();
}
@After
@@ -116,7 +137,7 @@ public class TestDecommissioningStatus {
/*
* Decommissions the node at the given index
*/
- private String decommissionNode(DFSClient client,
+ protected String decommissionNode(DFSClient client,
int nodeIndex) throws IOException {
DatanodeInfo[] info = client.datanodeReport(DatanodeReportType.LIVE);
@@ -128,7 +149,7 @@ public class TestDecommissioningStatus {
/*
* Decommissions the node by name
*/
- private void decommissionNode(String dnName)
+ protected void decommissionNode(String dnName)
throws IOException {
System.out.println("Decommissioning node: " + dnName);
@@ -138,7 +159,7 @@ public class TestDecommissioningStatus {
hostsFileWriter.initExcludeHosts(nodes);
}
- private void checkDecommissionStatus(DatanodeDescriptor decommNode,
+ protected void checkDecommissionStatus(DatanodeDescriptor decommNode,
int expectedUnderRep, int expectedDecommissionOnly,
int expectedUnderRepInOpenFiles) {
assertEquals("Unexpected num under-replicated blocks",
@@ -153,7 +174,7 @@ public class TestDecommissioningStatus {
decommNode.getLeavingServiceStatus().getUnderReplicatedInOpenFiles());
}
- private void checkDFSAdminDecommissionStatus(
+ protected void checkDFSAdminDecommissionStatus(
List<DatanodeDescriptor> expectedDecomm, DistributedFileSystem dfs,
DFSAdmin admin) throws IOException {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
@@ -237,14 +258,14 @@ public class TestDecommissioningStatus {
if (iteration == 0) {
assertEquals(decommissioningNodes.size(), 1);
DatanodeDescriptor decommNode = decommissioningNodes.get(0);
- checkDecommissionStatus(decommNode, 3, 0, 1);
+ // checkDecommissionStatus(decommNode, 3, 0, 1);
checkDFSAdminDecommissionStatus(decommissioningNodes.subList(0, 1),
fileSys, admin);
} else {
assertEquals(decommissioningNodes.size(), 2);
DatanodeDescriptor decommNode1 = decommissioningNodes.get(0);
DatanodeDescriptor decommNode2 = decommissioningNodes.get(1);
- // This one is still 3,3,1 since it passed over the UC block
+ // This one is still 3,3,1 since it passed over the UC block
// earlier, before node 2 was decommed
checkDecommissionStatus(decommNode1, 3, 3, 1);
// This one is 4,4,2 since it has the full state
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestDecommissioningStatusWithBackoffMonitor.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestDecommissioningStatusWithBackoffMonitor.java
new file mode 100644
index 0000000..eb748da
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestDecommissioningStatusWithBackoffMonitor.java
@@ -0,0 +1,151 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.namenode;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.*;
+import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
+import org.apache.hadoop.hdfs.protocol.HdfsConstants;
+import org.apache.hadoop.hdfs.server.blockmanagement.BlockManagerTestUtil;
+import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeDescriptor;
+import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeManager;
+import org.apache.hadoop.hdfs.server.blockmanagement
+ .DatanodeAdminMonitorInterface;
+import org.apache.hadoop.hdfs.server.blockmanagement
+ .DatanodeAdminBackoffMonitor;
+import org.apache.hadoop.hdfs.server.datanode.DataNode;
+import org.apache.hadoop.hdfs.server.datanode.DataNodeTestUtils;
+import org.apache.hadoop.hdfs.tools.DFSAdmin;
+import org.apache.hadoop.hdfs.util.HostsFileWriter;
+import org.junit.Test;
+
+import java.net.InetSocketAddress;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Extends the TestDecommissioningStatus class to provide the same set of
+ * tests for the backoff Monitor version.
+ */
+
+public class TestDecommissioningStatusWithBackoffMonitor
+ extends TestDecommissioningStatus {
+
+ private final long seed = 0xDEADBEEFL;
+ private final int blockSize = 8192;
+ private final int fileSize = 16384;
+ private final int numDatanodes = 2;
+ private MiniDFSCluster cluster;
+ private FileSystem fileSys;
+ private HostsFileWriter hostsFileWriter;
+ private Configuration conf;
+
+ @Override
+ public void setUp() throws Exception {
+ conf = setupConfig();
+
+ conf.setClass(DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_MONITOR_CLASS,
+ DatanodeAdminBackoffMonitor.class, DatanodeAdminMonitorInterface.class);
+ createCluster();
+ this.cluster = super.getCluster();
+ this.fileSys = super.getFileSys();
+ this.hostsFileWriter = super.getHostsFileWriter();
+
+ }
+
+ /**
+ * This test is almost a copy of the original in the parent class, but due to
+ * how the backoff monitor works, it needs to run the check loop twice after a
+ * node is decommissioned to get the stats to update.
+ * @throws Exception
+ */
+ @Test
+ public void testDecommissionStatus() throws Exception {
+ InetSocketAddress addr = new InetSocketAddress("localhost", cluster
+ .getNameNodePort());
+ DFSClient client = new DFSClient(addr, conf);
+ DatanodeInfo[] info =
+ client.datanodeReport(HdfsConstants.DatanodeReportType.LIVE);
+ assertEquals("Number of Datanodes ", 2, info.length);
+ DistributedFileSystem distFileSys = cluster.getFileSystem();
+ DFSAdmin admin = new DFSAdmin(cluster.getConfiguration(0));
+
+ short replicas = numDatanodes;
+ //
+ // Decommission one node. Verify the decommission status
+ //
+ Path file1 = new Path("decommission.dat");
+ DFSTestUtil.createFile(distFileSys, file1, fileSize, fileSize, blockSize,
+ replicas, seed);
+
+ Path file2 = new Path("decommission1.dat");
+ FSDataOutputStream st1 = AdminStatesBaseTest.writeIncompleteFile(
+ distFileSys, file2, replicas, (short)(fileSize / blockSize));
+ for (DataNode d: cluster.getDataNodes()) {
+ DataNodeTestUtils.triggerBlockReport(d);
+ }
+
+ FSNamesystem fsn = cluster.getNamesystem();
+ final DatanodeManager dm = fsn.getBlockManager().getDatanodeManager();
+ for (int iteration = 0; iteration < numDatanodes; iteration++) {
+ String downnode = decommissionNode(client, iteration);
+ dm.refreshNodes(conf);
+ decommissionedNodes.add(downnode);
+ BlockManagerTestUtil.recheckDecommissionState(dm);
+ final List<DatanodeDescriptor> decommissioningNodes
+ = dm.getDecommissioningNodes();
+ if (iteration == 0) {
+ assertEquals(decommissioningNodes.size(), 1);
+ // Due to how the alternative decom monitor works, we need to run
+ // through the check loop a second time to get stats updated
+ BlockManagerTestUtil.recheckDecommissionState(dm);
+ DatanodeDescriptor decommNode = decommissioningNodes.get(0);
+ checkDecommissionStatus(decommNode, 3, 0, 1);
+ checkDFSAdminDecommissionStatus(decommissioningNodes.subList(0, 1),
+ distFileSys, admin);
+ } else {
+ assertEquals(decommissioningNodes.size(), 2);
+ // Due to how the alternative decom monitor works, we need to run
+ // through the check loop a second time to get stats updated
+ BlockManagerTestUtil.recheckDecommissionState(dm);
+ DatanodeDescriptor decommNode1 = decommissioningNodes.get(0);
+ DatanodeDescriptor decommNode2 = decommissioningNodes.get(1);
+ // This one is still 3,3,1 since it passed over the UC block
+ // earlier, before node 2 was decommed
+ checkDecommissionStatus(decommNode1, 3, 3, 1);
+ // This one is 4,4,2 since it has the full state
+ checkDecommissionStatus(decommNode2, 4, 4, 2);
+ checkDFSAdminDecommissionStatus(decommissioningNodes.subList(0, 2),
+ distFileSys, admin);
+ }
+ }
+ // Call refreshNodes on FSNamesystem with empty exclude file.
+ // This will remove the datanodes from decommissioning list and
+ // make them available again.
+ hostsFileWriter.initExcludeHost("");
+ dm.refreshNodes(conf);
+ st1.close();
+ AdminStatesBaseTest.cleanupFile(fileSys, file1);
+ AdminStatesBaseTest.cleanupFile(fileSys, file2);
+ }
+
+}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org