HDFS-3875. Issue handling checksum errors in write pipeline. Contributed by Kihwal Lee.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-0.23@1484811 13f79535-47bb-0310-9956-ffa450edef68

hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt

@@ -26,6 +26,8 @@ Release 0.23.8 - UNRELEASED
     HDFS-4835. Port trunk WebHDFS changes to branch-0.23. Includes backport
     of HDFS-4805 and HADOOP-9549. (Robert Parker via kihwal)
 
+    HDFS-3875. Issue handling checksum errors in write pipeline. (kihwal)
+
 Release 0.23.7 - 2013-04-18
 
   INCOMPATIBLE CHANGES

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSClientFaultInjector.java

@@ -0,0 +1,45 @@
+/**
+ * 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;
+import java.io.IOException;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+
+/**
+ * Used for injecting faults in DFSClient and DFSOutputStream tests.
+ * Calls into this are a no-op in production code. 
+ */
+@VisibleForTesting
+@InterfaceAudience.Private
+public class DFSClientFaultInjector {
+  public static DFSClientFaultInjector instance = new DFSClientFaultInjector();
+
+  public static DFSClientFaultInjector get() {
+    return instance;
+  }
+
+  public boolean corruptPacket() {
+    return false;
+  }
+
+  public boolean uncorruptPacket() {
+    return false;
+  }
+}

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSOutputStream.java

@@ -292,6 +292,9 @@ class DFSOutputStream extends FSOutputSummer implements Syncable {
     private ArrayList<DatanodeInfo> excludedNodes = new ArrayList<DatanodeInfo>();
     volatile boolean hasError = false;
     volatile int errorIndex = -1;
+    /** The last ack sequence number before pipeline failure. */
+    private long lastAckedSeqnoBeforeFailure = -1;
+    private int pipelineRecoveryCount = 0;
     private BlockConstructionStage stage;  // block construction stage
     private long bytesSent = 0; // number of bytes that've been sent
 
@@ -506,10 +509,22 @@ class DFSOutputStream extends FSOutputSummer implements Syncable {
                 " sending packet " + one);
           }
 
+          int cIdx = 0;
+          if (DFSClientFaultInjector.get().corruptPacket()) {
+            // flip a byte
+            cIdx = buf.position()+buf.remaining()-1;
+            buf.array()[cIdx] ^= 0xff;
+          }
+
           // write out data to remote datanode
           blockStream.write(buf.array(), buf.position(), buf.remaining());
           blockStream.flush();
           lastPacket = System.currentTimeMillis();
+
+          if (DFSClientFaultInjector.get().uncorruptPacket()) {
+            // flip back before retransmission
+            buf.array()[cIdx] ^= 0xff;
+          }
           
           if (one.isHeartbeatPacket()) {  //heartbeat packet
           }
@@ -738,6 +753,24 @@ class DFSOutputStream extends FSOutputSummer implements Syncable {
         ackQueue.clear();
       }
 
