test

Author: hfutatzhanghb

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java

@@ -172,6 +172,7 @@
 import org.apache.hadoop.hdfs.server.datanode.fsdataset.impl.BlockPoolSlice;
 import org.apache.hadoop.hdfs.server.datanode.fsdataset.impl.FsVolumeImpl;
 import org.apache.hadoop.hdfs.util.DataTransferThrottler;
+import org.apache.hadoop.ipc.RetriableException;
 import org.apache.hadoop.ipc.Server;
 import org.apache.hadoop.util.*;
 import org.apache.hadoop.hdfs.client.BlockReportOptions;
@@ -753,13 +754,13 @@ private String reconfDataXceiverParameters(String property, String newVal)
     try {
       LOG.info("Reconfiguring {} to {}", property, newVal);
       if (property.equals(DFS_DATANODE_MAX_RECEIVER_THREADS_KEY)) {
-        Preconditions.checkNotNull(getXferServer(), "DataXceiverServer has not been initialized.");
+        checkNotNull(getXferServer(), "DataXceiverServer has not been initialized.");
         int threads = (newVal == null ? DFS_DATANODE_MAX_RECEIVER_THREADS_DEFAULT :
             Integer.parseInt(newVal));
         result = Integer.toString(threads);
         getXferServer().setMaxXceiverCount(threads);
       } else if (property.equals(DFS_DATANODE_DATA_TRANSFER_BANDWIDTHPERSEC_KEY)) {
-        Preconditions.checkNotNull(getXferServer(), "DataXceiverServer has not been initialized.");
+        checkNotNull(getXferServer(), "DataXceiverServer has not been initialized.");
         long bandwidthPerSec = (newVal == null ?
             DFS_DATANODE_DATA_TRANSFER_BANDWIDTHPERSEC_DEFAULT : Long.parseLong(newVal));
         DataTransferThrottler transferThrottler = null;
@@ -771,7 +772,7 @@ private String reconfDataXceiverParameters(String property, String newVal)
         result = Long.toString(bandwidthPerSec);
         getXferServer().setTransferThrottler(transferThrottler);
       } else if (property.equals(DFS_DATANODE_DATA_WRITE_BANDWIDTHPERSEC_KEY)) {
-        Preconditions.checkNotNull(getXferServer(), "DataXceiverServer has not been initialized.");
+        checkNotNull(getXferServer(), "DataXceiverServer has not been initialized.");
         long bandwidthPerSec = (newVal == null ? DFS_DATANODE_DATA_WRITE_BANDWIDTHPERSEC_DEFAULT :
             Long.parseLong(newVal));
         DataTransferThrottler writeThrottler = null;
@@ -783,7 +784,7 @@ private String reconfDataXceiverParameters(String property, String newVal)
         result = Long.toString(bandwidthPerSec);
         getXferServer().setWriteThrottler(writeThrottler);
       } else if (property.equals(DFS_DATANODE_DATA_READ_BANDWIDTHPERSEC_KEY)) {
-        Preconditions.checkNotNull(getXferServer(), "DataXceiverServer has not been initialized.");
+        checkNotNull(getXferServer(), "DataXceiverServer has not been initialized.");
         long bandwidthPerSec = (newVal == null ? DFS_DATANODE_DATA_READ_BANDWIDTHPERSEC_DEFAULT :
             Long.parseLong(newVal));
         DataTransferThrottler readThrottler = null;
@@ -807,7 +808,7 @@ private String reconfCacheReportParameters(String property, String newVal)
     String result;
     try {
       LOG.info("Reconfiguring {} to {}", property, newVal);
-      Preconditions.checkNotNull(dnConf, "DNConf has not been initialized.");
+      checkNotNull(dnConf, "DNConf has not been initialized.");
       long reportInterval = (newVal == null ? DFS_CACHEREPORT_INTERVAL_MSEC_DEFAULT :
           Long.parseLong(newVal));
       result = Long.toString(reportInterval);
@@ -825,7 +826,7 @@ private String reconfBlockReportParameters(String property, String newVal)
     try {
       LOG.info("Reconfiguring {} to {}", property, newVal);
       if (property.equals(DFS_BLOCKREPORT_INTERVAL_MSEC_KEY)) {
-        Preconditions.checkNotNull(dnConf, "DNConf has not been initialized.");
+        checkNotNull(dnConf, "DNConf has not been initialized.");
         long intervalMs = newVal == null ? DFS_BLOCKREPORT_INTERVAL_MSEC_DEFAULT :
             Long.parseLong(newVal);
         result = Long.toString(intervalMs);
@@ -838,13 +839,13 @@ private String reconfBlockReportParameters(String property, String newVal)
           }
         }
       } else if (property.equals(DFS_BLOCKREPORT_SPLIT_THRESHOLD_KEY)) {
-        Preconditions.checkNotNull(dnConf, "DNConf has not been initialized.");
+        checkNotNull(dnConf, "DNConf has not been initialized.");
         long threshold = newVal == null ? DFS_BLOCKREPORT_SPLIT_THRESHOLD_DEFAULT :
             Long.parseLong(newVal);
         result = Long.toString(threshold);
         dnConf.setBlockReportSplitThreshold(threshold);
       } else if (property.equals(DFS_BLOCKREPORT_INITIAL_DELAY_KEY)) {
-        Preconditions.checkNotNull(dnConf, "DNConf has not been initialized.");
+        checkNotNull(dnConf, "DNConf has not been initialized.");
         int initialDelay = newVal == null ? DFS_BLOCKREPORT_INITIAL_DELAY_DEFAULT :
             Integer.parseInt(newVal);
         result = Integer.toString(initialDelay);
@@ -863,7 +864,7 @@ private String reconfSlowPeerParameters(String property, String newVal)
     try {
       LOG.info("Reconfiguring {} to {}", property, newVal);
       if (property.equals(DFS_DATANODE_PEER_STATS_ENABLED_KEY)) {
-        Preconditions.checkNotNull(dnConf, "DNConf has not been initialized.");
+        checkNotNull(dnConf, "DNConf has not been initialized.");
         if (newVal != null && !newVal.equalsIgnoreCase("true")
             && !newVal.equalsIgnoreCase("false")) {
           throw new IllegalArgumentException("Not a valid Boolean value for " + property +
@@ -878,19 +879,19 @@ private String reconfSlowPeerParameters(String property, String newVal)
           peerMetrics = DataNodePeerMetrics.create(getDisplayName(), getConf());
         }
       } else if (property.equals(DFS_DATANODE_MIN_OUTLIER_DETECTION_NODES_KEY)) {
-        Preconditions.checkNotNull(peerMetrics, "DataNode peer stats may be disabled.");
+        checkNotNull(peerMetrics, "DataNode peer stats may be disabled.");
         long minNodes = (newVal == null ? DFS_DATANODE_MIN_OUTLIER_DETECTION_NODES_DEFAULT :
             Long.parseLong(newVal));
         result = Long.toString(minNodes);
         peerMetrics.setMinOutlierDetectionNodes(minNodes);
       } else if (property.equals(DFS_DATANODE_SLOWPEER_LOW_THRESHOLD_MS_KEY)) {
-        Preconditions.checkNotNull(peerMetrics, "DataNode peer stats may be disabled.");
+        checkNotNull(peerMetrics, "DataNode peer stats may be disabled.");
         long threshold = (newVal == null ? DFS_DATANODE_SLOWPEER_LOW_THRESHOLD_MS_DEFAULT :
             Long.parseLong(newVal));
         result = Long.toString(threshold);
         peerMetrics.setLowThresholdMs(threshold);
       } else if (property.equals(DFS_DATANODE_PEER_METRICS_MIN_OUTLIER_DETECTION_SAMPLES_KEY)) {
-        Preconditions.checkNotNull(peerMetrics, "DataNode peer stats may be disabled.");
+        checkNotNull(peerMetrics, "DataNode peer stats may be disabled.");
         long minSamples = (newVal == null ?
             DFS_DATANODE_PEER_METRICS_MIN_OUTLIER_DETECTION_SAMPLES_DEFAULT :
             Long.parseLong(newVal));
@@ -1066,7 +1067,7 @@ private String reconfSlowIoWarningThresholdParameters(String property, String ne
     String result;
     try {
       LOG.info("Reconfiguring {} to {}", property, newVal);
-      Preconditions.checkNotNull(dnConf, "DNConf has not been initialized.");
+      checkNotNull(dnConf, "DNConf has not been initialized.");
       long slowIoWarningThreshold = (newVal == null ?
           DFS_DATANODE_SLOW_IO_WARNING_THRESHOLD_DEFAULT :
           Long.parseLong(newVal));
@@ -1484,7 +1485,7 @@ private static String getHostName(Configuration config)
   }
 
   /**
-   * @see DFSUtil#getHttpPolicy(org.apache.hadoop.conf.Configuration)
+   * @see DFSUtil#getHttpPolicy(Configuration)
    * for information related to the different configuration options and
    * Http Policy is decided.
    */
@@ -1812,7 +1813,7 @@ public void reportCorruptedBlocks(
     Map<ExtendedBlock, Set<DatanodeInfo>> corruptionMap =
         corruptedBlocks.getCorruptionMap();
     if (corruptionMap != null) {
-      for (Map.Entry<ExtendedBlock, Set<DatanodeInfo>> entry :
+      for (Entry<ExtendedBlock, Set<DatanodeInfo>> entry :
           corruptionMap.entrySet()) {
         for (DatanodeInfo dnInfo : entry.getValue()) {
           reportRemoteBadBlock(dnInfo, entry.getKey());
@@ -1828,7 +1829,7 @@ public void reportCorruptedBlocks(
    */
   private BPOfferService getBPOSForBlock(ExtendedBlock block)
       throws IOException {
-    Preconditions.checkNotNull(block);
+    checkNotNull(block);
     BPOfferService bpos = blockPoolManager.get(block.getBlockPoolId());
     if (bpos == null) {
       throw new IOException("cannot locate OfferService thread for bp="+
@@ -2438,7 +2439,7 @@ public long getMaxNumberOfBlocksToLog() {
   public BlockLocalPathInfo getBlockLocalPathInfo(ExtendedBlock block,
       Token<BlockTokenIdentifier> token) throws IOException {
     checkBlockLocalPathAccess();
-    checkBlockToken(block, token, BlockTokenIdentifier.AccessMode.READ);
+    checkBlockToken(block, token, AccessMode.READ);
     checkStorageState("getBlockLocalPathInfo");
     BlockLocalPathInfo info = data.getBlockLocalPathInfo(block);
     if (info != null) {
@@ -2886,7 +2887,7 @@ void transferBlock(ExtendedBlock block, DatanodeInfo[] xferTargets,
 
       final DataTransfer dataTransferTask = new DataTransfer(xferTargets,
           xferTargetStorageTypes, xferTargetStorageIDs, block,
-          BlockConstructionStage.PIPELINE_SETUP_CREATE, "");
+          PIPELINE_SETUP_CREATE, "");
 
       this.xferService.execute(dataTransferTask);
     }
@@ -3068,7 +3069,7 @@ public void run() {
         // Header info
         //
         Token<BlockTokenIdentifier> accessToken = getBlockAccessToken(b,
-            EnumSet.of(BlockTokenIdentifier.AccessMode.WRITE),
+            EnumSet.of(AccessMode.WRITE),
             targetStorageTypes, targetStorageIds);
 
         long writeTimeout = dnConf.socketWriteTimeout + 
@@ -3561,7 +3562,7 @@ private void checkReadAccess(final ExtendedBlock block) throws IOException {
       getDNRegistrationForBP(block.getBlockPoolId());
     } catch (IOException e) {
       // if it has not registered with the NN, throw an exception back.
-      throw new org.apache.hadoop.ipc.RetriableException(
+      throw new RetriableException(
           "Datanode not registered. Try again later.");
     }
 
@@ -3576,7 +3577,7 @@ private void checkReadAccess(final ExtendedBlock block) throws IOException {
         BlockTokenIdentifier id = (BlockTokenIdentifier) tokenId;
         LOG.debug("BlockTokenIdentifier: {}", id);
         blockPoolTokenSecretManager.checkAccess(id, null, block,
-            BlockTokenIdentifier.AccessMode.READ, null, null);
+            AccessMode.READ, null, null);
       }
     }
   }
@@ -4275,7 +4276,7 @@ public DiskBalancerWorkStatus queryDiskBalancerPlan() throws IOException {
   @Override
   public String getDiskBalancerSetting(String key) throws IOException {
     checkSuperuserPrivilege();
-    Preconditions.checkNotNull(key);
+    checkNotNull(key);
     switch (key) {
     case DiskBalancerConstants.DISKBALANCER_VOLUME_NAME:
       return getDiskBalancer().getVolumeNames();

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataSetSubLockStrategy.java

@@ -0,0 +1,36 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.util.List;
+
+/**
+ * This interface is used to generate sub lock name for a blockid.
+ */
+public interface DataSetSubLockStrategy {
+
+  /**
+   * Generate sub lock name for the given blockid.
+   * @param blockid
+   * @return sub lock name for the input blockid.
+   */
+  String blockIdToSubLock(long blockid);
+
+  List<String> getAllSubLockName();
+}

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DatanodeUtil.java

@@ -129,31 +129,6 @@ public static File idToBlockDir(File root, long blockId) {
     return new File(root, path);
   }
 
-  /**
-   * Take an example. We hava a block with blockid mapping to:
-   * "/data1/hadoop/hdfs/datanode/current/BP-xxxx/current/finalized/subdir0/subdir0"
-   * We return "subdir0/subdir0"
-   * @param blockId blockId
-   * @return The two-level subdir name
-   */
-  public static String idToBlockDirSuffixName(long blockId) {
-    int d1 = (int) ((blockId >> 16) & 0x1F);
-    int d2 = (int) ((blockId >> 8) & 0x1F);
-    return DataStorage.BLOCK_SUBDIR_PREFIX + d1 + SEP +
-        DataStorage.BLOCK_SUBDIR_PREFIX + d2;
-  }
-
-  public static List<String> getAllSubDirNameForDataSetLock() {
-    List<String> res = new ArrayList<>();
-    for (int d1 = 0; d1 <= 0x1F; d1++) {
-      for (int d2 = 0; d2 <= 0x1F; d2++) {
-        res.add(DataStorage.BLOCK_SUBDIR_PREFIX + d1 + SEP +
-            DataStorage.BLOCK_SUBDIR_PREFIX + d2);
-      }
-    }
-    return res;
-  }
-
   /**
    * @return the FileInputStream for the meta data of the given block.
    * @throws FileNotFoundException

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/ModDataSetSubLockStrategy.java

@@ -0,0 +1,53 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class ModDataSetSubLockStrategy implements DataSetSubLockStrategy {
+  public static final Logger LOG = LoggerFactory.getLogger(DataSetSubLockStrategy.class);
+  
+  private static final String LOCK_NAME_PERFIX = "SubLock";
+  private long modFactor;
+  
+  public ModDataSetSubLockStrategy(long mod) {
+    if (mod <= 0) {
+      mod = 1L;
+    }
+    this.modFactor = mod;
+  }
+  
+  @Override
+  public String blockIdToSubLock(long blockid) {
+    return LOCK_NAME_PERFIX + String.valueOf(blockid % modFactor);
+  }
+
+  @Override
+  public List<String> getAllSubLockName() {
+    List<String> res = new ArrayList<>();
+    for (long i = 0L; i < modFactor; i++) {
+      res.add(LOCK_NAME_PERFIX + i);
+    }
+    return res;
+  }
+}