From 6f63566694f8cec64a469448a8fa00ce921ce367 Mon Sep 17 00:00:00 2001
From: Kihwal Lee <kihwal@apache.org>
Date: Wed, 3 Aug 2016 15:19:31 -0500
Subject: [PATCH] HDFS-10569. A bug causes OutOfIndex error in
 BlockListAsLongs. Contributed by Weiwei Yang.

---
 .../hdfs/protocol/BlockListAsLongs.java       |  4 ++--
 .../hdfs/protocol/TestBlockListAsLongs.java   | 19 ++++++++++++++++++-
 2 files changed, 20 insertions(+), 3 deletions(-)

diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/BlockListAsLongs.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/BlockListAsLongs.java
index 26c7ffb02ce..8f482e3531b 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/BlockListAsLongs.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/BlockListAsLongs.java
@@ -468,8 +468,8 @@ public long[] getBlockListAsLongs() {
       long[] longs = new long[2+values.size()];
       longs[0] = finalizedBlocks;
       longs[1] = numBlocks - finalizedBlocks;
-      for (int i=0; i < longs.length; i++) {
-        longs[i] = values.get(i);
+      for(int i=0; i<values.size(); i++) {
+        longs[2+i] = values.get(i);
       }
       return longs;
     }
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/protocol/TestBlockListAsLongs.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/protocol/TestBlockListAsLongs.java
index 2cc1f7d80bd..cd2b5e86e35 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/protocol/TestBlockListAsLongs.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/protocol/TestBlockListAsLongs.java
@@ -188,6 +188,20 @@ private void checkReplicas(Map<Long,Replica> expectedReplicas,
     assertTrue(reportReplicas.isEmpty());
   }
 
+  private BlockListAsLongs getBlockList(Replica ... replicas) {
+    int numBlocks = replicas.length;
+    List<Long> longs = new ArrayList<Long>(2 + numBlocks);
+    longs.add(Long.valueOf(numBlocks));
+    longs.add(0L);
+    for(Replica r : replicas) {
+      longs.add(r.getBlockId());
+      longs.add(r.getBytesOnDisk());
+      longs.add(r.getGenerationStamp());
+    }
+    BlockListAsLongs blockList = BlockListAsLongs.decodeLongs(longs);
+    return blockList;
+  }
+
   @Test
   public void testCapabilitiesInited() {
     NamespaceInfo nsInfo = new NamespaceInfo();
@@ -237,7 +251,10 @@ public BlockReportResponseProto answer(InvocationOnMock invocation) {
     // back up to prior version and check DN sends old-style BR
     request.set(null);
     nsInfo.setCapabilities(Capability.UNKNOWN.getMask());
-    nn.blockReport(reg, "pool", sbr,
+    BlockListAsLongs blockList = getBlockList(r);
+    StorageBlockReport[] obp = new StorageBlockReport[] {
+        new StorageBlockReport(new DatanodeStorage("s1"), blockList) };
+    nn.blockReport(reg, "pool", obp,
         new BlockReportContext(1, 0, System.nanoTime(), 0L, true));
     proto = request.get();
     assertNotNull(proto);