gt:           + nsInfo.getBlockPoolID(), e);
context: /**
   * 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.datanode;
  
  import java.io.File;
  import java.io.IOException;
  import java.net.URI;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Properties;
  import java.util.Set;
  import java.util.concurrent.Callable;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.regex.Matcher;
  import java.util.regex.Pattern;
  
  import org.apache.hadoop.classification.InterfaceAudience;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.fs.FileUtil;
  import org.apache.hadoop.fs.HardLink;
  import org.apache.hadoop.hdfs.protocol.LayoutVersion;
  import org.apache.hadoop.hdfs.server.common.HdfsServerConstants;
  import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.NodeType;
  import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.StartupOption;
  import org.apache.hadoop.hdfs.server.common.InconsistentFSStateException;
  import org.apache.hadoop.hdfs.server.common.Storage;
  import org.apache.hadoop.hdfs.server.common.StorageInfo;
  import org.apache.hadoop.hdfs.server.protocol.NamespaceInfo;
  import org.apache.hadoop.util.Daemon;
  
  import com.google.common.annotations.VisibleForTesting;
  import com.google.common.base.Preconditions;
  import com.google.common.collect.Lists;
  
  /**
   * Manages storage for the set of BlockPoolSlices which share a particular 
   * block pool id, on this DataNode.
   * 
   * This class supports the following functionality:
   * <ol>
   * <li> Formatting a new block pool storage</li>
   * <li> Recovering a storage state to a consistent state (if possible></li>
   * <li> Taking a snapshot of the block pool during upgrade</li>
   * <li> Rolling back a block pool to a previous snapshot</li>
   * <li> Finalizing block storage by deletion of a snapshot</li>
   * </ul>
   * 
   * @see Storage
   */
  @InterfaceAudience.Private
  public class BlockPoolSliceStorage extends Storage {
    static final String TRASH_ROOT_DIR = "trash";
  
    /**
     * A marker file that is created on each root directory if a rolling upgrade
     * is in progress. The NN does not inform the DN when a rolling upgrade is
     * finalized. All the DN can infer is whether or not a rolling upgrade is
     * currently in progress. When the rolling upgrade is not in progress:
     *   1. If the marker file is present, then a rolling upgrade just completed.
     *      If a 'previous' directory exists, it can be deleted now.
     *   2. If the marker file is absent, then a regular upgrade may be in
     *      progress. Do not delete the 'previous' directory.
     */
    static final String ROLLING_UPGRADE_MARKER_FILE = "RollingUpgradeInProgress";
  
    private static final String BLOCK_POOL_ID_PATTERN_BASE =
        Pattern.quote(File.separator) +
        "BP-\\d+-\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}-\\d+" +
        Pattern.quote(File.separator);
  
    private static final Pattern BLOCK_POOL_PATH_PATTERN = Pattern.compile(
        "^(.*)(" + BLOCK_POOL_ID_PATTERN_BASE + ")(.*)$");
  
    private static final Pattern BLOCK_POOL_CURRENT_PATH_PATTERN = Pattern.compile(
        "^(.*)(" + BLOCK_POOL_ID_PATTERN_BASE + ")(" + STORAGE_DIR_CURRENT + ")(.*)$");
  
    private static final Pattern BLOCK_POOL_TRASH_PATH_PATTERN = Pattern.compile(
        "^(.*)(" + BLOCK_POOL_ID_PATTERN_BASE + ")(" + TRASH_ROOT_DIR + ")(.*)$");
  
    private String blockpoolID = ""; // id of the blockpool
    private Daemon trashCleaner;
  
    public BlockPoolSliceStorage(StorageInfo storageInfo, String bpid) {
      super(storageInfo);
      blockpoolID = bpid;
    }
  
    /**
     * These maps are used as an optimization to avoid one filesystem operation
     * per storage on each heartbeat response.
     */
    private static Set<String> storagesWithRollingUpgradeMarker;
    private static Set<String> storagesWithoutRollingUpgradeMarker;
  
    BlockPoolSliceStorage(int namespaceID, String bpID, long cTime,
        String clusterId) {
      super(NodeType.DATA_NODE);
      this.namespaceID = namespaceID;
      this.blockpoolID = bpID;
      this.cTime = cTime;
      this.clusterID = clusterId;
      storagesWithRollingUpgradeMarker = Collections.newSetFromMap(
          new ConcurrentHashMap<String, Boolean>());
      storagesWithoutRollingUpgradeMarker = Collections.newSetFromMap(
          new ConcurrentHashMap<String, Boolean>());
    }
  
    private BlockPoolSliceStorage() {
      super(NodeType.DATA_NODE);
      storagesWithRollingUpgradeMarker = Collections.newSetFromMap(
          new ConcurrentHashMap<String, Boolean>());
      storagesWithoutRollingUpgradeMarker = Collections.newSetFromMap(
          new ConcurrentHashMap<String, Boolean>());
    }
  
    // Expose visibility for VolumeBuilder#commit().
    public void addStorageDir(StorageDirectory sd) {
      super.addStorageDir(sd);
    }
  
    /**
     * Load one storage directory. Recover from previous transitions if required.
     *
     * @param nsInfo namespace information
     * @param dataDir the root path of the storage directory
     * @param startOpt startup option
     * @return the StorageDirectory successfully loaded.
     * @throws IOException
     */
    private StorageDirectory loadStorageDirectory(NamespaceInfo nsInfo,
        StorageLocation location, StartupOption startOpt,
        List<Callable<StorageDirectory>> callables, Configuration conf)
            throws IOException {
      StorageDirectory sd = new StorageDirectory(
          nsInfo.getBlockPoolID(), null, true, location);
      try {
        StorageState curState = sd.analyzeStorage(startOpt, this, true);
        // sd is locked but not opened
        switch (curState) {
        case NORMAL:
          break;
        case NON_EXISTENT:
          LOG.info("Block pool storage directory for location " + location +
              " and block pool id " + nsInfo.getBlockPoolID() +
              " does not exist");
          throw new IOException("Storage directory for location " + location +
              " and block pool id " + nsInfo.getBlockPoolID() +
              " does not exist");
        case NOT_FORMATTED: // format
          LOG.info("Block pool storage directory for location " + location +
              " and block pool id " + nsInfo.getBlockPoolID()
              + " is not formatted for " + nsInfo.getBlockPoolID()
              + ". Formatting ...");
          format(sd, nsInfo);
          break;
        default:  // recovery part is common
          sd.doRecover(curState);
        }
  
        // 2. Do transitions
        // Each storage directory is treated individually.
        // During startup some of them can upgrade or roll back
        // while others could be up-to-date for the regular startup.
        if (!doTransition(sd, nsInfo, startOpt, callables, conf)) {
  
          // 3. Check CTime and update successfully loaded storage.
          if (getCTime() != nsInfo.getCTime()) {
            throw new IOException("Datanode CTime (=" + getCTime()
                + ") is not equal to namenode CTime (=" + nsInfo.getCTime() + ")");
          }
          setServiceLayoutVersion(getServiceLayoutVersion());
          writeProperties(sd);
        }
  
        return sd;
      } catch (IOException ioe) {
        sd.unlock();
        throw ioe;
      }
    }
  
    /**
     * Analyze and load storage directories. Recover from previous transitions if
     * required.
     *
     * The block pool storages are either all analyzed or none of them is loaded.
     * Therefore, a failure on loading any block pool storage results a faulty
     * data volume.
     *
     * @param nsInfo namespace information
     * @param dataDirs storage directories of block pool
     * @param startOpt startup option
     * @return an array of loaded block pool directories.
     * @throws IOException on error
     */
    List<StorageDirectory> loadBpStorageDirectories(NamespaceInfo nsInfo,
        StorageLocation location, StartupOption startOpt,
        List<Callable<StorageDirectory>> callables, Configuration conf)
            throws IOException {
      List<StorageDirectory> succeedDirs = Lists.newArrayList();
      try {
        if (containsStorageDir(location, nsInfo.getBlockPoolID())) {
          throw new IOException(
              "BlockPoolSliceStorage.recoverTransitionRead: " +
                  "attempt to load an used block storage: " + location);
        }
        final StorageDirectory sd = loadStorageDirectory(
            nsInfo, location, startOpt, callables, conf);
        succeedDirs.add(sd);
      } catch (IOException e) {
        LOG.warn("Failed to analyze storage directories for block pool "
  