HDFS-9388. Decommission related code to support Maintenance State for datanodes.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/065a906b Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/065a906b Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/065a906b Branch: refs/heads/branch-2 Commit: 065a906be7a0812540bb276a8e37eaab3ffcf101 Parents: 45cde76 Author: Manoj Govindassamy <[email protected]> Authored: Wed Aug 2 14:22:41 2017 -0700 Committer: Manoj Govindassamy <[email protected]> Committed: Wed Aug 2 16:58:54 2017 -0700 ---------------------------------------------------------------------- .../blockmanagement/DatanodeAdminManager.java | 782 +++++++++++++++++++ .../server/blockmanagement/DatanodeManager.java | 32 +- .../blockmanagement/DecommissionManager.java | 766 ------------------ .../hadoop/hdfs/server/namenode/BackupNode.java | 6 +- .../src/main/resources/hdfs-default.xml | 21 +- .../apache/hadoop/hdfs/TestDecommission.java | 49 +- .../blockmanagement/BlockManagerTestUtil.java | 2 +- .../TestReplicationPolicyConsiderLoad.java | 2 +- .../namenode/TestDecommissioningStatus.java | 6 +- .../TestDefaultBlockPlacementPolicy.java | 4 +- .../hadoop/hdfs/server/namenode/TestFsck.java | 125 ++- .../server/namenode/TestNameNodeMXBean.java | 104 ++- .../namenode/TestNamenodeCapacityReport.java | 8 +- 13 files changed, 1076 insertions(+), 831 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hadoop/blob/065a906b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeAdminManager.java ---------------------------------------------------------------------- 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 new file mode 100644 index 0000000..a74b1c2 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeAdminManager.java @@ -0,0 +1,782 @@ +/** + * 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 static com.google.common.base.Preconditions.checkArgument; +import static org.apache.hadoop.util.Time.monotonicNow; + +import java.util.AbstractList; +import java.util.Iterator; +import java.util.LinkedList; +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.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.util.ChunkedArrayList; +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; + +/** + * Manages decommissioning and maintenance state for DataNodes. A background + * monitor thread periodically checks the status of DataNodes that are + * decommissioning or entering maintenance state. + * <p/> + * A DataNode can be decommissioned in a few situations: + * <ul> + * <li>If a DN is dead, it is decommissioned immediately.</li> + * <li>If a DN is alive, it is decommissioned after all of its blocks + * are sufficiently replicated. Merely under-replicated blocks do not + * block decommissioning as long as they are above a replication + * threshold.</li> + * </ul> + * In the second case, the DataNode transitions to a DECOMMISSION_INPROGRESS + * state and is tracked by the monitor thread. The monitor periodically scans + * through the list of insufficiently replicated blocks on these DataNodes to + * determine if they can be DECOMMISSIONED. The monitor also prunes this list + * as blocks become replicated, so monitor scans will become more efficient + * over time. + * <p/> + * DECOMMISSION_INPROGRESS nodes that become dead do not progress to + * DECOMMISSIONED until they become live again. This prevents potential + * durability loss for singly-replicated blocks (see HDFS-6791). + * <p/> + * DataNodes can also be put under maintenance state for any short duration + * maintenance operations. Unlike decommissioning, blocks are not always + * re-replicated for the DataNodes to enter maintenance state. When the + * blocks are replicated at least dfs.namenode.maintenance.replication.min, + * DataNodes transition to IN_MAINTENANCE state. Otherwise, just like + * decommissioning, DataNodes transition to ENTERING_MAINTENANCE state and + * wait for the blocks to be sufficiently replicated and then transition to + * IN_MAINTENANCE state. The block replication factor is relaxed for a maximum + * of maintenance expiry time. When DataNodes don't transition or join the + * cluster back by expiry time, blocks are re-replicated just as in + * decommissioning case as to avoid read or write performance degradation. + * <p/> + * This class depends on the FSNamesystem lock for synchronization. + */ [email protected] +public class DatanodeAdminManager { + private static final Logger LOG = + LoggerFactory.getLogger(DatanodeAdminManager.class); + private final Namesystem namesystem; + private final BlockManager blockManager; + 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; + + DatanodeAdminManager(final Namesystem namesystem, + final BlockManager blockManager, final HeartbeatManager hbManager) { + this.namesystem = namesystem; + this.blockManager = blockManager; + this.hbManager = hbManager; + + executor = Executors.newScheduledThreadPool(1, + new ThreadFactoryBuilder().setNameFormat("DatanodeAdminMonitor-%d") + .setDaemon(true).build()); + outOfServiceNodeBlocks = new TreeMap<>(); + pendingNodes = new LinkedList<>(); + } + + /** + * Start the DataNode admin monitor thread. + * @param conf + */ + void activate(Configuration conf) { + final int intervalSecs = + conf.getInt(DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_INTERVAL_KEY, + DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_INTERVAL_DEFAULT); + checkArgument(intervalSecs >= 0, "Cannot set a negative " + + "value for " + DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_INTERVAL_KEY); + + // By default, the new configuration key overrides the deprecated one. + // No # node limit is set. + int blocksPerInterval = conf.getInt( + DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_BLOCKS_PER_INTERVAL_KEY, + DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_BLOCKS_PER_INTERVAL_DEFAULT); + int nodesPerInterval = Integer.MAX_VALUE; + + // If the expected key isn't present and the deprecated one is, + // use the deprecated one into the new one. This overrides the + // default. + // + // Also print a deprecation warning. + final String deprecatedKey = + "dfs.namenode.decommission.nodes.per.interval"; + final String strNodes = conf.get(deprecatedKey); + if (strNodes != null) { + nodesPerInterval = Integer.parseInt(strNodes); + blocksPerInterval = Integer.MAX_VALUE; + LOG.warn("Deprecated configuration key {} will be ignored.", + deprecatedKey); + LOG.warn("Please update your configuration to use {} instead.", + DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_BLOCKS_PER_INTERVAL_KEY); + } + checkArgument(blocksPerInterval > 0, + "Must set a positive value for " + + DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_BLOCKS_PER_INTERVAL_KEY); + + final int maxConcurrentTrackedNodes = conf.getInt( + DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_MAX_CONCURRENT_TRACKED_NODES, + DFSConfigKeys + .DFS_NAMENODE_DECOMMISSION_MAX_CONCURRENT_TRACKED_NODES_DEFAULT); + checkArgument(maxConcurrentTrackedNodes >= 0, "Cannot set a negative " + + "value for " + + DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_MAX_CONCURRENT_TRACKED_NODES); + + monitor = new Monitor(blocksPerInterval, + nodesPerInterval, maxConcurrentTrackedNodes); + executor.scheduleAtFixedRate(monitor, intervalSecs, intervalSecs, + TimeUnit.SECONDS); + + LOG.debug("Activating DatanodeAdminManager with interval {} seconds, " + + "{} max blocks per interval, " + + "{} max concurrently tracked nodes.", intervalSecs, + blocksPerInterval, nodesPerInterval, maxConcurrentTrackedNodes); + } + + /** + * Stop the admin monitor thread, waiting briefly for it to terminate. + */ + void close() { + executor.shutdownNow(); + try { + executor.awaitTermination(3000, TimeUnit.MILLISECONDS); + } catch (InterruptedException e) {} + } + + /** + * Start decommissioning the specified datanode. + * @param node + */ + @VisibleForTesting + public void startDecommission(DatanodeDescriptor node) { + if (!node.isDecommissionInProgress() && !node.isDecommissioned()) { + // Update DN stats maintained by HeartbeatManager + hbManager.startDecommission(node); + // hbManager.startDecommission will set dead node to decommissioned. + if (node.isDecommissionInProgress()) { + for (DatanodeStorageInfo storage : node.getStorageInfos()) { + LOG.info("Starting decommission of {} {} with {} blocks", + node, storage, storage.numBlocks()); + } + node.getLeavingServiceStatus().setStartTime(monotonicNow()); + pendingNodes.add(node); + } + } else { + LOG.trace("startDecommission: Node {} in {}, nothing to do." + + node, node.getAdminState()); + } + } + + /** + * Stop decommissioning the specified datanode. + * @param node + */ + @VisibleForTesting + public void stopDecommission(DatanodeDescriptor node) { + if (node.isDecommissionInProgress() || node.isDecommissioned()) { + // Update DN stats maintained by HeartbeatManager + hbManager.stopDecommission(node); + // Over-replicated blocks will be detected and processed when + // the dead node comes back and send in its full block report. + if (node.isAlive()) { + blockManager.processExtraRedundancyBlocksOnInService(node); + } + // Remove from tracking in DatanodeAdminManager + pendingNodes.remove(node); + outOfServiceNodeBlocks.remove(node); + } else { + LOG.trace("stopDecommission: Node {} in {}, nothing to do." + + node, node.getAdminState()); + } + } + + /** + * Start maintenance of the specified datanode. + * @param node + */ + @VisibleForTesting + public void startMaintenance(DatanodeDescriptor node, + long maintenanceExpireTimeInMS) { + // Even if the node is already in maintenance, we still need to adjust + // the expiration time. + node.setMaintenanceExpireTimeInMS(maintenanceExpireTimeInMS); + if (!node.isMaintenance()) { + // Update DN stats maintained by HeartbeatManager + hbManager.startMaintenance(node); + // hbManager.startMaintenance will set dead node to IN_MAINTENANCE. + if (node.isEnteringMaintenance()) { + for (DatanodeStorageInfo storage : node.getStorageInfos()) { + LOG.info("Starting maintenance of {} {} with {} blocks", + node, storage, storage.numBlocks()); + } + node.getLeavingServiceStatus().setStartTime(monotonicNow()); + } + // Track the node regardless whether it is ENTERING_MAINTENANCE or + // IN_MAINTENANCE to support maintenance expiration. + pendingNodes.add(node); + } else { + LOG.trace("startMaintenance: Node {} in {}, nothing to do." + + node, node.getAdminState()); + } + } + + + /** + * Stop maintenance of the specified datanode. + * @param node + */ + @VisibleForTesting + public void stopMaintenance(DatanodeDescriptor node) { + if (node.isMaintenance()) { + // Update DN stats maintained by HeartbeatManager + hbManager.stopMaintenance(node); + + // extra redundancy blocks will be detected and processed when + // the dead node comes back and send in its full block report. + if (!node.isAlive()) { + // The node became dead when it was in maintenance, at which point + // the replicas weren't removed from block maps. + // When the node leaves maintenance, the replicas should be removed + // from the block maps to trigger the necessary replication to + // maintain the safety property of "# of live replicas + maintenance + // replicas" >= the expected redundancy. + blockManager.removeBlocksAssociatedTo(node); + } else { + // Even though putting nodes in maintenance node doesn't cause live + // replicas to match expected replication factor, it is still possible + // to have over replicated when the node leaves maintenance node. + // First scenario: + // a. Node became dead when it is at AdminStates.NORMAL, thus + // block is replicated so that 3 replicas exist on other nodes. + // b. Admins put the dead node into maintenance mode and then + // have the node rejoin the cluster. + // c. Take the node out of maintenance mode. + // Second scenario: + // a. With replication factor 3, set one replica to maintenance node, + // thus block has 1 maintenance replica and 2 live replicas. + // b. Change the replication factor to 2. The block will still have + // 1 maintenance replica and 2 live replicas. + // c. Take the node out of maintenance mode. + blockManager.processExtraRedundancyBlocksOnInService(node); + } + + // Remove from tracking in DatanodeAdminManager + pendingNodes.remove(node); + outOfServiceNodeBlocks.remove(node); + } else { + LOG.trace("stopMaintenance: Node {} in {}, nothing to do." + + node, node.getAdminState()); + } + } + + private void setDecommissioned(DatanodeDescriptor dn) { + dn.setDecommissioned(); + LOG.info("Decommissioning complete for node {}", dn); + } + + private void setInMaintenance(DatanodeDescriptor dn) { + dn.setInMaintenance(); + LOG.info("Node {} has entered maintenance mode.", dn); + } + + /** + * Checks whether a block is sufficiently replicated/stored for + * DECOMMISSION_INPROGRESS or ENTERING_MAINTENANCE datanodes. For replicated + * blocks or striped blocks, full-strength replication or storage is not + * always necessary, hence "sufficient". + * @return true if sufficient, else false. + */ + private boolean isSufficientlyReplicated(BlockInfo block, BlockCollection bc, + NumberReplicas numberReplicas, boolean isDecommission) { + if (blockManager.hasEnoughEffectiveReplicas(block, numberReplicas, 0)) { + // Block has enough replica, skip + LOG.trace("Block {} does not need replication.", block); + return true; + } + + final int numExpected = blockManager.getExpectedLiveRedundancyNum(block, + numberReplicas); + final int numLive = numberReplicas.liveReplicas(); + + // Block is under-replicated + LOG.trace("Block {} numExpected={}, numLive={}", block, numExpected, + numLive); + if (isDecommission && numExpected > numLive) { + if (bc.isUnderConstruction() && block.equals(bc.getLastBlock())) { + // Can decom a UC block as long as there will still be minReplicas + if (numLive >= blockManager.minReplication) { + LOG.trace("UC block {} sufficiently-replicated since numLive ({}) " + + ">= minR ({})", block, numLive, blockManager.minReplication); + return true; + } else { + LOG.trace("UC block {} insufficiently-replicated since numLive " + + "({}) < minR ({})", block, numLive, + blockManager.minReplication); + } + } else { + // Can decom a non-UC as long as the default replication is met + if (numLive >= blockManager.defaultReplication) { + return true; + } + } + } + return false; + } + + private static void logBlockReplicationInfo(BlockInfo block, + BlockCollection bc, + DatanodeDescriptor srcNode, NumberReplicas num, + Iterable<DatanodeStorageInfo> storages) { + if (!NameNode.blockStateChangeLog.isInfoEnabled()) { + return; + } + + int curReplicas = num.liveReplicas(); + int curExpectedReplicas = block.getReplication(); + StringBuilder nodeList = new StringBuilder(); + for (DatanodeStorageInfo storage : storages) { + final DatanodeDescriptor node = storage.getDatanodeDescriptor(); + nodeList.append(node); + nodeList.append(" "); + } + NameNode.blockStateChangeLog.info( + "Block: " + block + ", Expected Replicas: " + + curExpectedReplicas + ", live replicas: " + curReplicas + + ", corrupt replicas: " + num.corruptReplicas() + + ", decommissioned replicas: " + num.decommissioned() + + ", decommissioning replicas: " + num.decommissioning() + + ", maintenance replicas: " + num.maintenanceReplicas() + + ", live entering maintenance replicas: " + + num.liveEnteringMaintenanceReplicas() + + ", excess replicas: " + num.excessReplicas() + + ", Is Open File: " + bc.isUnderConstruction() + + ", Datanodes having this block: " + nodeList + ", Current Datanode: " + + srcNode + ", Is current datanode decommissioning: " + + srcNode.isDecommissionInProgress() + + ", Is current datanode entering maintenance: " + + srcNode.isEnteringMaintenance()); + } + + @VisibleForTesting + public int getNumPendingNodes() { + return pendingNodes.size(); + } + + @VisibleForTesting + public int getNumTrackedNodes() { + return outOfServiceNodeBlocks.size(); + } + + @VisibleForTesting + public int getNumNodesChecked() { + return monitor.numNodesChecked; + } + + /** + * 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 check per tick. + */ + private final int numNodesPerCheck; + /** + * The maximum number of nodes to track in decomNodeBlocks. 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 + * testing. + */ + 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 numNodesPerCheck, int + maxConcurrentTrackedNodes) { + this.numBlocksPerCheck = numBlocksPerCheck; + this.numNodesPerCheck = numNodesPerCheck; + this.maxConcurrentTrackedNodes = maxConcurrentTrackedNodes; + } + + private boolean exceededNumBlocksPerCheck() { + LOG.trace("Processed {} blocks so far this tick", numBlocksChecked); + return numBlocksChecked >= numBlocksPerCheck; + } + + @Deprecated + private boolean exceededNumNodesPerCheck() { + LOG.trace("Processed {} nodes so far this tick", numNodesChecked); + return numNodesChecked >= numNodesPerCheck; + } + + @Override + public void run() { + 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(); + } finally { + namesystem.writeUnlock(); + } + if (numBlocksChecked + numNodesChecked > 0) { + LOG.info("Checked {} blocks and {} nodes this tick", numBlocksChecked, + numNodesChecked); + } + } + + /** + * 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 LinkedList<DatanodeDescriptor> toRemove = new LinkedList<>(); + + while (it.hasNext() + && !exceededNumBlocksPerCheck() + && !exceededNumNodesPerCheck() + && namesystem.isRunning()) { + numNodesChecked++; + final Map.Entry<DatanodeDescriptor, AbstractList<BlockInfo>> + entry = it.next(); + final DatanodeDescriptor dn = entry.getKey(); + 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 = handleInsufficientlyReplicated(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); + pruneSufficientlyReplicated(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 = handleInsufficientlyReplicated(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, + "A node is in an invalid state!"); + } + LOG.debug("Node {} is sufficiently replicated and healthy, " + + "marked as {}.", dn.getAdminState()); + } else { + LOG.debug("Node {} {} healthy." + + " It needs to replicate {} more blocks." + + " {} is still in progress.", dn, + isHealthy? "is": "isn't", blocks.size(), dn.getAdminState()); + } + } else { + LOG.debug("Node {} still has {} blocks to replicate " + + "before it is a candidate to finish {}.", + dn, blocks.size(), dn.getAdminState()); + } + 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 a node that is not yet decommissioned or in service!"); + outOfServiceNodeBlocks.remove(dn); + } + } + + /** + * Removes sufficiently replicated blocks from the block list of a + * datanode. + */ + private void pruneSufficientlyReplicated(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 work for + * any under-replicated blocks. + * + * @param datanode + * @return List of insufficiently replicated blocks + */ + private AbstractList<BlockInfo> handleInsufficientlyReplicated( + 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 insufficientlyReplicated Return parameter. If it's not null, + * will contain the insufficiently + * replicated-blocks from the list. + * @param pruneSufficientlyReplicated whether to remove sufficiently + * replicated blocks from the iterator + * @return true if there are under-replicated blocks in the provided block + * iterator, else false. + */ + private void processBlocksInternal( + final DatanodeDescriptor datanode, + final Iterator<BlockInfo> it, + final List<BlockInfo> insufficientlyReplicated, + boolean pruneSufficientlyReplicated) { + boolean firstReplicationLog = true; + int underReplicatedBlocks = 0; + // All maintenance and decommission replicas. + int outOfServiceOnlyReplicas = 0; + // Low redundancy in UC Blocks only + int underReplicatedInOpenFiles = 0; + while (it.hasNext()) { + if (insufficientlyReplicated == 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(); + final int curReplicas = liveReplicas; + + // Schedule under-replicated blocks for replication if not already + // pending + boolean isDecommission = datanode.isDecommissionInProgress(); + boolean neededReplication = isDecommission ? + blockManager.isNeededReplication(block, num) : + blockManager.isNeededReplicationForMaintenance(block, num); + if (neededReplication) { + if (!blockManager.neededReplications.contains(block) && + blockManager.pendingReplications.getNumReplicas(block) == 0 && + blockManager.isPopulatingReplQueues()) { + // Process these blocks only when active NN is out of safe mode. + blockManager.neededReplications.add(block, + liveReplicas, num.readOnlyReplicas(), + num.outOfServiceReplicas(), + block.getReplication()); + } + } + + // Even if the block is under-replicated, + // it doesn't block decommission if it's sufficiently replicated + if (isSufficientlyReplicated(block, bc, num, isDecommission)) { + if (pruneSufficientlyReplicated) { + it.remove(); + } + continue; + } + + // We've found an insufficiently replicated block. + if (insufficientlyReplicated != null) { + insufficientlyReplicated.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 + underReplicatedBlocks++; + if (bc.isUnderConstruction()) { + underReplicatedInOpenFiles++; + } + if ((curReplicas == 0) && (num.outOfServiceReplicas() > 0)) { + outOfServiceOnlyReplicas++; + } + } + + datanode.getLeavingServiceStatus().set(underReplicatedBlocks, + outOfServiceOnlyReplicas, underReplicatedInOpenFiles); + } + } + + @VisibleForTesting + void runMonitorForTest() throws ExecutionException, InterruptedException { + executor.submit(monitor).get(); + } +} http://git-wip-us.apache.org/repos/asf/hadoop/blob/065a906b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeManager.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeManager.java index 7e82938..e4cbb8f 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeManager.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeManager.java @@ -68,7 +68,7 @@ public class DatanodeManager { private final Namesystem namesystem; private final BlockManager blockManager; - private final DecommissionManager decomManager; + private final DatanodeAdminManager datanodeAdminManager; private final HeartbeatManager heartbeatManager; private final FSClusterStats fsClusterStats; @@ -199,10 +199,11 @@ public class DatanodeManager { final Configuration conf) throws IOException { this.namesystem = namesystem; this.blockManager = blockManager; - - this.heartbeatManager = new HeartbeatManager(namesystem, blockManager, conf); - this.decomManager = new DecommissionManager(namesystem, blockManager, - heartbeatManager); + + this.heartbeatManager = new HeartbeatManager(namesystem, + blockManager, conf); + this.datanodeAdminManager = new DatanodeAdminManager(namesystem, + blockManager, heartbeatManager); this.fsClusterStats = newFSClusterStats(); this.dataNodePeerStatsEnabled = conf.getBoolean( DFSConfigKeys.DFS_DATANODE_PEER_STATS_ENABLED_KEY, @@ -351,12 +352,12 @@ public class DatanodeManager { } void activate(final Configuration conf) { - decomManager.activate(conf); + datanodeAdminManager.activate(conf); heartbeatManager.activate(); } void close() { - decomManager.close(); + datanodeAdminManager.close(); heartbeatManager.close(); } @@ -371,8 +372,8 @@ public class DatanodeManager { } @VisibleForTesting - public DecommissionManager getDecomManager() { - return decomManager; + public DatanodeAdminManager getDatanodeAdminManager() { + return datanodeAdminManager; } public HostConfigManager getHostConfigManager() { @@ -904,9 +905,9 @@ public class DatanodeManager { hostConfigManager.getMaintenanceExpirationTimeInMS(nodeReg); // If the registered node is in exclude list, then decommission it if (getHostConfigManager().isExcluded(nodeReg)) { - decomManager.startDecommission(nodeReg); + datanodeAdminManager.startDecommission(nodeReg); } else if (nodeReg.maintenanceNotExpired(maintenanceExpireTimeInMS)) { - decomManager.startMaintenance(nodeReg, maintenanceExpireTimeInMS); + datanodeAdminManager.startMaintenance(nodeReg, maintenanceExpireTimeInMS); } } @@ -1132,12 +1133,13 @@ public class DatanodeManager { long maintenanceExpireTimeInMS = hostConfigManager.getMaintenanceExpirationTimeInMS(node); if (node.maintenanceNotExpired(maintenanceExpireTimeInMS)) { - decomManager.startMaintenance(node, maintenanceExpireTimeInMS); + datanodeAdminManager.startMaintenance( + node, maintenanceExpireTimeInMS); } else if (hostConfigManager.isExcluded(node)) { - decomManager.startDecommission(node); + datanodeAdminManager.startDecommission(node); } else { - decomManager.stopMaintenance(node); - decomManager.stopDecommission(node); + datanodeAdminManager.stopMaintenance(node); + datanodeAdminManager.stopDecommission(node); } } node.setUpgradeDomain(hostConfigManager.getUpgradeDomain(node)); http://git-wip-us.apache.org/repos/asf/hadoop/blob/065a906b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DecommissionManager.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DecommissionManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DecommissionManager.java deleted file mode 100644 index cea2ee3..0000000 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DecommissionManager.java +++ /dev/null @@ -1,766 +0,0 @@ -/** - * 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 static com.google.common.base.Preconditions.checkArgument; -import static org.apache.hadoop.util.Time.monotonicNow; - -import java.util.AbstractList; -import java.util.Iterator; -import java.util.LinkedList; -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.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.util.ChunkedArrayList; -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; - -/** - * Manages datanode decommissioning. A background monitor thread - * periodically checks the status of datanodes that are in-progress of - * decommissioning. - * <p/> - * A datanode can be decommissioned in a few situations: - * <ul> - * <li>If a DN is dead, it is decommissioned immediately.</li> - * <li>If a DN is alive, it is decommissioned after all of its blocks - * are sufficiently replicated. Merely under-replicated blocks do not - * block decommissioning as long as they are above a replication - * threshold.</li> - * </ul> - * In the second case, the datanode transitions to a - * decommission-in-progress state and is tracked by the monitor thread. The - * monitor periodically scans through the list of insufficiently replicated - * blocks on these datanodes to - * determine if they can be decommissioned. The monitor also prunes this list - * as blocks become replicated, so monitor scans will become more efficient - * over time. - * <p/> - * Decommission-in-progress nodes that become dead do not progress to - * decommissioned until they become live again. This prevents potential - * durability loss for singly-replicated blocks (see HDFS-6791). - * <p/> - * This class depends on the FSNamesystem lock for synchronization. - */ [email protected] -public class DecommissionManager { - private static final Logger LOG = LoggerFactory.getLogger(DecommissionManager - .class); - - private final Namesystem namesystem; - private final BlockManager blockManager; - 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; - - DecommissionManager(final Namesystem namesystem, - final BlockManager blockManager, final HeartbeatManager hbManager) { - this.namesystem = namesystem; - this.blockManager = blockManager; - this.hbManager = hbManager; - - executor = Executors.newScheduledThreadPool(1, - new ThreadFactoryBuilder().setNameFormat("DecommissionMonitor-%d") - .setDaemon(true).build()); - outOfServiceNodeBlocks = new TreeMap<>(); - pendingNodes = new LinkedList<>(); - } - - /** - * Start the decommission monitor thread. - * @param conf - */ - void activate(Configuration conf) { - final int intervalSecs = - conf.getInt(DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_INTERVAL_KEY, - DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_INTERVAL_DEFAULT); - checkArgument(intervalSecs >= 0, "Cannot set a negative " + - "value for " + DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_INTERVAL_KEY); - - // By default, the new configuration key overrides the deprecated one. - // No # node limit is set. - int blocksPerInterval = conf.getInt( - DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_BLOCKS_PER_INTERVAL_KEY, - DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_BLOCKS_PER_INTERVAL_DEFAULT); - int nodesPerInterval = Integer.MAX_VALUE; - - // If the expected key isn't present and the deprecated one is, - // use the deprecated one into the new one. This overrides the - // default. - // - // Also print a deprecation warning. - final String deprecatedKey = - "dfs.namenode.decommission.nodes.per.interval"; - final String strNodes = conf.get(deprecatedKey); - if (strNodes != null) { - nodesPerInterval = Integer.parseInt(strNodes); - blocksPerInterval = Integer.MAX_VALUE; - LOG.warn("Using deprecated configuration key {} value of {}.", - deprecatedKey, nodesPerInterval); - LOG.warn("Please update your configuration to use {} instead.", - DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_BLOCKS_PER_INTERVAL_KEY); - } - checkArgument(blocksPerInterval > 0, - "Must set a positive value for " - + DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_BLOCKS_PER_INTERVAL_KEY); - - final int maxConcurrentTrackedNodes = conf.getInt( - DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_MAX_CONCURRENT_TRACKED_NODES, - DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_MAX_CONCURRENT_TRACKED_NODES_DEFAULT); - checkArgument(maxConcurrentTrackedNodes >= 0, "Cannot set a negative " + - "value for " - + DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_MAX_CONCURRENT_TRACKED_NODES); - - monitor = new Monitor(blocksPerInterval, - nodesPerInterval, maxConcurrentTrackedNodes); - executor.scheduleAtFixedRate(monitor, intervalSecs, intervalSecs, - TimeUnit.SECONDS); - - LOG.debug("Activating DecommissionManager with interval {} seconds, " + - "{} max blocks per interval, {} max nodes per interval, " + - "{} max concurrently tracked nodes.", intervalSecs, - blocksPerInterval, nodesPerInterval, maxConcurrentTrackedNodes); - } - - /** - * Stop the decommission monitor thread, waiting briefly for it to terminate. - */ - void close() { - executor.shutdownNow(); - try { - executor.awaitTermination(3000, TimeUnit.MILLISECONDS); - } catch (InterruptedException e) {} - } - - /** - * Start decommissioning the specified datanode. - * @param node - */ - @VisibleForTesting - public void startDecommission(DatanodeDescriptor node) { - if (!node.isDecommissionInProgress() && !node.isDecommissioned()) { - // Update DN stats maintained by HeartbeatManager - hbManager.startDecommission(node); - // hbManager.startDecommission will set dead node to decommissioned. - if (node.isDecommissionInProgress()) { - for (DatanodeStorageInfo storage : node.getStorageInfos()) { - LOG.info("Starting decommission of {} {} with {} blocks", - node, storage, storage.numBlocks()); - } - node.getLeavingServiceStatus().setStartTime(monotonicNow()); - pendingNodes.add(node); - } - } else { - LOG.trace("startDecommission: Node {} in {}, nothing to do." + - node, node.getAdminState()); - } - } - - /** - * Stop decommissioning the specified datanode. - * @param node - */ - @VisibleForTesting - public void stopDecommission(DatanodeDescriptor node) { - if (node.isDecommissionInProgress() || node.isDecommissioned()) { - // Update DN stats maintained by HeartbeatManager - hbManager.stopDecommission(node); - // Over-replicated blocks will be detected and processed when - // the dead node comes back and send in its full block report. - if (node.isAlive()) { - blockManager.processExtraRedundancyBlocksOnInService(node); - } - // Remove from tracking in DecommissionManager - pendingNodes.remove(node); - outOfServiceNodeBlocks.remove(node); - } else { - LOG.trace("stopDecommission: Node {} in {}, nothing to do." + - node, node.getAdminState()); - } - } - - /** - * Start maintenance of the specified datanode. - * @param node - */ - @VisibleForTesting - public void startMaintenance(DatanodeDescriptor node, - long maintenanceExpireTimeInMS) { - // Even if the node is already in maintenance, we still need to adjust - // the expiration time. - node.setMaintenanceExpireTimeInMS(maintenanceExpireTimeInMS); - if (!node.isMaintenance()) { - // Update DN stats maintained by HeartbeatManager - hbManager.startMaintenance(node); - // hbManager.startMaintenance will set dead node to IN_MAINTENANCE. - if (node.isEnteringMaintenance()) { - for (DatanodeStorageInfo storage : node.getStorageInfos()) { - LOG.info("Starting maintenance of {} {} with {} blocks", - node, storage, storage.numBlocks()); - } - node.getLeavingServiceStatus().setStartTime(monotonicNow()); - } - // Track the node regardless whether it is ENTERING_MAINTENANCE or - // IN_MAINTENANCE to support maintenance expiration. - pendingNodes.add(node); - } else { - LOG.trace("startMaintenance: Node {} in {}, nothing to do." + - node, node.getAdminState()); - } - } - - - /** - * Stop maintenance of the specified datanode. - * @param node - */ - @VisibleForTesting - public void stopMaintenance(DatanodeDescriptor node) { - if (node.isMaintenance()) { - // Update DN stats maintained by HeartbeatManager - hbManager.stopMaintenance(node); - - // extra redundancy blocks will be detected and processed when - // the dead node comes back and send in its full block report. - if (!node.isAlive()) { - // The node became dead when it was in maintenance, at which point - // the replicas weren't removed from block maps. - // When the node leaves maintenance, the replicas should be removed - // from the block maps to trigger the necessary replication to - // maintain the safety property of "# of live replicas + maintenance - // replicas" >= the expected redundancy. - blockManager.removeBlocksAssociatedTo(node); - } else { - // Even though putting nodes in maintenance node doesn't cause live - // replicas to match expected replication factor, it is still possible - // to have over replicated when the node leaves maintenance node. - // First scenario: - // a. Node became dead when it is at AdminStates.NORMAL, thus - // block is replicated so that 3 replicas exist on other nodes. - // b. Admins put the dead node into maintenance mode and then - // have the node rejoin the cluster. - // c. Take the node out of maintenance mode. - // Second scenario: - // a. With replication factor 3, set one replica to maintenance node, - // thus block has 1 maintenance replica and 2 live replicas. - // b. Change the replication factor to 2. The block will still have - // 1 maintenance replica and 2 live replicas. - // c. Take the node out of maintenance mode. - blockManager.processExtraRedundancyBlocksOnInService(node); - } - - // Remove from tracking in DecommissionManager - pendingNodes.remove(node); - outOfServiceNodeBlocks.remove(node); - } else { - LOG.trace("stopMaintenance: Node {} in {}, nothing to do." + - node, node.getAdminState()); - } - } - - private void setDecommissioned(DatanodeDescriptor dn) { - dn.setDecommissioned(); - LOG.info("Decommissioning complete for node {}", dn); - } - - private void setInMaintenance(DatanodeDescriptor dn) { - dn.setInMaintenance(); - LOG.info("Node {} has entered maintenance mode.", dn); - } - - /** - * Checks whether a block is sufficiently replicated for decommissioning. - * Full-strength replication is not always necessary, hence "sufficient". - * @return true if sufficient, else false. - */ - private boolean isSufficientlyReplicated(BlockInfo block, BlockCollection bc, - NumberReplicas numberReplicas, boolean isDecommission) { - if (blockManager.hasEnoughEffectiveReplicas(block, numberReplicas, 0)) { - // Block has enough replica, skip - LOG.trace("Block {} does not need replication.", block); - return true; - } - - final int numExpected = blockManager.getExpectedLiveRedundancyNum(block, - numberReplicas); - final int numLive = numberReplicas.liveReplicas(); - - // Block is under-replicated - LOG.trace("Block {} numExpected={}, numLive={}", block, numExpected, - numLive); - if (isDecommission && numExpected > numLive) { - if (bc.isUnderConstruction() && block.equals(bc.getLastBlock())) { - // Can decom a UC block as long as there will still be minReplicas - if (numLive >= blockManager.minReplication) { - LOG.trace("UC block {} sufficiently-replicated since numLive ({}) " - + ">= minR ({})", block, numLive, blockManager.minReplication); - return true; - } else { - LOG.trace("UC block {} insufficiently-replicated since numLive " - + "({}) < minR ({})", block, numLive, - blockManager.minReplication); - } - } else { - // Can decom a non-UC as long as the default replication is met - if (numLive >= blockManager.defaultReplication) { - return true; - } - } - } - return false; - } - - private static void logBlockReplicationInfo(BlockInfo block, - BlockCollection bc, - DatanodeDescriptor srcNode, NumberReplicas num, - Iterable<DatanodeStorageInfo> storages) { - if (!NameNode.blockStateChangeLog.isInfoEnabled()) { - return; - } - - int curReplicas = num.liveReplicas(); - int curExpectedReplicas = block.getReplication(); - StringBuilder nodeList = new StringBuilder(); - for (DatanodeStorageInfo storage : storages) { - final DatanodeDescriptor node = storage.getDatanodeDescriptor(); - nodeList.append(node); - nodeList.append(" "); - } - NameNode.blockStateChangeLog.info( - "Block: " + block + ", Expected Replicas: " - + curExpectedReplicas + ", live replicas: " + curReplicas - + ", corrupt replicas: " + num.corruptReplicas() - + ", decommissioned replicas: " + num.decommissioned() - + ", decommissioning replicas: " + num.decommissioning() - + ", maintenance replicas: " + num.maintenanceReplicas() - + ", live entering maintenance replicas: " - + num.liveEnteringMaintenanceReplicas() - + ", excess replicas: " + num.excessReplicas() - + ", Is Open File: " + bc.isUnderConstruction() - + ", Datanodes having this block: " + nodeList + ", Current Datanode: " - + srcNode + ", Is current datanode decommissioning: " - + srcNode.isDecommissionInProgress() + - ", Is current datanode entering maintenance: " - + srcNode.isEnteringMaintenance()); - } - - @VisibleForTesting - public int getNumPendingNodes() { - return pendingNodes.size(); - } - - @VisibleForTesting - public int getNumTrackedNodes() { - return outOfServiceNodeBlocks.size(); - } - - @VisibleForTesting - public int getNumNodesChecked() { - return monitor.numNodesChecked; - } - - /** - * Checks to see if DNs have finished decommissioning. - * <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 check per tick. - */ - private final int numNodesPerCheck; - /** - * The maximum number of nodes to track in decomNodeBlocks. 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 - * testing. - */ - 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 numNodesPerCheck, int - maxConcurrentTrackedNodes) { - this.numBlocksPerCheck = numBlocksPerCheck; - this.numNodesPerCheck = numNodesPerCheck; - this.maxConcurrentTrackedNodes = maxConcurrentTrackedNodes; - } - - private boolean exceededNumBlocksPerCheck() { - LOG.trace("Processed {} blocks so far this tick", numBlocksChecked); - return numBlocksChecked >= numBlocksPerCheck; - } - - @Deprecated - private boolean exceededNumNodesPerCheck() { - LOG.trace("Processed {} nodes so far this tick", numNodesChecked); - return numNodesChecked >= numNodesPerCheck; - } - - @Override - public void run() { - if (!namesystem.isRunning()) { - LOG.info("Namesystem is not running, skipping decommissioning 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(); - } finally { - namesystem.writeUnlock(); - } - if (numBlocksChecked + numNodesChecked > 0) { - LOG.info("Checked {} blocks and {} nodes this tick", numBlocksChecked, - numNodesChecked); - } - } - - /** - * 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 LinkedList<DatanodeDescriptor> toRemove = new LinkedList<>(); - - while (it.hasNext() - && !exceededNumBlocksPerCheck() - && !exceededNumNodesPerCheck() - && namesystem.isRunning()) { - numNodesChecked++; - final Map.Entry<DatanodeDescriptor, AbstractList<BlockInfo>> - entry = it.next(); - final DatanodeDescriptor dn = entry.getKey(); - 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 = handleInsufficientlyReplicated(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 be decommed - LOG.debug("Processing {} node {}", dn.getAdminState(), dn); - pruneSufficientlyReplicated(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 - LOG.debug("Node {} has finished replicating current set of " - + "blocks, checking with the full block map.", dn); - blocks = handleInsufficientlyReplicated(dn); - outOfServiceNodeBlocks.put(dn, blocks); - } - // If the full scan is clean AND the node liveness is okay, - // we can finally mark as decommissioned. - 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, - "A node is in an invalid state!"); - } - LOG.debug("Node {} is sufficiently replicated and healthy, " - + "marked as {}.", dn.getAdminState()); - } else { - LOG.debug("Node {} {} healthy." - + " It needs to replicate {} more blocks." - + " {} is still in progress.", dn, - isHealthy? "is": "isn't", blocks.size(), dn.getAdminState()); - } - } else { - LOG.debug("Node {} still has {} blocks to replicate " - + "before it is a candidate to finish {}.", - dn, blocks.size(), dn.getAdminState()); - } - 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 a node that is not yet decommissioned or in service!"); - outOfServiceNodeBlocks.remove(dn); - } - } - - /** - * Removes sufficiently replicated blocks from the block list of a - * datanode. - */ - private void pruneSufficientlyReplicated(final DatanodeDescriptor datanode, - AbstractList<BlockInfo> blocks) { - processBlocksInternal(datanode, blocks.iterator(), null, true); - } - - /** - * Returns a list of blocks on a datanode that are insufficiently - * replicated, i.e. are under-replicated enough to prevent decommission. - * <p/> - * As part of this, it also schedules replication work for - * any under-replicated blocks. - * - * @param datanode - * @return List of insufficiently replicated blocks - */ - private AbstractList<BlockInfo> handleInsufficientlyReplicated( - final DatanodeDescriptor datanode) { - AbstractList<BlockInfo> insufficient = new ChunkedArrayList<>(); - processBlocksInternal(datanode, datanode.getBlockIterator(), - insufficient, false); - return insufficient; - } - - /** - * Used while checking if decommission-in-progress datanodes can be marked - * as decommissioned. Combines shared logic of - * pruneSufficientlyReplicated and handleInsufficientlyReplicated. - * - * @param datanode Datanode - * @param it Iterator over the blocks on the - * datanode - * @param insufficientlyReplicated Return parameter. If it's not null, - * will contain the insufficiently - * replicated-blocks from the list. - * @param pruneSufficientlyReplicated whether to remove sufficiently - * replicated blocks from the iterator - * @return true if there are under-replicated blocks in the provided block - * iterator, else false. - */ - private void processBlocksInternal( - final DatanodeDescriptor datanode, - final Iterator<BlockInfo> it, - final List<BlockInfo> insufficientlyReplicated, - boolean pruneSufficientlyReplicated) { - boolean firstReplicationLog = true; - int underReplicatedBlocks = 0; - // All maintenance and decommission replicas. - int outOfServiceOnlyReplicas = 0; - // Low redundancy in UC Blocks only - int underReplicatedInOpenFiles = 0; - while (it.hasNext()) { - if (insufficientlyReplicated == 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 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(); - final int curReplicas = liveReplicas; - - // Schedule under-replicated blocks for replication if not already - // pending - boolean isDecommission = datanode.isDecommissionInProgress(); - boolean neededReplication = isDecommission ? - blockManager.isNeededReplication(block, num) : - blockManager.isNeededReplicationForMaintenance(block, num); - if (neededReplication) { - if (!blockManager.neededReplications.contains(block) && - blockManager.pendingReplications.getNumReplicas(block) == 0 && - blockManager.isPopulatingReplQueues()) { - // Process these blocks only when active NN is out of safe mode. - blockManager.neededReplications.add(block, - liveReplicas, num.readOnlyReplicas(), - num.outOfServiceReplicas(), - block.getReplication()); - } - } - - // Even if the block is under-replicated, - // it doesn't block decommission if it's sufficiently replicated - if (isSufficientlyReplicated(block, bc, num, isDecommission)) { - if (pruneSufficientlyReplicated) { - it.remove(); - } - continue; - } - - // We've found an insufficiently replicated block. - if (insufficientlyReplicated != null) { - insufficientlyReplicated.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 - underReplicatedBlocks++; - if (bc.isUnderConstruction()) { - underReplicatedInOpenFiles++; - } - if ((curReplicas == 0) && (num.outOfServiceReplicas() > 0)) { - outOfServiceOnlyReplicas++; - } - } - - datanode.getLeavingServiceStatus().set(underReplicatedBlocks, - outOfServiceOnlyReplicas, underReplicatedInOpenFiles); - } - } - - @VisibleForTesting - void runMonitorForTest() throws ExecutionException, InterruptedException { - executor.submit(monitor).get(); - } -} http://git-wip-us.apache.org/repos/asf/hadoop/blob/065a906b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/BackupNode.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/BackupNode.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/BackupNode.java index 1d582c5..9ebd5fc 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/BackupNode.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/BackupNode.java @@ -463,9 +463,9 @@ public class BackupNode extends NameNode { * {@link LeaseManager.Monitor} protected by SafeMode. * {@link BlockManager.ReplicationMonitor} protected by SafeMode. * {@link HeartbeatManager.Monitor} protected by SafeMode. - * {@link DecommissionManager.Monitor} need to prohibit refreshNodes(). - * {@link PendingReplicationBlocks.PendingReplicationMonitor} harmless, - * because ReplicationMonitor is muted. + * {@link DatanodeAdminManager.Monitor} need to prohibit refreshNodes(). + * {@link PendingReconstructionBlocks.PendingReconstructionMonitor} + * harmless, because RedundancyMonitor is muted. */ @Override public void startActiveServices() throws IOException { http://git-wip-us.apache.org/repos/asf/hadoop/blob/065a906b/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml ---------------------------------------------------------------------- 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 1ae9161..051301e 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml @@ -927,15 +927,17 @@ <property> <name>dfs.namenode.decommission.interval</name> <value>30</value> - <description>Namenode periodicity in seconds to check if decommission is - complete.</description> + <description>Namenode periodicity in seconds to check if + decommission or maintenance is complete. Support multiple time unit + suffix(case insensitive), as described in dfs.heartbeat.interval. + </description> </property> <property> <name>dfs.namenode.decommission.blocks.per.interval</name> <value>500000</value> - <description>The approximate number of blocks to process per - decommission interval, as defined in dfs.namenode.decommission.interval. + <description>The approximate number of blocks to process per decommission + or maintenance interval, as defined in dfs.namenode.decommission.interval. </description> </property> @@ -943,11 +945,12 @@ <name>dfs.namenode.decommission.max.concurrent.tracked.nodes</name> <value>100</value> <description> - The maximum number of decommission-in-progress datanodes nodes that will be - tracked at one time by the namenode. Tracking a decommission-in-progress - datanode consumes additional NN memory proportional to the number of blocks - on the datnode. Having a conservative limit reduces the potential impact - of decomissioning a large number of nodes at once. + The maximum number of decommission-in-progress or + entering-maintenance datanodes nodes that will be tracked at one time by + the namenode. Tracking these datanode consumes additional NN memory + proportional to the number of blocks on the datnode. Having a conservative + limit reduces the potential impact of decommissioning or maintenance of + a large number of nodes at once. A value of 0 means no limit will be enforced. </description> http://git-wip-us.apache.org/repos/asf/hadoop/blob/065a906b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDecommission.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDecommission.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDecommission.java index ebfa13a..445d3d2 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDecommission.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDecommission.java @@ -51,7 +51,7 @@ import org.apache.hadoop.hdfs.server.blockmanagement.BlockManager; 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.DecommissionManager; +import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeAdminManager; import org.apache.hadoop.hdfs.server.datanode.DataNode; import org.apache.hadoop.hdfs.server.datanode.DataNodeTestUtils; import org.apache.hadoop.hdfs.server.datanode.SimulatedFSDataset; @@ -256,9 +256,10 @@ public class TestDecommission extends AdminStatesBaseTest { startSimpleHACluster(3); - // Step 1, create a cluster with 4 DNs. Blocks are stored on the first 3 DNs. - // The last DN is empty. Also configure the last DN to have slow heartbeat - // so that it will be chosen as excess replica candidate during recommission. + // Step 1, create a cluster with 4 DNs. Blocks are stored on the + // first 3 DNs. The last DN is empty. Also configure the last DN to have + // slow heartbeat so that it will be chosen as excess replica candidate + // during recommission. // Step 1.a, copy blocks to the first 3 DNs. Given the replica count is the // same as # of DNs, each DN will have a replica for any block. @@ -290,9 +291,9 @@ public class TestDecommission extends AdminStatesBaseTest { // Step 3, recommission the first DN on SBN and ANN to create excess replica // It recommissions the node on SBN first to create potential - // inconsistent state. In production cluster, such insistent state can happen - // even if recommission command was issued on ANN first given the async nature - // of the system. + // inconsistent state. In production cluster, such insistent state can + // happen even if recommission command was issued on ANN first given the + // async nature of the system. // Step 3.a, ask SBN to recomm the first DN. // SBN has been fixed so that it no longer invalidates excess replica during @@ -301,10 +302,10 @@ public class TestDecommission extends AdminStatesBaseTest { // 1. the last DN would have been chosen as excess replica, given its // heartbeat is considered old. // Please refer to BlockPlacementPolicyDefault#chooseReplicaToDelete - // 2. After recommissionNode finishes, SBN has 3 live replicas ( 0, 1, 2 ) + // 2. After recommissionNode finishes, SBN has 3 live replicas (0, 1, 2) // and one excess replica ( 3 ) // After the fix, - // After recommissionNode finishes, SBN has 4 live replicas ( 0, 1, 2, 3 ) + // After recommissionNode finishes, SBN has 4 live replicas (0, 1, 2, 3) Thread.sleep(slowHeartbeatDNwaitTime); putNodeInService(1, decomNodeFromSBN); @@ -562,7 +563,8 @@ public class TestDecommission extends AdminStatesBaseTest { * federated cluster. */ @Test(timeout=360000) - public void testHostsFileFederation() throws IOException, InterruptedException { + public void testHostsFileFederation() + throws IOException, InterruptedException { // Test for 3 namenode federated cluster testHostsFile(3); } @@ -599,7 +601,8 @@ public class TestDecommission extends AdminStatesBaseTest { } @Test(timeout=120000) - public void testDecommissionWithOpenfile() throws IOException, InterruptedException { + public void testDecommissionWithOpenfile() + throws IOException, InterruptedException { LOG.info("Starting test testDecommissionWithOpenfile"); //At most 4 nodes will be decommissioned @@ -743,14 +746,15 @@ public class TestDecommission extends AdminStatesBaseTest { // make sure the two datanodes remain in decomm in progress state BlockManagerTestUtil.recheckDecommissionState(dm); - assertTrackedAndPending(dm.getDecomManager(), 2, 0); + assertTrackedAndPending(dm.getDatanodeAdminManager(), 2, 0); } /** * Tests restart of namenode while datanode hosts are added to exclude file **/ @Test(timeout=360000) - public void testDecommissionWithNamenodeRestart()throws IOException, InterruptedException { + public void testDecommissionWithNamenodeRestart() + throws IOException, InterruptedException { LOG.info("Starting test testDecommissionWithNamenodeRestart"); int numNamenodes = 1; int numDatanodes = 1; @@ -915,7 +919,7 @@ public class TestDecommission extends AdminStatesBaseTest { @Test(timeout=120000) public void testBlocksPerInterval() throws Exception { - org.apache.log4j.Logger.getLogger(DecommissionManager.class) + org.apache.log4j.Logger.getLogger(DatanodeAdminManager.class) .setLevel(Level.TRACE); // Turn the blocks per interval way down getConf().setInt( @@ -928,7 +932,8 @@ public class TestDecommission extends AdminStatesBaseTest { final FileSystem fs = getCluster().getFileSystem(); final DatanodeManager datanodeManager = getCluster().getNamesystem().getBlockManager().getDatanodeManager(); - final DecommissionManager decomManager = datanodeManager.getDecomManager(); + final DatanodeAdminManager decomManager = + datanodeManager.getDatanodeAdminManager(); // Write a 3 block file, so each node has one block. Should scan 3 nodes. DFSTestUtil.createFile(fs, new Path("/file1"), 64, (short) 3, 0xBAD1DEA); @@ -947,7 +952,7 @@ public class TestDecommission extends AdminStatesBaseTest { @Deprecated @Test(timeout=120000) public void testNodesPerInterval() throws Exception { - org.apache.log4j.Logger.getLogger(DecommissionManager.class) + org.apache.log4j.Logger.getLogger(DatanodeAdminManager.class) .setLevel(Level.TRACE); // Set the deprecated configuration key which limits the # of nodes per // interval @@ -959,7 +964,8 @@ public class TestDecommission extends AdminStatesBaseTest { final FileSystem fs = getCluster().getFileSystem(); final DatanodeManager datanodeManager = getCluster().getNamesystem().getBlockManager().getDatanodeManager(); - final DecommissionManager decomManager = datanodeManager.getDecomManager(); + final DatanodeAdminManager decomManager = + datanodeManager.getDatanodeAdminManager(); // Write a 3 block file, so each node has one block. Should scan 1 node // each time. @@ -970,7 +976,7 @@ public class TestDecommission extends AdminStatesBaseTest { } private void doDecomCheck(DatanodeManager datanodeManager, - DecommissionManager decomManager, int expectedNumCheckedNodes) + DatanodeAdminManager decomManager, int expectedNumCheckedNodes) throws IOException, ExecutionException, InterruptedException { // Decom all nodes ArrayList<DatanodeInfo> decommissionedNodes = Lists.newArrayList(); @@ -991,7 +997,7 @@ public class TestDecommission extends AdminStatesBaseTest { @Test(timeout=120000) public void testPendingNodes() throws Exception { - org.apache.log4j.Logger.getLogger(DecommissionManager.class) + org.apache.log4j.Logger.getLogger(DatanodeAdminManager.class) .setLevel(Level.TRACE); // Only allow one node to be decom'd at a time getConf().setInt( @@ -1004,7 +1010,8 @@ public class TestDecommission extends AdminStatesBaseTest { final FileSystem fs = getCluster().getFileSystem(); final DatanodeManager datanodeManager = getCluster().getNamesystem().getBlockManager().getDatanodeManager(); - final DecommissionManager decomManager = datanodeManager.getDecomManager(); + final DatanodeAdminManager decomManager = + datanodeManager.getDatanodeAdminManager(); // Keep a file open to prevent decom from progressing HdfsDataOutputStream open1 = @@ -1040,7 +1047,7 @@ public class TestDecommission extends AdminStatesBaseTest { assertTrackedAndPending(decomManager, 1, 0); } - private void assertTrackedAndPending(DecommissionManager decomManager, + private void assertTrackedAndPending(DatanodeAdminManager decomManager, int tracked, int pending) { assertEquals("Unexpected number of tracked nodes", tracked, decomManager.getNumTrackedNodes()); http://git-wip-us.apache.org/repos/asf/hadoop/blob/065a906b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManagerTestUtil.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManagerTestUtil.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManagerTestUtil.java index f99d84a..02fe506 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManagerTestUtil.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManagerTestUtil.java @@ -329,7 +329,7 @@ public class BlockManagerTestUtil { */ public static void recheckDecommissionState(DatanodeManager dm) throws ExecutionException, InterruptedException { - dm.getDecomManager().runMonitorForTest(); + dm.getDatanodeAdminManager().runMonitorForTest(); } public static void setStartupSafeModeForTest(BlockManager bm) { http://git-wip-us.apache.org/repos/asf/hadoop/blob/065a906b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestReplicationPolicyConsiderLoad.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestReplicationPolicyConsiderLoad.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestReplicationPolicyConsiderLoad.java index 1992fcb..bd3e4a6 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestReplicationPolicyConsiderLoad.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestReplicationPolicyConsiderLoad.java @@ -100,7 +100,7 @@ public class TestReplicationPolicyConsiderLoad // returns false for (int i = 0; i < 3; i++) { DatanodeDescriptor d = dataNodes[i]; - dnManager.getDecomManager().startDecommission(d); + dnManager.getDatanodeAdminManager().startDecommission(d); d.setDecommissioned(); } assertEquals((double)load/3, dnManager.getFSClusterStats() http://git-wip-us.apache.org/repos/asf/hadoop/blob/065a906b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestDecommissioningStatus.java ---------------------------------------------------------------------- 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 5b61b0a..cbb7970 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 @@ -52,7 +52,7 @@ import org.apache.hadoop.hdfs.server.blockmanagement.BlockManager; 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.DecommissionManager; +import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeAdminManager; import org.apache.hadoop.hdfs.server.datanode.DataNode; import org.apache.hadoop.hdfs.server.datanode.DataNodeTestUtils; import org.apache.hadoop.hdfs.tools.DFSAdmin; @@ -102,7 +102,7 @@ public class TestDecommissioningStatus { fileSys = cluster.getFileSystem(); cluster.getNamesystem().getBlockManager().getDatanodeManager() .setHeartbeatExpireInterval(3000); - Logger.getLogger(DecommissionManager.class).setLevel(Level.DEBUG); + Logger.getLogger(DatanodeAdminManager.class).setLevel(Level.DEBUG); LOG = Logger.getLogger(TestDecommissioningStatus.class); } @@ -365,7 +365,7 @@ public class TestDecommissioningStatus { */ @Test(timeout=120000) public void testDecommissionDeadDN() throws Exception { - Logger log = Logger.getLogger(DecommissionManager.class); + Logger log = Logger.getLogger(DatanodeAdminManager.class); log.setLevel(Level.DEBUG); DatanodeID dnID = cluster.getDataNodes().get(0).getDatanodeId(); String dnName = dnID.getXferAddr(); http://git-wip-us.apache.org/repos/asf/hadoop/blob/065a906b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestDefaultBlockPlacementPolicy.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestDefaultBlockPlacementPolicy.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestDefaultBlockPlacementPolicy.java index 1a10b7a..9be0b8f 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestDefaultBlockPlacementPolicy.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestDefaultBlockPlacementPolicy.java @@ -110,11 +110,11 @@ public class TestDefaultBlockPlacementPolicy { DatanodeDescriptor dnd3 = dnm.getDatanode( cluster.getDataNodes().get(3).getDatanodeId()); assertEquals(dnd3.getNetworkLocation(), clientRack); - dnm.getDecomManager().startDecommission(dnd3); + dnm.getDatanodeAdminManager().startDecommission(dnd3); try { testPlacement(clientMachine, clientRack, false); } finally { - dnm.getDecomManager().stopDecommission(dnd3); + dnm.getDatanodeAdminManager().stopDecommission(dnd3); } } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