+      // Record the new pipeline failure recovery.
+      if (lastAckedSeqnoBeforeFailure != lastAckedSeqno) {
+         lastAckedSeqnoBeforeFailure = lastAckedSeqno;
+         pipelineRecoveryCount = 1;
+      } else {
+        // If we had to recover the pipeline five times in a row for the
+        // same packet, this client likely has corrupt data or corrupting
+        // during transmission.
+        if (++pipelineRecoveryCount > 5) {
+          DFSClient.LOG.warn("Error recovering pipeline for writing " +
+              block + ". Already retried 5 times for the same packet.");
+          lastException = new IOException("Failing write. Tried pipeline " +
+              "recovery 5 times without success.");
+          streamerClosed = true;
+          return false;
+        }
+      }
+
       boolean doSleep = setupPipelineForAppendOrRecovery();
       
       if (!streamerClosed && dfsClient.clientRunning) {

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

@@ -327,7 +327,8 @@ class BlockReceiver implements Closeable {
       clientChecksum.update(dataBuf, dataOff, chunkLen);
 
       if (!clientChecksum.compare(checksumBuf, checksumOff)) {
-        if (srcDataNode != null) {
+        // No need to report to namenode when client is writing.
+        if (srcDataNode != null && isDatanode) {
           try {
             LOG.info("report corrupt block " + block + " from datanode " +
                       srcDataNode + " to namenode");
@@ -503,6 +504,19 @@ class BlockReceiver implements Closeable {
     }
   }
   
+  /** Check whether checksum needs to be verified. 
+   * Skip verifying checksum iff this is not the last one in the 
+   * pipeline and clientName is non-null. i.e. Checksum is verified
+   * on all the datanodes when the data is being written by a 
+   * datanode rather than a client. Whe client is writing the data, 
+   * protocol includes acks and only the last datanode needs to verify 
+   * checksum.
+   * @return true if checksum verification is needed, otherwise false.
+   */
+  private boolean shouldVerifyChecksum() {
+    return (mirrorOut == null || isDatanode || needsChecksumTranslation);
+  }
+
   /** 
    * Receives and processes a packet. It can contain many chunks.
    * returns the number of data bytes that the packet has.
@@ -567,9 +581,9 @@ class BlockReceiver implements Closeable {
     }
     
     // put in queue for pending acks
-    if (responder != null) {
+    if (responder != null && !shouldVerifyChecksum()) {
       ((PacketResponder)responder.getRunnable()).enqueue(seqno,
-                                      lastPacketInBlock, offsetInBlock); 
+          lastPacketInBlock, offsetInBlock, Status.SUCCESS); 
     }  
 
     //First write the packet to the mirror:
@@ -605,15 +619,24 @@ class BlockReceiver implements Closeable {
 
       buf.position(buf.limit()); // move to the end of the data.
 
-      /* skip verifying checksum iff this is not the last one in the 
-       * pipeline and clientName is non-null. i.e. Checksum is verified
-       * on all the datanodes when the data is being written by a 
-       * datanode rather than a client. Whe client is writing the data, 
-       * protocol includes acks and only the last datanode needs to verify 
-       * checksum.
-       */
-      if (mirrorOut == null || isDatanode || needsChecksumTranslation) {
-        verifyChunks(pktBuf, dataOff, len, pktBuf, checksumOff);
+      if (shouldVerifyChecksum()) {
+        try {
+          verifyChunks(pktBuf, dataOff, len, pktBuf, checksumOff);
+        } catch (IOException ioe) {
+          // checksum error detected locally. there is no reason to continue.
+          if (responder != null) {
+            try {
+              ((PacketResponder) responder.getRunnable()).enqueue(seqno,
+                  lastPacketInBlock, offsetInBlock,
+                  Status.ERROR_CHECKSUM);
+              // Wait until the responder sends back the response
+              // and interrupt this thread.
+              Thread.sleep(3000);
+            } catch (InterruptedException ie) { }
+          }
+          throw new IOException("Terminating due to a checksum error." + ioe);
+        }
+
         if (needsChecksumTranslation) {
           // overwrite the checksums in the packet buffer with the
           // appropriate polynomial for the disk storage.
@@ -695,6 +718,11 @@ class BlockReceiver implements Closeable {
       }
     }
 
+    if (responder != null && shouldVerifyChecksum()) {
+        ((PacketResponder) responder.getRunnable()).enqueue(seqno,
+          lastPacketInBlock, offsetInBlock, Status.SUCCESS);
+    }
+
     if (throttler != null) { // throttle I/O
       throttler.throttle(len);
     }
@@ -890,7 +918,7 @@ class BlockReceiver implements Closeable {
   }
   
   /**
-   * Processed responses from downstream datanodes in the pipeline
+   * Processes responses from downstream datanodes in the pipeline
    * and sends back replies to the originator.
    */
   class PacketResponder implements Runnable, Closeable {   
@@ -943,9 +971,11 @@ class BlockReceiver implements Closeable {
      * @param offsetInBlock
      */
     synchronized void enqueue(final long seqno,
-        final boolean lastPacketInBlock, final long offsetInBlock) {
+        final boolean lastPacketInBlock,
+        final long offsetInBlock, final Status ackStatus) {
       if (running) {
-        final Packet p = new Packet(seqno, lastPacketInBlock, offsetInBlock);
+        final Packet p = new Packet(seqno, lastPacketInBlock, offsetInBlock,
+            ackStatus);
         if(LOG.isDebugEnabled()) {
           LOG.debug(myString + ": enqueue " + p);
         }
@@ -1071,20 +1101,31 @@ class BlockReceiver implements Closeable {
               }
             }
 
+            Status myStatus = pkt == null ? Status.SUCCESS : pkt.ackStatus;
             // construct my ack message
             Status[] replies = null;
             if (mirrorError) { // ack read error
               replies = new Status[2];
-              replies[0] = Status.SUCCESS;
+              replies[0] = myStatus;
               replies[1] = Status.ERROR;
             } else {
               short ackLen = type == PacketResponderType.LAST_IN_PIPELINE? 0
                   : ack.getNumOfReplies();
               replies = new Status[1+ackLen];
-              replies[0] = Status.SUCCESS;
+              replies[0] = myStatus;
               for (int i=0; i<ackLen; i++) {
                 replies[i+1] = ack.getReply(i);
               }
+              // If the mirror has reported that it received a corrupt packet,
+              // do self-destruct to mark myself bad, instead of the mirror node.
+              // The mirror is guaranteed to be good without corrupt data.
+              if (ackLen > 0 && replies[1] == Status.ERROR_CHECKSUM) {
+                running = false;
+                removeAckHead();
+                LOG.warn("Shutting down writer and responder due to a checksum error.");
+                receiverThread.interrupt();
+                continue;
+              }
             }
             PipelineAck replyAck = new PipelineAck(expected, replies);
             
@@ -1103,6 +1144,14 @@ class BlockReceiver implements Closeable {
               removeAckHead();
               // update bytes acked
             }
+            // terminate after sending response if this node detected 
+            // a checksum error
+            if (myStatus == Status.ERROR_CHECKSUM) {
+              running = false;
+              LOG.warn("Shutting down writer and responder due to a checksum error.");
+              receiverThread.interrupt();
+              continue;
+            }
         } catch (IOException e) {
           LOG.warn("IOException in BlockReceiver.run(): ", e);
           if (running) {
@@ -1146,11 +1195,14 @@ class BlockReceiver implements Closeable {
     final long seqno;
     final boolean lastPacketInBlock;
     final long offsetInBlock;
+    final Status ackStatus;
 
-    Packet(long seqno, boolean lastPacketInBlock, long offsetInBlock) {
+    Packet(long seqno, boolean lastPacketInBlock,
+        long offsetInBlock, Status ackStatus) {
       this.seqno = seqno;
       this.lastPacketInBlock = lastPacketInBlock;
       this.offsetInBlock = offsetInBlock;
+      this.ackStatus = ackStatus;
     }
 
     @Override
@@ -1158,6 +1210,7 @@ class BlockReceiver implements Closeable {
       return getClass().getSimpleName() + "(seqno=" + seqno
         + ", lastPacketInBlock=" + lastPacketInBlock
         + ", offsetInBlock=" + offsetInBlock
+        + ", ackStatus=" + ackStatus
         + ")";
     }
   }

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

@@ -1655,13 +1655,25 @@ class FSDataset implements FSDatasetInterface {
     }
     
     // check replica length
-    if (rbw.getBytesAcked() < minBytesRcvd || rbw.getNumBytes() > maxBytesRcvd){
-      throw new ReplicaNotFoundException("Unmatched length replica " + 
-          replicaInfo + ": BytesAcked = " + rbw.getBytesAcked() + 
-          " BytesRcvd = " + rbw.getNumBytes() + " are not in the range of [" + 
+    long bytesAcked = rbw.getBytesAcked();
+    long numBytes = rbw.getNumBytes();
+    if (bytesAcked < minBytesRcvd || numBytes > maxBytesRcvd){
+      throw new ReplicaNotFoundException("Unmatched length replica " +
+          replicaInfo + ": BytesAcked = " + bytesAcked +
+          " BytesRcvd = " + numBytes + " are not in the range of [" +
           minBytesRcvd + ", " + maxBytesRcvd + "].");
     }
 
+    // Truncate the potentially corrupt portion.
+    // If the source was client and the last node in the pipeline was lost,
+    // any corrupt data written after the acked length can go unnoticed. 
+    if (numBytes > bytesAcked) {
+      final File replicafile = rbw.getBlockFile();
+      truncateBlock(replicafile, rbw.getMetaFile(), numBytes, bytesAcked);
+      rbw.setNumBytes(bytesAcked);
+      rbw.setLastChecksumAndDataLen(bytesAcked, null);
+    }
+
     // bump the replica's generation stamp to newGS
     bumpReplicaGS(rbw, newGS);
     

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestCrcCorruption.java

@@ -25,15 +25,21 @@ import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
 import java.util.Random;
 
+import org.junit.Before;
 import org.junit.Test;
 import static org.junit.Assert.*;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
 import org.apache.hadoop.io.IOUtils;
 
+import org.mockito.Mockito;
+import org.mockito.stubbing.Answer;
+
 /**
  * A JUnit test for corrupted file handling.
  * This test creates a bunch of files/directories with replication 
@@ -62,6 +68,77 @@ import org.apache.hadoop.io.IOUtils;
  *     replica was created from the non-corrupted replica.
  */
 public class TestCrcCorruption {
+  private DFSClientFaultInjector faultInjector;
+
+  @Before
+  public void setUp() throws IOException {
+    faultInjector = Mockito.mock(DFSClientFaultInjector.class);
+    DFSClientFaultInjector.instance = faultInjector;
+  }
+
+  /** 
+   * Test case for data corruption during data transmission for
+   * create/write. To recover from corruption while writing, at
+   * least two replicas are needed.
+   */
+  @Test
+  public void testCorruptionDuringWrt() throws Exception {
+    Configuration conf = new HdfsConfiguration();
+    MiniDFSCluster cluster = null;
+
+    try {
+      cluster = new MiniDFSCluster.Builder(conf).numDataNodes(10).build();
+      cluster.waitActive();
+      FileSystem fs = cluster.getFileSystem();
+      Path file = new Path("/test_corruption_file");
+      FSDataOutputStream out = fs.create(file, true, 8192, (short)3, (long)(128*1024*1024));
+      byte[] data = new byte[65536];
+      for (int i=0; i < 65536; i++) {
+        data[i] = (byte)(i % 256);
+      }
+
+      for (int i = 0; i < 5; i++) {
+        out.write(data, 0, 65535);
+      }
+      out.hflush();
+      // corrupt the packet once
+      Mockito.when(faultInjector.corruptPacket()).thenReturn(true, false);
+      Mockito.when(faultInjector.uncorruptPacket()).thenReturn(true, false);
+
+      for (int i = 0; i < 5; i++) {
+        out.write(data, 0, 65535);
+      }
+      out.close();
+      // read should succeed
+      FSDataInputStream in = fs.open(file);
+      for(int c; (c = in.read()) != -1; );
+      in.close();
+
+      // test the retry limit
+      out = fs.create(file, true, 8192, (short)3, (long)(128*1024*1024));
+
+      // corrupt the packet once and never fix it.
+      Mockito.when(faultInjector.corruptPacket()).thenReturn(true, false);
+      Mockito.when(faultInjector.uncorruptPacket()).thenReturn(false);
+
+      // the client should give up pipeline reconstruction after retries.
+      try {
+        for (int i = 0; i < 5; i++) {
+          out.write(data, 0, 65535);
+        }
+        out.close();
+        fail("Write did not fail");
+      } catch (IOException ioe) {
+        // we should get an ioe
+        DFSClient.LOG.info("Got expected exception", ioe);
+      }
+    } finally {
+      if (cluster != null) { cluster.shutdown(); }
+      Mockito.when(faultInjector.corruptPacket()).thenReturn(false);
+      Mockito.when(faultInjector.uncorruptPacket()).thenReturn(false);
+    }
+  }
+
   /** 
    * check if DFS can handle corrupted CRC blocks
    */