gt:           if (!datanode.isRestarting()) {
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 static org.apache.hadoop.hdfs.server.datanode.DataNode.DN_CLIENTTRACE_FORMAT;
  
  import java.io.BufferedOutputStream;
  import java.io.Closeable;
  import java.io.DataInputStream;
  import java.io.DataOutputStream;
  import java.io.EOFException;
  import java.io.FileDescriptor;
  import java.io.FileOutputStream;
  import java.io.IOException;
  import java.io.OutputStream;
  import java.io.OutputStreamWriter;
  import java.io.Writer;
  import java.nio.ByteBuffer;
  import java.util.Arrays;
  import java.util.LinkedList;
  import java.util.zip.Checksum;
  
  import org.apache.commons.logging.Log;
  import org.apache.hadoop.fs.ChecksumException;
  import org.apache.hadoop.fs.FSOutputSummer;
  import org.apache.hadoop.fs.StorageType;
  import org.apache.hadoop.hdfs.DFSUtilClient;
  import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
  import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
  import org.apache.hadoop.hdfs.protocol.datatransfer.BlockConstructionStage;
  import org.apache.hadoop.hdfs.protocol.datatransfer.PacketHeader;
  import org.apache.hadoop.hdfs.protocol.datatransfer.PacketReceiver;
  import org.apache.hadoop.hdfs.protocol.datatransfer.PipelineAck;
  import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.BlockOpResponseProto;
  import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.Status;
  import org.apache.hadoop.hdfs.server.datanode.fsdataset.ReplicaInputStreams;
  import org.apache.hadoop.hdfs.server.datanode.fsdataset.ReplicaOutputStreams;
  import org.apache.hadoop.hdfs.server.protocol.DatanodeRegistration;
  import org.apache.hadoop.hdfs.util.DataTransferThrottler;
  import org.apache.hadoop.io.IOUtils;
  import org.apache.hadoop.io.nativeio.NativeIO;
  import org.apache.hadoop.util.Daemon;
  import org.apache.hadoop.util.DataChecksum;
  import org.apache.hadoop.util.StringUtils;
  import org.apache.hadoop.util.Time;
  import org.apache.htrace.core.Span;
  import org.apache.htrace.core.Tracer;
  
  import static org.apache.hadoop.io.nativeio.NativeIO.POSIX.POSIX_FADV_DONTNEED;
  import static org.apache.hadoop.io.nativeio.NativeIO.POSIX.SYNC_FILE_RANGE_WRITE;
  
  import com.google.common.annotations.VisibleForTesting;
  import org.slf4j.Logger;
  
  /** A class that receives a block and writes to its own disk, meanwhile
   * may copies it to another site. If a throttler is provided,
   * streaming throttling is also supported.
   **/
  class BlockReceiver implements Closeable {
    public static final Logger LOG = DataNode.LOG;
    static final Log ClientTraceLog = DataNode.ClientTraceLog;
  
    @VisibleForTesting
    static long CACHE_DROP_LAG_BYTES = 8 * 1024 * 1024;
    private final long datanodeSlowLogThresholdMs;
    private DataInputStream in = null; // from where data are read
    private DataChecksum clientChecksum; // checksum used by client
    private DataChecksum diskChecksum; // checksum we write to disk
    
    /**
     * In the case that the client is writing with a different
     * checksum polynomial than the block is stored with on disk,
     * the DataNode needs to recalculate checksums before writing.
     */
    private final boolean needsChecksumTranslation;
    private OutputStream out = null; // to block file at local disk
    private FileDescriptor outFd;
    private DataOutputStream checksumOut = null; // to crc file at local disk
    private final int bytesPerChecksum;
    private final int checksumSize;
    
    private final PacketReceiver packetReceiver = new PacketReceiver(false);
    
    protected final String inAddr;
    protected final String myAddr;
    private String mirrorAddr;
    private DataOutputStream mirrorOut;
    private Daemon responder = null;
    private DataTransferThrottler throttler;
    private ReplicaOutputStreams streams;
    private DatanodeInfo srcDataNode = null;
    private final DataNode datanode;
    volatile private boolean mirrorError;
  
    // Cache management state
    private boolean dropCacheBehindWrites;
    private long lastCacheManagementOffset = 0;
    private boolean syncBehindWrites;
    private boolean syncBehindWritesInBackground;
  
    /** The client name.  It is empty if a datanode is the client */
    private final String clientname;
    private final boolean isClient; 
    private final boolean isDatanode;
  
    /** the block to receive */
    private final ExtendedBlock block; 
    /** the replica to write */
    private ReplicaInPipeline replicaInfo;
    /** pipeline stage */
    private final BlockConstructionStage stage;
    private final boolean isTransfer;
  
    private boolean syncOnClose;
    private long restartBudget;
    /** the reference of the volume where the block receiver writes to */
    private ReplicaHandler replicaHandler;
  
    /**
     * for replaceBlock response
     */
    private final long responseInterval;
    private long lastResponseTime = 0;
    private boolean isReplaceBlock = false;
    private DataOutputStream replyOut = null;
    private long maxWriteToDiskMs = 0;
    
    private boolean pinning;
    private long lastSentTime;
    private long maxSendIdleTime;
  
    BlockReceiver(final ExtendedBlock block, final StorageType storageType,
        final DataInputStream in,
        final String inAddr, final String myAddr,
        final BlockConstructionStage stage, 
        final long newGs, final long minBytesRcvd, final long maxBytesRcvd, 
        final String clientname, final DatanodeInfo srcDataNode,
        final DataNode datanode, DataChecksum requestedChecksum,
        CachingStrategy cachingStrategy,
        final boolean allowLazyPersist,
        final boolean pinning) throws IOException {
      try{
        this.block = block;
        this.in = in;
        this.inAddr = inAddr;
        this.myAddr = myAddr;
        this.srcDataNode = srcDataNode;
        this.datanode = datanode;
  
        this.clientname = clientname;
        this.isDatanode = clientname.length() == 0;
        this.isClient = !this.isDatanode;
        this.restartBudget = datanode.getDnConf().restartReplicaExpiry;
        this.datanodeSlowLogThresholdMs = datanode.getDnConf().datanodeSlowIoWarningThresholdMs;
        // For replaceBlock() calls response should be sent to avoid socketTimeout
        // at clients. So sending with the interval of 0.5 * socketTimeout
        final long readTimeout = datanode.getDnConf().socketTimeout;
        this.responseInterval = (long) (readTimeout * 0.5);
        //for datanode, we have
        //1: clientName.length() == 0, and
        //2: stage == null or PIPELINE_SETUP_CREATE
        this.stage = stage;
        this.isTransfer = stage == BlockConstructionStage.TRANSFER_RBW
            || stage == BlockConstructionStage.TRANSFER_FINALIZED;
  
        this.pinning = pinning;
        this.lastSentTime = Time.monotonicNow();
        // Downstream will timeout in readTimeout on receiving the next packet.
        // If there is no data traffic, a heartbeat packet is sent at
        // the interval of 0.5*readTimeout. Here, we set 0.9*readTimeout to be
        // the threshold for detecting congestion.
        this.maxSendIdleTime = (long) (readTimeout * 0.9);
        if (LOG.isDebugEnabled()) {
          LOG.debug(getClass().getSimpleName() + ": " + block
              + "\n storageType=" + storageType + ", inAddr=" + inAddr
              + ", myAddr=" + myAddr + "\n stage=" + stage + ", newGs=" + newGs
              + ", minBytesRcvd=" + minBytesRcvd
              + ", maxBytesRcvd=" + maxBytesRcvd + "\n clientname=" + clientname
              + ", srcDataNode=" + srcDataNode
              + ", datanode=" + datanode.getDisplayName()
              + "\n requestedChecksum=" + requestedChecksum
              + "\n cachingStrategy=" + cachingStrategy
              + "\n allowLazyPersist=" + allowLazyPersist + ", pinning=" + pinning
              + ", isClient=" + isClient + ", isDatanode=" + isDatanode
              + ", responseInterval=" + responseInterval
          );
        }
  
        //
        // Open local disk out
        //
        if (isDatanode) { //replication or move
          replicaHandler = datanode.data.createTemporary(storageType, block);
        } else {
          switch (stage) {
          case PIPELINE_SETUP_CREATE:
            replicaHandler = datanode.data.createRbw(storageType, block, allowLazyPersist);
            datanode.notifyNamenodeReceivingBlock(
                block, replicaHandler.getReplica().getStorageUuid());
            break;
          case PIPELINE_SETUP_STREAMING_RECOVERY:
            replicaHandler = datanode.data.recoverRbw(
                block, newGs, minBytesRcvd, maxBytesRcvd);
            block.setGenerationStamp(newGs);
            break;
          case PIPELINE_SETUP_APPEND:
            replicaHandler = datanode.data.append(block, newGs, minBytesRcvd);
            block.setGenerationStamp(newGs);
            datanode.notifyNamenodeReceivingBlock(
                block, replicaHandler.getReplica().getStorageUuid());
            break;
          case PIPELINE_SETUP_APPEND_RECOVERY:
            replicaHandler = datanode.data.recoverAppend(block, newGs, minBytesRcvd);
            block.setGenerationStamp(newGs);
            datanode.notifyNamenodeReceivingBlock(
                block, replicaHandler.getReplica().getStorageUuid());
            break;
          case TRANSFER_RBW:
          case TRANSFER_FINALIZED:
            // this is a transfer destination
            replicaHandler =
                datanode.data.createTemporary(storageType, block);
            break;
          default: throw new IOException("Unsupported stage " + stage + 
                " while receiving block " + block + " from " + inAddr);
          }
        }
        replicaInfo = replicaHandler.getReplica();
        this.dropCacheBehindWrites = (cachingStrategy.getDropBehind() == null) ?
          datanode.getDnConf().dropCacheBehindWrites :
            cachingStrategy.getDropBehind();
        this.syncBehindWrites = datanode.getDnConf().syncBehindWrites;
        this.syncBehindWritesInBackground = datanode.getDnConf().
            syncBehindWritesInBackground;
        
        final boolean isCreate = isDatanode || isTransfer 
            || stage == BlockConstructionStage.PIPELINE_SETUP_CREATE;
        streams = replicaInfo.createStreams(isCreate, requestedChecksum);
        assert streams != null : "null streams!";
  
        // read checksum meta information
        this.clientChecksum = requestedChecksum;
        this.diskChecksum = streams.getChecksum();
        this.needsChecksumTranslation = !clientChecksum.equals(diskChecksum);
        this.bytesPerChecksum = diskChecksum.getBytesPerChecksum();
        this.checksumSize = diskChecksum.getChecksumSize();
  
        this.out = streams.getDataOut();
        if (out instanceof FileOutputStream) {
          this.outFd = ((FileOutputStream)out).getFD();
        } else {
          LOG.warn("Could not get file descriptor for outputstream of class " +
              out.getClass());
        }
        this.checksumOut = new DataOutputStream(new BufferedOutputStream(
            streams.getChecksumOut(), DFSUtilClient.getSmallBufferSize(
            datanode.getConf())));
        // write data chunk header if creating a new replica
        if (isCreate) {
          BlockMetadataHeader.writeHeader(checksumOut, diskChecksum);
        } 
      } catch (ReplicaAlreadyExistsException bae) {
        throw bae;
      } catch (ReplicaNotFoundException bne) {
        throw bne;
      } catch(IOException ioe) {
        if (replicaInfo != null) {
          replicaInfo.releaseAllBytesReserved();
        }
        IOUtils.closeStream(this);
        cleanupBlock();
        
        // check if there is a disk error
        IOException cause = DatanodeUtil.getCauseIfDiskError(ioe);
        DataNode.LOG.warn("IOException in BlockReceiver constructor"
            + (cause == null ? "" : ". Cause is "), cause);
        
        if (cause != null) { // possible disk error
          ioe = cause;
          datanode.checkDiskErrorAsync();
        }
        
        throw ioe;
      }
    }
  
    /** Return the datanode object. */
    DataNode getDataNode() {return datanode;}
  
    Replica getReplica() {
      return replicaInfo;
    }
  
    /**
     * close files and release volume reference.
     */
    @Override
    public void close() throws IOException {
      Span span = Tracer.getCurrentSpan();
      if (span != null) {
        span.addKVAnnotation("maxWriteToDiskMs",
              Long.toString(maxWriteToDiskMs));
      }
      packetReceiver.close();
  
      IOException ioe = null;
      if (syncOnClose && (out != null || checksumOut != null)) {
        datanode.metrics.incrFsyncCount();      
      }
      long flushTotalNanos = 0;
      boolean measuredFlushTime = false;
      // close checksum file
      try {
        if (checksumOut != null) {
          long flushStartNanos = System.nanoTime();
          checksumOut.flush();
          long flushEndNanos = System.nanoTime();
          if (syncOnClose) {
            long fsyncStartNanos = flushEndNanos;
            streams.syncChecksumOut();
            datanode.metrics.addFsyncNanos(System.nanoTime() - fsyncStartNanos);
          }
          flushTotalNanos += flushEndNanos - flushStartNanos;
          measuredFlushTime = true;
          checksumOut.close();
          checksumOut = null;
        }
      } catch(IOException e) {
        ioe = e;
      }
      finally {
        IOUtils.closeStream(checksumOut);
      }
      // close block file
      try {
        if (out != null) {
          long flushStartNanos = System.nanoTime();
          out.flush();
          long flushEndNanos = System.nanoTime();
          if (syncOnClose) {
            long fsyncStartNanos = flushEndNanos;
            streams.syncDataOut();
            datanode.metrics.addFsyncNanos(System.nanoTime() - fsyncStartNanos);
          }
          flushTotalNanos += flushEndNanos - flushStartNanos;
          measuredFlushTime = true;
          out.close();
          out = null;
        }
      } catch (IOException e) {
        ioe = e;
      }
      finally{
        IOUtils.closeStream(out);
      }
      if (replicaHandler != null) {
        IOUtils.cleanup(null, replicaHandler);
        replicaHandler = null;
      }
      if (measuredFlushTime) {
        datanode.metrics.addFlushNanos(flushTotalNanos);
      }
      // disk check
      if(ioe != null) {
        datanode.checkDiskErrorAsync();
        throw ioe;
      }
    }
  
    synchronized void setLastSentTime(long sentTime) {
      lastSentTime = sentTime;
    }
  
    /**
     * It can return false if
     * - upstream did not send packet for a long time
     * - a packet was received but got stuck in local disk I/O.
     * - a packet was received but got stuck on send to mirror.
     */
    synchronized boolean packetSentInTime() {
      long diff = Time.monotonicNow() - lastSentTime;
      if (diff > maxSendIdleTime) {
        LOG.info("A packet was last sent " + diff + " milliseconds ago.");
        return false;
      }
      return true;
    }
  
    /**
     * Flush block data and metadata files to disk.
     * @throws IOException
     */
    void flushOrSync(boolean isSync) throws IOException {
      long flushTotalNanos = 0;
      long begin = Time.monotonicNow();
      if (checksumOut != null) {
        long flushStartNanos = System.nanoTime();
        checksumOut.flush();
        long flushEndNanos = System.nanoTime();
        if (isSync) {
          long fsyncStartNanos = flushEndNanos;
          streams.syncChecksumOut();
          datanode.metrics.addFsyncNanos(System.nanoTime() - fsyncStartNanos);
        }
        flushTotalNanos += flushEndNanos - flushStartNanos;
      }
      if (out != null) {
        long flushStartNanos = System.nanoTime();
        out.flush();
        long flushEndNanos = System.nanoTime();
        if (isSync) {
          long fsyncStartNanos = flushEndNanos;
          streams.syncDataOut();
          datanode.metrics.addFsyncNanos(System.nanoTime() - fsyncStartNanos);
        }
        flushTotalNanos += flushEndNanos - flushStartNanos;
      }
      if (checksumOut != null || out != null) {
        datanode.metrics.addFlushNanos(flushTotalNanos);
        if (isSync) {
      	  datanode.metrics.incrFsyncCount();      
        }
      }
      long duration = Time.monotonicNow() - begin;
      if (duration > datanodeSlowLogThresholdMs) {
        LOG.warn("Slow flushOrSync took " + duration + "ms (threshold="
            + datanodeSlowLogThresholdMs + "ms), isSync:" + isSync + ", flushTotalNanos="
            + flushTotalNanos + "ns");
      }
    }
  
    /**
     * While writing to mirrorOut, failure to write to mirror should not
     * affect this datanode unless it is caused by interruption.
     */
    private void handleMirrorOutError(IOException ioe) throws IOException {
      String bpid = block.getBlockPoolId();
      LOG.info(datanode.getDNRegistrationForBP(bpid)
          + ":Exception writing " + block + " to mirror " + mirrorAddr, ioe);
      if (Thread.interrupted()) { // shut down if the thread is interrupted
        throw ioe;
      } else { // encounter an error while writing to mirror
        // continue to run even if can not write to mirror
        // notify client of the error
        // and wait for the client to shut down the pipeline
        mirrorError = true;
      }
    }
    
    /**
     * Verify multiple CRC chunks. 
     */
    private void verifyChunks(ByteBuffer dataBuf, ByteBuffer checksumBuf)
        throws IOException {
      try {
        clientChecksum.verifyChunkedSums(dataBuf, checksumBuf, clientname, 0);
      } catch (ChecksumException ce) {
        PacketHeader header = packetReceiver.getHeader();
        String specificOffset = "specific offsets are:"
            + " offsetInBlock = " + header.getOffsetInBlock()
            + " offsetInPacket = " + ce.getPos();
        LOG.warn("Checksum error in block "
            + block + " from " + inAddr
            + ", " + specificOffset, ce);
        // No need to report to namenode when client is writing.
        if (srcDataNode != null && isDatanode) {
          try {
            LOG.info("report corrupt " + block + " from datanode " +
                      srcDataNode + " to namenode");
            datanode.reportRemoteBadBlock(srcDataNode, block);
          } catch (IOException e) {
            LOG.warn("Failed to report bad " + block + 
                      " from datanode " + srcDataNode + " to namenode");
          }
        }
        throw new IOException("Unexpected checksum mismatch while writing "
            + block + " from " + inAddr);
      }
    }
    
      
    /**
     * Translate CRC chunks from the client's checksum implementation
     * to the disk checksum implementation.
     * 
     * This does not verify the original checksums, under the assumption
     * that they have already been validated.
     */
    private void translateChunks(ByteBuffer dataBuf, ByteBuffer checksumBuf) {
      diskChecksum.calculateChunkedSums(dataBuf, checksumBuf);
    }
  
    /** 
     * 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.
     */
    private int receivePacket() throws IOException {
      // read the next packet
      packetReceiver.receiveNextPacket(in);
  
      PacketHeader header = packetReceiver.getHeader();
      if (LOG.isDebugEnabled()){
        LOG.debug("Receiving one packet for block " + block +
                  ": " + header);
      }
  
      // Sanity check the header
      if (header.getOffsetInBlock() > replicaInfo.getNumBytes()) {
        throw new IOException("Received an out-of-sequence packet for " + block + 
            "from " + inAddr + " at offset " + header.getOffsetInBlock() +
            ". Expecting packet starting at " + replicaInfo.getNumBytes());
      }
      if (header.getDataLen() < 0) {
        throw new IOException("Got wrong length during writeBlock(" + block + 
                              ") from " + inAddr + " at offset " + 
                              header.getOffsetInBlock() + ": " +
                              header.getDataLen()); 
      }
  
      long offsetInBlock = header.getOffsetInBlock();
      long seqno = header.getSeqno();
      boolean lastPacketInBlock = header.isLastPacketInBlock();
      final int len = header.getDataLen();
      boolean syncBlock = header.getSyncBlock();
  
      // avoid double sync'ing on close
      if (syncBlock && lastPacketInBlock) {
        this.syncOnClose = false;
      }
  
      // update received bytes
      final long firstByteInBlock = offsetInBlock;
      offsetInBlock += len;
      if (replicaInfo.getNumBytes() < offsetInBlock) {
        replicaInfo.setNumBytes(offsetInBlock);
      }
      
      // put in queue for pending acks, unless sync was requested
      if (responder != null && !syncBlock && !shouldVerifyChecksum()) {
        ((PacketResponder) responder.getRunnable()).enqueue(seqno,
            lastPacketInBlock, offsetInBlock, Status.SUCCESS);
      }
  
      // Drop heartbeat for testing.
      if (seqno < 0 && len == 0 &&
          DataNodeFaultInjector.get().dropHeartbeatPacket()) {
        return 0;
      }
  
      //First write the packet to the mirror:
      if (mirrorOut != null && !mirrorError) {
        try {
          long begin = Time.monotonicNow();
          // For testing. Normally no-op.
          DataNodeFaultInjector.get().stopSendingPacketDownstream();
          packetReceiver.mirrorPacketTo(mirrorOut);
          mirrorOut.flush();
          long now = Time.monotonicNow();
          setLastSentTime(now);
          long duration = now - begin;
          if (duration > datanodeSlowLogThresholdMs) {
            LOG.warn("Slow BlockReceiver write packet to mirror took " + duration
                + "ms (threshold=" + datanodeSlowLogThresholdMs + "ms)");
          }
        } catch (IOException e) {
          handleMirrorOutError(e);
        }
      }
      
      ByteBuffer dataBuf = packetReceiver.getDataSlice();
      ByteBuffer checksumBuf = packetReceiver.getChecksumSlice();
      
      if (lastPacketInBlock || len == 0) {
        if(LOG.isDebugEnabled()) {
          LOG.debug("Receiving an empty packet or the end of the block " + block);
        }
        // sync block if requested
        if (syncBlock) {
          flushOrSync(true);
        }
      } else {
        final int checksumLen = diskChecksum.getChecksumSize(len);
        final int checksumReceivedLen = checksumBuf.capacity();
  
        if (checksumReceivedLen > 0 && checksumReceivedLen != checksumLen) {
          throw new IOException("Invalid checksum length: received length is "
              + checksumReceivedLen + " but expected length is " + checksumLen);
        }
  
        if (checksumReceivedLen > 0 && shouldVerifyChecksum()) {
          try {
            verifyChunks(dataBuf, checksumBuf);
          } 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 e) { }
            }
            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.
            translateChunks(dataBuf, checksumBuf);
          }
        }
  
        if (checksumReceivedLen == 0 && !streams.isTransientStorage()) {
          // checksum is missing, need to calculate it
          checksumBuf = ByteBuffer.allocate(checksumLen);
          diskChecksum.calculateChunkedSums(dataBuf, checksumBuf);
        }
        
        // by this point, the data in the buffer uses the disk checksum
  
        final boolean shouldNotWriteChecksum = checksumReceivedLen == 0
            && streams.isTransientStorage();
        try {
          long onDiskLen = replicaInfo.getBytesOnDisk();
          if (onDiskLen<offsetInBlock) {
            // Normally the beginning of an incoming packet is aligned with the
            // existing data on disk. If the beginning packet data offset is not
            // checksum chunk aligned, the end of packet will not go beyond the
            // next chunk boundary.
            // When a failure-recovery is involved, the client state and the
            // the datanode state may not exactly agree. I.e. the client may
            // resend part of data that is already on disk. Correct number of
            // bytes should be skipped when writing the data and checksum
            // buffers out to disk.
            long partialChunkSizeOnDisk = onDiskLen % bytesPerChecksum;
            long lastChunkBoundary = onDiskLen - partialChunkSizeOnDisk;
            boolean alignedOnDisk = partialChunkSizeOnDisk == 0;
            boolean alignedInPacket = firstByteInBlock % bytesPerChecksum == 0;
  
            // If the end of the on-disk data is not chunk-aligned, the last
            // checksum needs to be overwritten.
            boolean overwriteLastCrc = !alignedOnDisk && !shouldNotWriteChecksum;
            // If the starting offset of the packat data is at the last chunk
            // boundary of the data on disk, the partial checksum recalculation
            // can be skipped and the checksum supplied by the client can be used
            // instead. This reduces disk reads and cpu load.
            boolean doCrcRecalc = overwriteLastCrc &&
                (lastChunkBoundary != firstByteInBlock);
  
            // If this is a partial chunk, then verify that this is the only
            // chunk in the packet. If the starting offset is not chunk
            // aligned, the packet should terminate at or before the next
            // chunk boundary.
            if (!alignedInPacket && len > bytesPerChecksum) {
              throw new IOException("Unexpected packet data length for "
                  +  block + " from " + inAddr + ": a partial chunk must be "
                  + " sent in an individual packet (data length = " + len
                  +  " > bytesPerChecksum = " + bytesPerChecksum + ")");
            }
  
            // If the last portion of the block file is not a full chunk,
            // then read in pre-existing partial data chunk and recalculate
            // the checksum so that the checksum calculation can continue
            // from the right state. If the client provided the checksum for
            // the whole chunk, this is not necessary.
            Checksum partialCrc = null;
            if (doCrcRecalc) {
              if (LOG.isDebugEnabled()) {
                LOG.debug("receivePacket for " + block 
                    + ": previous write did not end at the chunk boundary."
                    + " onDiskLen=" + onDiskLen);
              }
              long offsetInChecksum = BlockMetadataHeader.getHeaderSize() +
                  onDiskLen / bytesPerChecksum * checksumSize;
              partialCrc = computePartialChunkCrc(onDiskLen, offsetInChecksum);
            }
  
            // The data buffer position where write will begin. If the packet
            // data and on-disk data have no overlap, this will not be at the
            // beginning of the buffer.
            int startByteToDisk = (int)(onDiskLen-firstByteInBlock) 
                + dataBuf.arrayOffset() + dataBuf.position();
  
            // Actual number of data bytes to write.
            int numBytesToDisk = (int)(offsetInBlock-onDiskLen);
            
            // Write data to disk.
            long begin = Time.monotonicNow();
            out.write(dataBuf.array(), startByteToDisk, numBytesToDisk);
            long duration = Time.monotonicNow() - begin;
            if (duration > maxWriteToDiskMs) {
              maxWriteToDiskMs = duration;
            }
            if (duration > datanodeSlowLogThresholdMs) {
              LOG.warn("Slow BlockReceiver write data to disk cost:" + duration
                  + "ms (threshold=" + datanodeSlowLogThresholdMs + "ms)");
            }
  
            final byte[] lastCrc;
            if (shouldNotWriteChecksum) {
              lastCrc = null;
            } else {
              int skip = 0;
              byte[] crcBytes = null;
  
              // First, prepare to overwrite the partial crc at the end.
              if (overwriteLastCrc) { // not chunk-aligned on disk
                // prepare to overwrite last checksum
                adjustCrcFilePosition();
              }
  
              // The CRC was recalculated for the last partial chunk. Update the
              // CRC by reading the rest of the chunk, then write it out.
              if (doCrcRecalc) {
                // Calculate new crc for this chunk.
                int bytesToReadForRecalc =
                    (int)(bytesPerChecksum - partialChunkSizeOnDisk);
                if (numBytesToDisk < bytesToReadForRecalc) {
                  bytesToReadForRecalc = numBytesToDisk;
                }
  
                partialCrc.update(dataBuf.array(), startByteToDisk,
                    bytesToReadForRecalc);
                byte[] buf = FSOutputSummer.convertToByteStream(partialCrc,
                    checksumSize);
                crcBytes = copyLastChunkChecksum(buf, checksumSize, buf.length);
                checksumOut.write(buf);
                if(LOG.isDebugEnabled()) {
                  LOG.debug("Writing out partial crc for data len " + len +
                      ", skip=" + skip);
                }
                skip++; //  For the partial chunk that was just read.
              }
  
              // Determine how many checksums need to be skipped up to the last
              // boundary. The checksum after the boundary was already counted
              // above. Only count the number of checksums skipped up to the
              // boundary here.
              long skippedDataBytes = lastChunkBoundary - firstByteInBlock;
  
              if (skippedDataBytes > 0) {
                skip += (int)(skippedDataBytes / bytesPerChecksum) +
                    ((skippedDataBytes % bytesPerChecksum == 0) ? 0 : 1);
              }
              skip *= checksumSize; // Convert to number of bytes
  
              // write the rest of checksum
              final int offset = checksumBuf.arrayOffset() +
                  checksumBuf.position() + skip;
              final int end = offset + checksumLen - skip;
              // If offset >= end, there is no more checksum to write.
              // I.e. a partial chunk checksum rewrite happened and there is no
              // more to write after that.
              if (offset >= end && doCrcRecalc) {
                lastCrc = crcBytes;
              } else {
                final int remainingBytes = checksumLen - skip;
                lastCrc = copyLastChunkChecksum(checksumBuf.array(),
                    checksumSize, end);
                checksumOut.write(checksumBuf.array(), offset, remainingBytes);
              }
            }
  
            /// flush entire packet, sync if requested
            flushOrSync(syncBlock);
            
            replicaInfo.setLastChecksumAndDataLen(offsetInBlock, lastCrc);
  
            datanode.metrics.incrBytesWritten(len);
            datanode.metrics.incrTotalWriteTime(duration);
  
            manageWriterOsCache(offsetInBlock);
          }
        } catch (IOException iex) {
          datanode.checkDiskErrorAsync();
          throw iex;
        }
      }
  
      // if sync was requested, put in queue for pending acks here
      // (after the fsync finished)
      if (responder != null && (syncBlock || shouldVerifyChecksum())) {
        ((PacketResponder) responder.getRunnable()).enqueue(seqno,
            lastPacketInBlock, offsetInBlock, Status.SUCCESS);
      }
  
      /*
       * Send in-progress responses for the replaceBlock() calls back to caller to
       * avoid timeouts due to balancer throttling. HDFS-6247
       */
      if (isReplaceBlock
          && (Time.monotonicNow() - lastResponseTime > responseInterval)) {
        BlockOpResponseProto.Builder response = BlockOpResponseProto.newBuilder()
            .setStatus(Status.IN_PROGRESS);
        response.build().writeDelimitedTo(replyOut);
        replyOut.flush();
  
        lastResponseTime = Time.monotonicNow();
      }
  
      if (throttler != null) { // throttle I/O
        throttler.throttle(len);
      }
      
      return lastPacketInBlock?-1:len;
    }
  
    private static byte[] copyLastChunkChecksum(byte[] array, int size, int end) {
      return Arrays.copyOfRange(array, end - size, end);
    }
  
    private void manageWriterOsCache(long offsetInBlock) {
      try {
        if (outFd != null &&
            offsetInBlock > lastCacheManagementOffset + CACHE_DROP_LAG_BYTES) {
          long begin = Time.monotonicNow();
          //
          // For SYNC_FILE_RANGE_WRITE, we want to sync from
          // lastCacheManagementOffset to a position "two windows ago"
          //
          //                         <========= sync ===========>
          // +-----------------------O--------------------------X
          // start                  last                      curPos
          // of file                 
          //
          if (syncBehindWrites) {
            if (syncBehindWritesInBackground) {
              this.datanode.getFSDataset().submitBackgroundSyncFileRangeRequest(
                  block, outFd, lastCacheManagementOffset,
                  offsetInBlock - lastCacheManagementOffset,
                  SYNC_FILE_RANGE_WRITE);
            } else {
              NativeIO.POSIX.syncFileRangeIfPossible(outFd,
                  lastCacheManagementOffset,
                  offsetInBlock - lastCacheManagementOffset,
                  SYNC_FILE_RANGE_WRITE);
            }
          }
          //
          // For POSIX_FADV_DONTNEED, we want to drop from the beginning 
          // of the file to a position prior to the current position.
          //
          // <=== drop =====> 
          //                 <---W--->
          // +--------------+--------O--------------------------X
          // start        dropPos   last                      curPos
          // of file             
          //                     
          long dropPos = lastCacheManagementOffset - CACHE_DROP_LAG_BYTES;
          if (dropPos > 0 && dropCacheBehindWrites) {
            NativeIO.POSIX.getCacheManipulator().posixFadviseIfPossible(
                block.getBlockName(), outFd, 0, dropPos, POSIX_FADV_DONTNEED);
          }
          lastCacheManagementOffset = offsetInBlock;
          long duration = Time.monotonicNow() - begin;
          if (duration > datanodeSlowLogThresholdMs) {
            LOG.warn("Slow manageWriterOsCache took " + duration
                + "ms (threshold=" + datanodeSlowLogThresholdMs + "ms)");
          }
        }
      } catch (Throwable t) {
        LOG.warn("Error managing cache for writer of block " + block, t);
      }
    }
    
    public void sendOOB() throws IOException, InterruptedException {
      if (isDatanode) {
        return;
      }
      ((PacketResponder) responder.getRunnable()).sendOOBResponse(PipelineAck
          .getRestartOOBStatus());
    }
    
    void receiveBlock(
        DataOutputStream mirrOut, // output to next datanode
        DataInputStream mirrIn,   // input from next datanode
        DataOutputStream replyOut,  // output to previous datanode
        String mirrAddr, DataTransferThrottler throttlerArg,
        DatanodeInfo[] downstreams,
        boolean isReplaceBlock) throws IOException {
  
        syncOnClose = datanode.getDnConf().syncOnClose;
        boolean responderClosed = false;
        mirrorOut = mirrOut;
        mirrorAddr = mirrAddr;
        throttler = throttlerArg;
  
        this.replyOut = replyOut;
        this.isReplaceBlock = isReplaceBlock;
  
      try {
        if (isClient && !isTransfer) {
          responder = new Daemon(datanode.threadGroup, 
              new PacketResponder(replyOut, mirrIn, downstreams));
          responder.start(); // start thread to processes responses
        }
  
        while (receivePacket() >= 0) { /* Receive until the last packet */ }
  
        // wait for all outstanding packet responses. And then
        // indicate responder to gracefully shutdown.
        // Mark that responder has been closed for future processing
        if (responder != null) {
          ((PacketResponder)responder.getRunnable()).close();
          responderClosed = true;
        }
  
        // If this write is for a replication or transfer-RBW/Finalized,
        // then finalize block or convert temporary to RBW.
        // For client-writes, the block is finalized in the PacketResponder.
        if (isDatanode || isTransfer) {
          // Hold a volume reference to finalize block.
          try (ReplicaHandler handler = claimReplicaHandler()) {
            // close the block/crc files
            close();
            block.setNumBytes(replicaInfo.getNumBytes());
  
            if (stage == BlockConstructionStage.TRANSFER_RBW) {
              // for TRANSFER_RBW, convert temporary to RBW
              datanode.data.convertTemporaryToRbw(block);
            } else {
              // for isDatnode or TRANSFER_FINALIZED
              // Finalize the block.
              datanode.data.finalizeBlock(block);
            }
          }
          datanode.metrics.incrBlocksWritten();
        }
  
      } catch (IOException ioe) {
        replicaInfo.releaseAllBytesReserved();
        if (datanode.isRestarting()) {
          // Do not throw if shutting down for restart. Otherwise, it will cause
          // premature termination of responder.
          LOG.info("Shutting down for restart (" + block + ").");
        } else {
          LOG.info("Exception for " + block, ioe);
          throw ioe;
        }
      } finally {
        // Clear the previous interrupt state of this thread.
        Thread.interrupted();
  
        // If a shutdown for restart was initiated, upstream needs to be notified.
        // There is no need to do anything special if the responder was closed
        // normally.
        if (!responderClosed) { // Data transfer was not complete.
          if (responder != null) {
            // In case this datanode is shutting down for quick restart,
            // send a special ack upstream.
            if (datanode.isRestarting() && isClient && !isTransfer) {
              try (Writer out = new OutputStreamWriter(
                  replicaInfo.createRestartMetaStream(), "UTF-8")) {
                // write out the current time.
                out.write(Long.toString(Time.now() + restartBudget));
                out.flush();
              } catch (IOException ioe) {
                // The worst case is not recovering this RBW replica. 
                // Client will fall back to regular pipeline recovery.
              } finally {
                IOUtils.closeStream(out);
              }
              try {              
                // Even if the connection is closed after the ack packet is
                // flushed, the client can react to the connection closure 
                // first. Insert a delay to lower the chance of client 
                // missing the OOB ack.
                Thread.sleep(1000);
              } catch (InterruptedException ie) {
                // It is already going down. Ignore this.
              }
            }
            responder.interrupt();
          }
          IOUtils.closeStream(this);
          cleanupBlock();
        }
        if (responder != null) {
          try {
            responder.interrupt();
            // join() on the responder should timeout a bit earlier than the
            // configured deadline. Otherwise, the join() on this thread will
            // likely timeout as well.
            long joinTimeout = datanode.getDnConf().getXceiverStopTimeout();
            joinTimeout = joinTimeout > 1  ? joinTimeout*8/10 : joinTimeout;
            responder.join(joinTimeout);
            if (responder.isAlive()) {
              String msg = "Join on responder thread " + responder
                  + " timed out";
              LOG.warn(msg + "\n" + StringUtils.getStackTrace(responder));
              throw new IOException(msg);
            }
          } catch (InterruptedException e) {
            responder.interrupt();
            // do not throw if shutting down for restart.
  