HDFS-680. Add new access method to a copy of a block's replica. Contributed by Konstantin Shvachko.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/hdfs/trunk@825033 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -276,6 +276,8 @@ Release 0.21.0 - Unreleased
     HDFS-29. Validate the consistency of the lengths of replica and its file 
     in replica recovery.  (szetszwo)
 
+    HDFS-680. Add new access method to a copy of a block's replica. (shv)
+
   BUG FIXES
 
     HDFS-76. Better error message to users when commands fail because of 

src/java/org/apache/hadoop/hdfs/server/datanode/FSDataset.java

@@ -775,6 +775,31 @@ public class FSDataset implements FSConstants, FSDatasetInterface {
         parseGenerationStamp(blockfile, metafile));
   }
 
+  /**
+   * Returns a clone of a replica stored in data-node memory.
+   * Should be primarily used for testing.
+   * @param blockId
+   * @return
+   */
+  synchronized ReplicaInfo fetchReplicaInfo(long blockId) {
+    ReplicaInfo r = volumeMap.get(blockId);
+    if(r == null)
+      return null;
+    switch(r.getState()) {
+    case FINALIZED:
+      return new FinalizedReplica((FinalizedReplica)r);
+    case RBW:
+      return new ReplicaBeingWritten((ReplicaBeingWritten)r);
+    case RWR:
+      return new ReplicaWaitingToBeRecovered((ReplicaWaitingToBeRecovered)r);
+    case RUR:
+      return new ReplicaUnderRecovery((ReplicaUnderRecovery)r);
+    case TEMPORARY:
+      return new ReplicaInPipeline((ReplicaInPipeline)r);
+    }
+    return null;
+  }
+
   public boolean metaFileExists(Block b) throws IOException {
     return getMetaFile(b).exists();
   }
@@ -1916,7 +1941,11 @@ public class FSDataset implements FSConstants, FSDatasetInterface {
     }
   }
 
+  /**
+   * @deprecated use {@link #fetchReplicaInfo(long)} instead.
+   */
   @Override
+  @Deprecated
   public ReplicaInfo getReplica(long blockId) {
     assert(Thread.holdsLock(this));
     return volumeMap.get(blockId);
@@ -2070,7 +2099,7 @@ public class FSDataset implements FSConstants, FSDatasetInterface {
   @Override // FSDatasetInterface
   public synchronized long getReplicaVisibleLength(final Block block)
   throws IOException {
-    final Replica replica = getReplica(block.getBlockId());
+    final Replica replica = volumeMap.get(block.getBlockId());
     if (replica == null) {
       throw new ReplicaNotFoundException(block);
     }

src/java/org/apache/hadoop/hdfs/server/datanode/FSDatasetInterface.java

@@ -102,6 +102,7 @@ public interface FSDatasetInterface extends FSDatasetMBean {
    * @param blockId
    * @return replica from the replicas map
    */
+  @Deprecated
   public Replica getReplica(long blockId);
 
   /**

src/java/org/apache/hadoop/hdfs/server/datanode/FinalizedReplica.java

@@ -51,7 +51,16 @@ class FinalizedReplica extends ReplicaInfo {
   FinalizedReplica(Block block, FSVolume vol, File dir) {
     super(block, vol, dir);
   }
-  
+
+  /**
+   * Copy constructor.
+   * @param from
+   */
+  FinalizedReplica(FinalizedReplica from) {
+    super(from);
+    this.unlinked = from.isUnlinked();
+  }
+
   @Override  // ReplicaInfo
   public ReplicaState getState() {
     return ReplicaState.FINALIZED;

src/java/org/apache/hadoop/hdfs/server/datanode/ReplicaBeingWritten.java

@@ -35,7 +35,7 @@ class ReplicaBeingWritten extends ReplicaInPipeline {
    * @param vol volume where replica is located
    * @param dir directory path where block and meta files are located
    */
-    ReplicaBeingWritten(long blockId, long genStamp, 
+  ReplicaBeingWritten(long blockId, long genStamp, 
         FSVolume vol, File dir) {
     super( blockId, genStamp, vol, dir);
   }
@@ -65,7 +65,15 @@ class ReplicaBeingWritten extends ReplicaInPipeline {
       FSVolume vol, File dir, Thread writer ) {
     super( blockId, len, genStamp, vol, dir, writer);
   }
-  
+
+  /**
+   * Copy constructor.
+   * @param from
+   */
+  ReplicaBeingWritten(ReplicaBeingWritten from) {
+    super(from);
+  }
+
   @Override
   public long getVisibleLength() {
     return getBytesAcked();       // all acked bytes are visible

src/java/org/apache/hadoop/hdfs/server/datanode/ReplicaInPipeline.java

@@ -85,6 +85,17 @@ class ReplicaInPipeline extends ReplicaInfo
     this.writer = writer;
   }
 
+  /**
+   * Copy constructor.
+   * @param from
+   */
+  ReplicaInPipeline(ReplicaInPipeline from) {
+    super(from);
+    this.bytesAcked = from.getBytesAcked();
+    this.bytesOnDisk = from.getBytesOnDisk();
+    this.writer = from.writer;
+  }
+
   @Override
   public long getVisibleLength() {
     return -1;

src/java/org/apache/hadoop/hdfs/server/datanode/ReplicaInfo.java

@@ -73,6 +73,14 @@ abstract public class ReplicaInfo extends Block implements Replica {
     this.dir = dir;
   }
 
+  /**
+   * Copy constructor.
+   * @param from
+   */
+  ReplicaInfo(ReplicaInfo from) {
+    this(from, from.getVolume(), from.getDir());
+  }
+
   /**
    * Get this replica's meta file name
    * @return this replica's meta file name

src/java/org/apache/hadoop/hdfs/server/datanode/ReplicaUnderRecovery.java

@@ -48,6 +48,16 @@ class ReplicaUnderRecovery extends ReplicaInfo {
     this.recoveryId = recoveryId;
   }
 
+  /**
+   * Copy constructor.
+   * @param from
+   */
+  ReplicaUnderRecovery(ReplicaUnderRecovery from) {
+    super(from);
+    this.original = from.getOriginalReplica();
+    this.recoveryId = from.getRecoveryID();
+  }
+
   /** 
    * Get the recovery id
    * @return the generation stamp that the replica will be bumped to 

src/java/org/apache/hadoop/hdfs/server/datanode/ReplicaWaitingToBeRecovered.java

@@ -58,6 +58,15 @@ class ReplicaWaitingToBeRecovered extends ReplicaInfo {
     super(block, vol, dir);
   }
   
+  /**
+   * Copy constructor.
+   * @param from
+   */
+  ReplicaWaitingToBeRecovered(ReplicaWaitingToBeRecovered from) {
+    super(from);
+    this.unlinked = from.isUnlinked();
+  }
+
   @Override //ReplicaInfo
   public ReplicaState getState() {
     return ReplicaState.RWR;

src/test/hdfs/org/apache/hadoop/hdfs/server/datanode/SimulatedFSDataset.java

@@ -393,6 +393,7 @@ public class SimulatedFSDataset  implements FSConstants, FSDatasetInterface, Con
   }
 
   @Override
+  @Deprecated
   public Replica getReplica(long blockId) {
     return blockMap.get(new Block(blockId));
   }

src/test/hdfs/org/apache/hadoop/hdfs/server/datanode/TestDirectoryScanner.java

@@ -120,12 +120,7 @@ public class TestDirectoryScanner extends TestCase {
     long id = rand.nextLong();
     while (true) {
       id = rand.nextLong();
-      Block b = new Block(id);
-      ReplicaInfo info = null;
-      synchronized(fds) {
-        info = fds.volumeMap.get(b);
-      }
-      if (info == null) {
+      if (fds.fetchReplicaInfo(id) == null) {
         break;
       }
     }
@@ -320,14 +315,12 @@ public class TestDirectoryScanner extends TestCase {
 
   private void verifyAddition(long blockId, long genStamp, long size) {
     final ReplicaInfo replicainfo;
-    synchronized(fds) {
-      replicainfo = fds.getReplica(blockId);
-    }
+    replicainfo = fds.fetchReplicaInfo(blockId);
     assertNotNull(replicainfo);
 
     // Added block has the same file as the one created by the test
     File file = new File(getBlockFile(blockId));
-    assertEquals(file.getName(), replicainfo.getBlockFile().getName());
+    assertEquals(file.getName(), fds.findBlockFile(blockId).getName());
 
     // Generation stamp is same as that of created file
     assertEquals(genStamp, replicainfo.getGenerationStamp());
@@ -338,16 +331,12 @@ public class TestDirectoryScanner extends TestCase {
 
   private void verifyDeletion(long blockId) {
     // Ensure block does not exist in memory
-    synchronized(fds) {
-      assertEquals(null, fds.volumeMap.get(new Block(blockId)));
-    }
+    assertNull(fds.fetchReplicaInfo(blockId));
   }
 
   private void verifyGenStamp(long blockId, long genStamp) {
-    final Replica memBlock;
-    synchronized(fds) {
-      memBlock = fds.getReplica(blockId);
-    }
+    final ReplicaInfo memBlock;
+    memBlock = fds.fetchReplicaInfo(blockId);
     assertNotNull(memBlock);
     assertEquals(genStamp, memBlock.getGenerationStamp());
   }

src/test/hdfs/org/apache/hadoop/hdfs/server/datanode/TestInterDatanodeProtocol.java

@@ -34,6 +34,7 @@ import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
 import org.apache.hadoop.hdfs.protocol.LocatedBlock;
 import org.apache.hadoop.hdfs.protocol.LocatedBlocks;
 import org.apache.hadoop.hdfs.protocol.RecoveryInProgressException;
+import org.apache.hadoop.hdfs.server.common.HdfsConstants.ReplicaState;
 import org.apache.hadoop.hdfs.server.protocol.InterDatanodeProtocol;
 import org.apache.hadoop.hdfs.server.protocol.ReplicaRecoveryInfo;
 import org.apache.hadoop.hdfs.server.protocol.BlockRecoveryCommand.RecoveringBlock;
@@ -225,16 +226,15 @@ public class TestInterDatanodeProtocol {
       final Block b = locatedblock.getBlock();
       final long recoveryid = b.getGenerationStamp() + 1;
       final long newlength = b.getNumBytes() - 1;
-      final ReplicaRecoveryInfo rri = FSDataset.initReplicaRecovery(
-          fsdataset.volumeMap, b, recoveryid);
+      final ReplicaRecoveryInfo rri = fsdataset.initReplicaRecovery(
+          new RecoveringBlock(b, null, recoveryid));
 
       //check replica
-      final ReplicaInfo replica = fsdataset.volumeMap.get(b.getBlockId());
-      Assert.assertTrue(replica instanceof ReplicaUnderRecovery);
-      final ReplicaUnderRecovery rur = (ReplicaUnderRecovery)replica;
+      final ReplicaInfo replica = fsdataset.fetchReplicaInfo(b.getBlockId());
+      Assert.assertEquals(ReplicaState.RUR, replica.getState());
 
       //check meta data before update
-      FSDataset.checkReplicaFiles(rur);
+      FSDataset.checkReplicaFiles(replica);
 
       //case "THIS IS NOT SUPPOSED TO HAPPEN"
       //with (block length) != (stored replica's on disk length).