HDFS-7854. Separate class DataStreamer out of DFSOutputStream. Contributed by Li Bo.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/a16bfff7 Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/a16bfff7 Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/a16bfff7 Branch: refs/heads/trunk Commit: a16bfff71bd7f00e06e1f59bfe5445a154bb8c66 Parents: 570a83a Author: Jing Zhao <ji...@apache.org> Authored: Tue Mar 24 11:06:13 2015 -0700 Committer: Jing Zhao <ji...@apache.org> Committed: Tue Mar 24 11:06:13 2015 -0700 ---------------------------------------------------------------------- hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt | 3 + .../dev-support/findbugsExcludeFile.xml | 2 +- .../org/apache/hadoop/hdfs/DFSOutputStream.java | 1694 ++--------------- .../org/apache/hadoop/hdfs/DataStreamer.java | 1754 ++++++++++++++++++ .../org/apache/hadoop/hdfs/DFSTestUtil.java | 2 +- .../apache/hadoop/hdfs/TestDFSOutputStream.java | 5 +- .../apache/hadoop/hdfs/TestFileCreation.java | 18 +- 7 files changed, 1893 insertions(+), 1585 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hadoop/blob/a16bfff7/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt index 5dae029..4ec0891 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt +++ b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt @@ -332,6 +332,9 @@ Release 2.8.0 - UNRELEASED HDFS-7829. Code clean up for LocatedBlock. (Takanobu Asanuma via jing9) + HDFS-7854. Separate class DataStreamer out of DFSOutputStream. (Li Bo via + jing9) + OPTIMIZATIONS BUG FIXES http://git-wip-us.apache.org/repos/asf/hadoop/blob/a16bfff7/hadoop-hdfs-project/hadoop-hdfs/dev-support/findbugsExcludeFile.xml ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/dev-support/findbugsExcludeFile.xml b/hadoop-hdfs-project/hadoop-hdfs/dev-support/findbugsExcludeFile.xml index dedeece..224d2fb 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/dev-support/findbugsExcludeFile.xml +++ b/hadoop-hdfs-project/hadoop-hdfs/dev-support/findbugsExcludeFile.xml @@ -77,7 +77,7 @@ ResponseProccessor is thread that is designed to catch RuntimeException. --> <Match> - <Class name="org.apache.hadoop.hdfs.DFSOutputStream$DataStreamer$ResponseProcessor" /> + <Class name="org.apache.hadoop.hdfs.DataStreamer$ResponseProcessor" /> <Method name="run" /> <Bug pattern="REC_CATCH_EXCEPTION" /> </Match> http://git-wip-us.apache.org/repos/asf/hadoop/blob/a16bfff7/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSOutputStream.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSOutputStream.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSOutputStream.java index d7d59af..ee3e6f6 100755 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSOutputStream.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSOutputStream.java @@ -17,29 +17,12 @@ */ package org.apache.hadoop.hdfs; -import static org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.Status.SUCCESS; - -import java.io.BufferedOutputStream; -import java.io.DataInputStream; -import java.io.DataOutputStream; import java.io.FileNotFoundException; import java.io.IOException; -import java.io.InputStream; import java.io.InterruptedIOException; -import java.io.OutputStream; -import java.net.InetAddress; -import java.net.InetSocketAddress; import java.net.Socket; import java.nio.channels.ClosedChannelException; -import java.util.ArrayList; -import java.util.Arrays; import java.util.EnumSet; -import java.util.HashSet; -import java.util.LinkedList; -import java.util.List; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import org.apache.hadoop.HadoopIllegalArgumentException; @@ -52,64 +35,37 @@ import org.apache.hadoop.fs.FileAlreadyExistsException; import org.apache.hadoop.fs.FileEncryptionInfo; import org.apache.hadoop.fs.ParentNotDirectoryException; import org.apache.hadoop.fs.permission.FsPermission; -import org.apache.hadoop.fs.StorageType; import org.apache.hadoop.fs.Syncable; import org.apache.hadoop.hdfs.client.HdfsDataOutputStream; import org.apache.hadoop.hdfs.client.HdfsDataOutputStream.SyncFlag; -import org.apache.hadoop.hdfs.protocol.BlockStoragePolicy; -import org.apache.hadoop.hdfs.protocol.QuotaExceededException; import org.apache.hadoop.hdfs.protocol.DSQuotaExceededException; import org.apache.hadoop.hdfs.protocol.DatanodeInfo; import org.apache.hadoop.hdfs.protocol.ExtendedBlock; -import org.apache.hadoop.hdfs.protocol.HdfsConstants; import org.apache.hadoop.hdfs.protocol.HdfsFileStatus; import org.apache.hadoop.hdfs.protocol.LocatedBlock; import org.apache.hadoop.hdfs.protocol.NSQuotaExceededException; import org.apache.hadoop.hdfs.protocol.SnapshotAccessControlException; import org.apache.hadoop.hdfs.protocol.UnresolvedPathException; -import org.apache.hadoop.hdfs.protocol.datatransfer.BlockConstructionStage; -import org.apache.hadoop.hdfs.protocol.datatransfer.DataTransferProtocol; -import org.apache.hadoop.hdfs.protocol.datatransfer.DataTransferProtoUtil; -import org.apache.hadoop.hdfs.protocol.datatransfer.IOStreamPair; -import org.apache.hadoop.hdfs.protocol.datatransfer.InvalidEncryptionKeyException; import org.apache.hadoop.hdfs.protocol.datatransfer.PacketHeader; -import org.apache.hadoop.hdfs.protocol.datatransfer.PipelineAck; -import org.apache.hadoop.hdfs.protocol.datatransfer.Sender; -import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.BlockOpResponseProto; -import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.Status; -import org.apache.hadoop.hdfs.protocolPB.PBHelper; import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier; -import org.apache.hadoop.hdfs.server.blockmanagement.BlockStoragePolicySuite; import org.apache.hadoop.hdfs.server.datanode.CachingStrategy; -import org.apache.hadoop.hdfs.server.namenode.NotReplicatedYetException; import org.apache.hadoop.hdfs.server.namenode.RetryStartFileException; import org.apache.hadoop.hdfs.server.namenode.SafeModeException; import org.apache.hadoop.hdfs.util.ByteArrayManager; import org.apache.hadoop.io.EnumSetWritable; -import org.apache.hadoop.io.IOUtils; import org.apache.hadoop.ipc.RemoteException; -import org.apache.hadoop.net.NetUtils; import org.apache.hadoop.security.AccessControlException; import org.apache.hadoop.security.token.Token; -import org.apache.hadoop.util.Daemon; import org.apache.hadoop.util.DataChecksum; import org.apache.hadoop.util.DataChecksum.Type; import org.apache.hadoop.util.Progressable; import org.apache.hadoop.util.Time; -import org.apache.htrace.NullScope; import org.apache.htrace.Sampler; -import org.apache.htrace.Span; import org.apache.htrace.Trace; -import org.apache.htrace.TraceInfo; import org.apache.htrace.TraceScope; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; -import com.google.common.cache.CacheBuilder; -import com.google.common.cache.CacheLoader; -import com.google.common.cache.LoadingCache; -import com.google.common.cache.RemovalListener; -import com.google.common.cache.RemovalNotification; /**************************************************************** @@ -121,19 +77,11 @@ import com.google.common.cache.RemovalNotification; * is typically 512 bytes and has an associated checksum with it. * * When a client application fills up the currentPacket, it is - * enqueued into dataQueue. The DataStreamer thread picks up - * packets from the dataQueue, sends it to the first datanode in - * the pipeline and moves it from the dataQueue to the ackQueue. - * The ResponseProcessor receives acks from the datanodes. When an - * successful ack for a packet is received from all datanodes, the - * ResponseProcessor removes the corresponding packet from the - * ackQueue. + * enqueued into the dataQueue of DataStreamer. DataStreamer is a + * thread that picks up packets from the dataQueue and sends it to + * the first datanode in the pipeline. * - * In case of error, all outstanding packets and moved from - * ackQueue. A new pipeline is setup by eliminating the bad - * datanode from the original pipeline. The DataStreamer now - * starts sending packets from the dataQueue. -****************************************************************/ + ****************************************************************/ @InterfaceAudience.Private public class DFSOutputStream extends FSOutputSummer implements Syncable, CanSetDropBehind { @@ -148,45 +96,25 @@ public class DFSOutputStream extends FSOutputSummer CryptoProtocolVersion.supported(); private final DFSClient dfsClient; - private final long dfsclientSlowLogThresholdMs; private final ByteArrayManager byteArrayManager; - private Socket s; // closed is accessed by different threads under different locks. private volatile boolean closed = false; - private String src; + private final String src; private final long fileId; private final long blockSize; - /** Only for DataTransferProtocol.writeBlock(..) */ - private final DataChecksum checksum4WriteBlock; - private final int bytesPerChecksum; + private final int bytesPerChecksum; - // both dataQueue and ackQueue are protected by dataQueue lock - private final LinkedList<DFSPacket> dataQueue = new LinkedList<DFSPacket>(); - private final LinkedList<DFSPacket> ackQueue = new LinkedList<DFSPacket>(); private DFSPacket currentPacket = null; private DataStreamer streamer; - private long currentSeqno = 0; - private long lastQueuedSeqno = -1; - private long lastAckedSeqno = -1; - private long bytesCurBlock = 0; // bytes written in current block private int packetSize = 0; // write packet size, not including the header. private int chunksPerPacket = 0; - private final AtomicReference<IOException> lastException = new AtomicReference<IOException>(); - private long artificialSlowdown = 0; private long lastFlushOffset = 0; // offset when flush was invoked - //persist blocks on namenode - private final AtomicBoolean persistBlocks = new AtomicBoolean(false); - private volatile boolean appendChunk = false; // appending to existing partial block private long initialFileSize = 0; // at time of file open - private final Progressable progress; private final short blockReplication; // replication factor of file private boolean shouldSyncBlock = false; // force blocks to disk upon close private final AtomicReference<CachingStrategy> cachingStrategy; - private boolean failPacket = false; private FileEncryptionInfo fileEncryptionInfo; - private static final BlockStoragePolicySuite blockStoragePolicySuite = - BlockStoragePolicySuite.createDefaultSuite(); /** Use {@link ByteArrayManager} to create buffer for non-heartbeat packets.*/ private DFSPacket createPacket(int packetSize, int chunksPerPkt, long offsetInBlock, @@ -207,1326 +135,10 @@ public class DFSOutputStream extends FSOutputSummer getChecksumSize(), lastPacketInBlock); } - /** - * For heartbeat packets, create buffer directly by new byte[] - * since heartbeats should not be blocked. - */ - private DFSPacket createHeartbeatPacket() throws InterruptedIOException { - final byte[] buf = new byte[PacketHeader.PKT_MAX_HEADER_LEN]; - return new DFSPacket(buf, 0, 0, DFSPacket.HEART_BEAT_SEQNO, - getChecksumSize(), false); - } - - - // - // The DataStreamer class is responsible for sending data packets to the - // datanodes in the pipeline. It retrieves a new blockid and block locations - // from the namenode, and starts streaming packets to the pipeline of - // Datanodes. Every packet has a sequence number associated with - // it. When all the packets for a block are sent out and acks for each - // if them are received, the DataStreamer closes the current block. - // - class DataStreamer extends Daemon { - private volatile boolean streamerClosed = false; - private ExtendedBlock block; // its length is number of bytes acked - private Token<BlockTokenIdentifier> accessToken; - private DataOutputStream blockStream; - private DataInputStream blockReplyStream; - private ResponseProcessor response = null; - private volatile DatanodeInfo[] nodes = null; // list of targets for current block - private volatile StorageType[] storageTypes = null; - private volatile String[] storageIDs = null; - private final LoadingCache<DatanodeInfo, DatanodeInfo> excludedNodes = - CacheBuilder.newBuilder() - .expireAfterWrite( - dfsClient.getConf().excludedNodesCacheExpiry, - TimeUnit.MILLISECONDS) - .removalListener(new RemovalListener<DatanodeInfo, DatanodeInfo>() { - @Override - public void onRemoval( - RemovalNotification<DatanodeInfo, DatanodeInfo> notification) { - DFSClient.LOG.info("Removing node " + - notification.getKey() + " from the excluded nodes list"); - } - }) - .build(new CacheLoader<DatanodeInfo, DatanodeInfo>() { - @Override - public DatanodeInfo load(DatanodeInfo key) throws Exception { - return key; - } - }); - private String[] favoredNodes; - volatile boolean hasError = false; - volatile int errorIndex = -1; - // Restarting node index - AtomicInteger restartingNodeIndex = new AtomicInteger(-1); - private long restartDeadline = 0; // Deadline of DN restart - private BlockConstructionStage stage; // block construction stage - private long bytesSent = 0; // number of bytes that've been sent - private final boolean isLazyPersistFile; - - /** Nodes have been used in the pipeline before and have failed. */ - private final List<DatanodeInfo> failed = new ArrayList<DatanodeInfo>(); - /** The last ack sequence number before pipeline failure. */ - private long lastAckedSeqnoBeforeFailure = -1; - private int pipelineRecoveryCount = 0; - /** Has the current block been hflushed? */ - private boolean isHflushed = false; - /** Append on an existing block? */ - private final boolean isAppend; - - private DataStreamer(HdfsFileStatus stat, ExtendedBlock block) { - isAppend = false; - isLazyPersistFile = isLazyPersist(stat); - this.block = block; - stage = BlockConstructionStage.PIPELINE_SETUP_CREATE; - } - - /** - * Construct a data streamer for appending to the last partial block - * @param lastBlock last block of the file to be appended - * @param stat status of the file to be appended - * @param bytesPerChecksum number of bytes per checksum - * @throws IOException if error occurs - */ - private DataStreamer(LocatedBlock lastBlock, HdfsFileStatus stat, - int bytesPerChecksum) throws IOException { - isAppend = true; - stage = BlockConstructionStage.PIPELINE_SETUP_APPEND; - block = lastBlock.getBlock(); - bytesSent = block.getNumBytes(); - accessToken = lastBlock.getBlockToken(); - isLazyPersistFile = isLazyPersist(stat); - long usedInLastBlock = stat.getLen() % blockSize; - int freeInLastBlock = (int)(blockSize - usedInLastBlock); - - // calculate the amount of free space in the pre-existing - // last crc chunk - int usedInCksum = (int)(stat.getLen() % bytesPerChecksum); - int freeInCksum = bytesPerChecksum - usedInCksum; - - // if there is space in the last block, then we have to - // append to that block - if (freeInLastBlock == blockSize) { - throw new IOException("The last block for file " + - src + " is full."); - } - - if (usedInCksum > 0 && freeInCksum > 0) { - // if there is space in the last partial chunk, then - // setup in such a way that the next packet will have only - // one chunk that fills up the partial chunk. - // - computePacketChunkSize(0, freeInCksum); - setChecksumBufSize(freeInCksum); - appendChunk = true; - } else { - // if the remaining space in the block is smaller than - // that expected size of of a packet, then create - // smaller size packet. - // - computePacketChunkSize(Math.min(dfsClient.getConf().writePacketSize, freeInLastBlock), - bytesPerChecksum); - } - - // setup pipeline to append to the last block XXX retries?? - setPipeline(lastBlock); - errorIndex = -1; // no errors yet. - if (nodes.length < 1) { - throw new IOException("Unable to retrieve blocks locations " + - " for last block " + block + - "of file " + src); - - } - } - - private void setPipeline(LocatedBlock lb) { - setPipeline(lb.getLocations(), lb.getStorageTypes(), lb.getStorageIDs()); - } - private void setPipeline(DatanodeInfo[] nodes, StorageType[] storageTypes, - String[] storageIDs) { - this.nodes = nodes; - this.storageTypes = storageTypes; - this.storageIDs = storageIDs; - } - - private void setFavoredNodes(String[] favoredNodes) { - this.favoredNodes = favoredNodes; - } - - /** - * Initialize for data streaming - */ - private void initDataStreaming() { - this.setName("DataStreamer for file " + src + - " block " + block); - response = new ResponseProcessor(nodes); - response.start(); - stage = BlockConstructionStage.DATA_STREAMING; - } - - private void endBlock() { - if(DFSClient.LOG.isDebugEnabled()) { - DFSClient.LOG.debug("Closing old block " + block); - } - this.setName("DataStreamer for file " + src); - closeResponder(); - closeStream(); - setPipeline(null, null, null); - stage = BlockConstructionStage.PIPELINE_SETUP_CREATE; - } - - /* - * streamer thread is the only thread that opens streams to datanode, - * and closes them. Any error recovery is also done by this thread. - */ - @Override - public void run() { - long lastPacket = Time.monotonicNow(); - TraceScope scope = NullScope.INSTANCE; - while (!streamerClosed && dfsClient.clientRunning) { - // if the Responder encountered an error, shutdown Responder - if (hasError && response != null) { - try { - response.close(); - response.join(); - response = null; - } catch (InterruptedException e) { - DFSClient.LOG.warn("Caught exception ", e); - } - } - - DFSPacket one; - try { - // process datanode IO errors if any - boolean doSleep = false; - if (hasError && (errorIndex >= 0 || restartingNodeIndex.get() >= 0)) { - doSleep = processDatanodeError(); - } - - synchronized (dataQueue) { - // wait for a packet to be sent. - long now = Time.monotonicNow(); - while ((!streamerClosed && !hasError && dfsClient.clientRunning - && dataQueue.size() == 0 && - (stage != BlockConstructionStage.DATA_STREAMING || - stage == BlockConstructionStage.DATA_STREAMING && - now - lastPacket < dfsClient.getConf().socketTimeout/2)) || doSleep ) { - long timeout = dfsClient.getConf().socketTimeout/2 - (now-lastPacket); - timeout = timeout <= 0 ? 1000 : timeout; - timeout = (stage == BlockConstructionStage.DATA_STREAMING)? - timeout : 1000; - try { - dataQueue.wait(timeout); - } catch (InterruptedException e) { - DFSClient.LOG.warn("Caught exception ", e); - } - doSleep = false; - now = Time.monotonicNow(); - } - if (streamerClosed || hasError || !dfsClient.clientRunning) { - continue; - } - // get packet to be sent. - if (dataQueue.isEmpty()) { - one = createHeartbeatPacket(); - assert one != null; - } else { - one = dataQueue.getFirst(); // regular data packet - long parents[] = one.getTraceParents(); - if (parents.length > 0) { - scope = Trace.startSpan("dataStreamer", new TraceInfo(0, parents[0])); - // TODO: use setParents API once it's available from HTrace 3.2 -// scope = Trace.startSpan("dataStreamer", Sampler.ALWAYS); -// scope.getSpan().setParents(parents); - } - } - } - - // get new block from namenode. - if (stage == BlockConstructionStage.PIPELINE_SETUP_CREATE) { - if(DFSClient.LOG.isDebugEnabled()) { - DFSClient.LOG.debug("Allocating new block"); - } - setPipeline(nextBlockOutputStream()); - initDataStreaming(); - } else if (stage == BlockConstructionStage.PIPELINE_SETUP_APPEND) { - if(DFSClient.LOG.isDebugEnabled()) { - DFSClient.LOG.debug("Append to block " + block); - } - setupPipelineForAppendOrRecovery(); - initDataStreaming(); - } - - long lastByteOffsetInBlock = one.getLastByteOffsetBlock(); - if (lastByteOffsetInBlock > blockSize) { - throw new IOException("BlockSize " + blockSize + - " is smaller than data size. " + - " Offset of packet in block " + - lastByteOffsetInBlock + - " Aborting file " + src); - } - - if (one.isLastPacketInBlock()) { - // wait for all data packets have been successfully acked - synchronized (dataQueue) { - while (!streamerClosed && !hasError && - ackQueue.size() != 0 && dfsClient.clientRunning) { - try { - // wait for acks to arrive from datanodes - dataQueue.wait(1000); - } catch (InterruptedException e) { - DFSClient.LOG.warn("Caught exception ", e); - } - } - } - if (streamerClosed || hasError || !dfsClient.clientRunning) { - continue; - } - stage = BlockConstructionStage.PIPELINE_CLOSE; - } - - // send the packet - Span span = null; - synchronized (dataQueue) { - // move packet from dataQueue to ackQueue - if (!one.isHeartbeatPacket()) { - span = scope.detach(); - one.setTraceSpan(span); - dataQueue.removeFirst(); - ackQueue.addLast(one); - dataQueue.notifyAll(); - } - } - - if (DFSClient.LOG.isDebugEnabled()) { - DFSClient.LOG.debug("DataStreamer block " + block + - " sending packet " + one); - } - - // write out data to remote datanode - TraceScope writeScope = Trace.startSpan("writeTo", span); - try { - one.writeTo(blockStream); - blockStream.flush(); - } catch (IOException e) { - // HDFS-3398 treat primary DN is down since client is unable to - // write to primary DN. If a failed or restarting node has already - // been recorded by the responder, the following call will have no - // effect. Pipeline recovery can handle only one node error at a - // time. If the primary node fails again during the recovery, it - // will be taken out then. - tryMarkPrimaryDatanodeFailed(); - throw e; - } finally { - writeScope.close(); - } - lastPacket = Time.monotonicNow(); - - // update bytesSent - long tmpBytesSent = one.getLastByteOffsetBlock(); - if (bytesSent < tmpBytesSent) { - bytesSent = tmpBytesSent; - } - - if (streamerClosed || hasError || !dfsClient.clientRunning) { - continue; - } - - // Is this block full? - if (one.isLastPacketInBlock()) { - // wait for the close packet has been acked - synchronized (dataQueue) { - while (!streamerClosed && !hasError && - ackQueue.size() != 0 && dfsClient.clientRunning) { - dataQueue.wait(1000);// wait for acks to arrive from datanodes - } - } - if (streamerClosed || hasError || !dfsClient.clientRunning) { - continue; - } - - endBlock(); - } - if (progress != null) { progress.progress(); } - - // This is used by unit test to trigger race conditions. - if (artificialSlowdown != 0 && dfsClient.clientRunning) { - Thread.sleep(artificialSlowdown); - } - } catch (Throwable e) { - // Log warning if there was a real error. - if (restartingNodeIndex.get() == -1) { - // Since their messages are descriptive enough, do not always - // log a verbose stack-trace WARN for quota exceptions. - if (e instanceof QuotaExceededException) { - DFSClient.LOG.debug("DataStreamer Quota Exception", e); - } else { - DFSClient.LOG.warn("DataStreamer Exception", e); - } - } - if (e instanceof IOException) { - setLastException((IOException)e); - } else { - setLastException(new IOException("DataStreamer Exception: ",e)); - } - hasError = true; - if (errorIndex == -1 && restartingNodeIndex.get() == -1) { - // Not a datanode issue - streamerClosed = true; - } - } finally { - scope.close(); - } - } - closeInternal(); - } - - private void closeInternal() { - closeResponder(); // close and join - closeStream(); - streamerClosed = true; - setClosed(); - synchronized (dataQueue) { - dataQueue.notifyAll(); - } - } - - /* - * close both streamer and DFSOutputStream, should be called only - * by an external thread and only after all data to be sent has - * been flushed to datanode. - * - * Interrupt this data streamer if force is true - * - * @param force if this data stream is forced to be closed - */ - void close(boolean force) { - streamerClosed = true; - synchronized (dataQueue) { - dataQueue.notifyAll(); - } - if (force) { - this.interrupt(); - } - } - - private void closeResponder() { - if (response != null) { - try { - response.close(); - response.join(); - } catch (InterruptedException e) { - DFSClient.LOG.warn("Caught exception ", e); - } finally { - response = null; - } - } - } - - private void closeStream() { - if (blockStream != null) { - try { - blockStream.close(); - } catch (IOException e) { - setLastException(e); - } finally { - blockStream = null; - } - } - if (blockReplyStream != null) { - try { - blockReplyStream.close(); - } catch (IOException e) { - setLastException(e); - } finally { - blockReplyStream = null; - } - } - if (null != s) { - try { - s.close(); - } catch (IOException e) { - setLastException(e); - } finally { - s = null; - } - } - } - - // The following synchronized methods are used whenever - // errorIndex or restartingNodeIndex is set. This is because - // check & set needs to be atomic. Simply reading variables - // does not require a synchronization. When responder is - // not running (e.g. during pipeline recovery), there is no - // need to use these methods. - - /** Set the error node index. Called by responder */ - synchronized void setErrorIndex(int idx) { - errorIndex = idx; - } - - /** Set the restarting node index. Called by responder */ - synchronized void setRestartingNodeIndex(int idx) { - restartingNodeIndex.set(idx); - // If the data streamer has already set the primary node - // bad, clear it. It is likely that the write failed due to - // the DN shutdown. Even if it was a real failure, the pipeline - // recovery will take care of it. - errorIndex = -1; - } - - /** - * This method is used when no explicit error report was received, - * but something failed. When the primary node is a suspect or - * unsure about the cause, the primary node is marked as failed. - */ - synchronized void tryMarkPrimaryDatanodeFailed() { - // There should be no existing error and no ongoing restart. - if ((errorIndex == -1) && (restartingNodeIndex.get() == -1)) { - errorIndex = 0; - } - } - - /** - * Examine whether it is worth waiting for a node to restart. - * @param index the node index - */ - boolean shouldWaitForRestart(int index) { - // Only one node in the pipeline. - if (nodes.length == 1) { - return true; - } - - // Is it a local node? - InetAddress addr = null; - try { - addr = InetAddress.getByName(nodes[index].getIpAddr()); - } catch (java.net.UnknownHostException e) { - // we are passing an ip address. this should not happen. - assert false; - } - - if (addr != null && NetUtils.isLocalAddress(addr)) { - return true; - } - return false; - } - - // - // Processes responses from the datanodes. A packet is removed - // from the ackQueue when its response arrives. - // - private class ResponseProcessor extends Daemon { - - private volatile boolean responderClosed = false; - private DatanodeInfo[] targets = null; - private boolean isLastPacketInBlock = false; - - ResponseProcessor (DatanodeInfo[] targets) { - this.targets = targets; - } - - @Override - public void run() { - - setName("ResponseProcessor for block " + block); - PipelineAck ack = new PipelineAck(); - - TraceScope scope = NullScope.INSTANCE; - while (!responderClosed && dfsClient.clientRunning && !isLastPacketInBlock) { - // process responses from datanodes. - try { - // read an ack from the pipeline - long begin = Time.monotonicNow(); - ack.readFields(blockReplyStream); - long duration = Time.monotonicNow() - begin; - if (duration > dfsclientSlowLogThresholdMs - && ack.getSeqno() != DFSPacket.HEART_BEAT_SEQNO) { - DFSClient.LOG - .warn("Slow ReadProcessor read fields took " + duration - + "ms (threshold=" + dfsclientSlowLogThresholdMs + "ms); ack: " - + ack + ", targets: " + Arrays.asList(targets)); - } else if (DFSClient.LOG.isDebugEnabled()) { - DFSClient.LOG.debug("DFSClient " + ack); - } - - long seqno = ack.getSeqno(); - // processes response status from datanodes. - for (int i = ack.getNumOfReplies()-1; i >=0 && dfsClient.clientRunning; i--) { - final Status reply = PipelineAck.getStatusFromHeader(ack - .getReply(i)); - // Restart will not be treated differently unless it is - // the local node or the only one in the pipeline. - if (PipelineAck.isRestartOOBStatus(reply) && - shouldWaitForRestart(i)) { - restartDeadline = dfsClient.getConf().datanodeRestartTimeout - + Time.monotonicNow(); - setRestartingNodeIndex(i); - String message = "A datanode is restarting: " + targets[i]; - DFSClient.LOG.info(message); - throw new IOException(message); - } - // node error - if (reply != SUCCESS) { - setErrorIndex(i); // first bad datanode - throw new IOException("Bad response " + reply + - " for block " + block + - " from datanode " + - targets[i]); - } - } - - assert seqno != PipelineAck.UNKOWN_SEQNO : - "Ack for unknown seqno should be a failed ack: " + ack; - if (seqno == DFSPacket.HEART_BEAT_SEQNO) { // a heartbeat ack - continue; - } - - // a success ack for a data packet - DFSPacket one; - synchronized (dataQueue) { - one = ackQueue.getFirst(); - } - if (one.getSeqno() != seqno) { - throw new IOException("ResponseProcessor: Expecting seqno " + - " for block " + block + - one.getSeqno() + " but received " + seqno); - } - isLastPacketInBlock = one.isLastPacketInBlock(); - - // Fail the packet write for testing in order to force a - // pipeline recovery. - if (DFSClientFaultInjector.get().failPacket() && - isLastPacketInBlock) { - failPacket = true; - throw new IOException( - "Failing the last packet for testing."); - } - - // update bytesAcked - block.setNumBytes(one.getLastByteOffsetBlock()); - - synchronized (dataQueue) { - scope = Trace.continueSpan(one.getTraceSpan()); - one.setTraceSpan(null); - lastAckedSeqno = seqno; - ackQueue.removeFirst(); - dataQueue.notifyAll(); - - one.releaseBuffer(byteArrayManager); - } - } catch (Exception e) { - if (!responderClosed) { - if (e instanceof IOException) { - setLastException((IOException)e); - } - hasError = true; - // If no explicit error report was received, mark the primary - // node as failed. - tryMarkPrimaryDatanodeFailed(); - synchronized (dataQueue) { - dataQueue.notifyAll(); - } - if (restartingNodeIndex.get() == -1) { - DFSClient.LOG.warn("DFSOutputStream ResponseProcessor exception " - + " for block " + block, e); - } - responderClosed = true; - } - } finally { - scope.close(); - } - } - } - - void close() { - responderClosed = true; - this.interrupt(); - } - } - - // If this stream has encountered any errors so far, shutdown - // threads and mark stream as closed. Returns true if we should - // sleep for a while after returning from this call. - // - private boolean processDatanodeError() throws IOException { - if (response != null) { - DFSClient.LOG.info("Error Recovery for " + block + - " waiting for responder to exit. "); - return true; - } - closeStream(); - - // move packets from ack queue to front of the data queue - synchronized (dataQueue) { - dataQueue.addAll(0, ackQueue); - 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.set(new IOException("Failing write. Tried pipeline " + - "recovery 5 times without success.")); - streamerClosed = true; - return false; - } - } - boolean doSleep = setupPipelineForAppendOrRecovery(); - - if (!streamerClosed && dfsClient.clientRunning) { - if (stage == BlockConstructionStage.PIPELINE_CLOSE) { - - // If we had an error while closing the pipeline, we go through a fast-path - // where the BlockReceiver does not run. Instead, the DataNode just finalizes - // the block immediately during the 'connect ack' process. So, we want to pull - // the end-of-block packet from the dataQueue, since we don't actually have - // a true pipeline to send it over. - // - // We also need to set lastAckedSeqno to the end-of-block Packet's seqno, so that - // a client waiting on close() will be aware that the flush finished. - synchronized (dataQueue) { - DFSPacket endOfBlockPacket = dataQueue.remove(); // remove the end of block packet - Span span = endOfBlockPacket.getTraceSpan(); - if (span != null) { - // Close any trace span associated with this Packet - TraceScope scope = Trace.continueSpan(span); - scope.close(); - } - assert endOfBlockPacket.isLastPacketInBlock(); - assert lastAckedSeqno == endOfBlockPacket.getSeqno() - 1; - lastAckedSeqno = endOfBlockPacket.getSeqno(); - dataQueue.notifyAll(); - } - endBlock(); - } else { - initDataStreaming(); - } - } - - return doSleep; - } - - private void setHflush() { - isHflushed = true; - } - - private int findNewDatanode(final DatanodeInfo[] original - ) throws IOException { - if (nodes.length != original.length + 1) { - throw new IOException( - new StringBuilder() - .append("Failed to replace a bad datanode on the existing pipeline ") - .append("due to no more good datanodes being available to try. ") - .append("(Nodes: current=").append(Arrays.asList(nodes)) - .append(", original=").append(Arrays.asList(original)).append("). ") - .append("The current failed datanode replacement policy is ") - .append(dfsClient.dtpReplaceDatanodeOnFailure).append(", and ") - .append("a client may configure this via '") - .append(DFSConfigKeys.DFS_CLIENT_WRITE_REPLACE_DATANODE_ON_FAILURE_POLICY_KEY) - .append("' in its configuration.") - .toString()); - } - for(int i = 0; i < nodes.length; i++) { - int j = 0; - for(; j < original.length && !nodes[i].equals(original[j]); j++); - if (j == original.length) { - return i; - } - } - throw new IOException("Failed: new datanode not found: nodes=" - + Arrays.asList(nodes) + ", original=" + Arrays.asList(original)); - } - - private void addDatanode2ExistingPipeline() throws IOException { - if (DataTransferProtocol.LOG.isDebugEnabled()) { - DataTransferProtocol.LOG.debug("lastAckedSeqno = " + lastAckedSeqno); - } - /* - * Is data transfer necessary? We have the following cases. - * - * Case 1: Failure in Pipeline Setup - * - Append - * + Transfer the stored replica, which may be a RBW or a finalized. - * - Create - * + If no data, then no transfer is required. - * + If there are data written, transfer RBW. This case may happens - * when there are streaming failure earlier in this pipeline. - * - * Case 2: Failure in Streaming - * - Append/Create: - * + transfer RBW - * - * Case 3: Failure in Close - * - Append/Create: - * + no transfer, let NameNode replicates the block. - */ - if (!isAppend && lastAckedSeqno < 0 - && stage == BlockConstructionStage.PIPELINE_SETUP_CREATE) { - //no data have been written - return; - } else if (stage == BlockConstructionStage.PIPELINE_CLOSE - || stage == BlockConstructionStage.PIPELINE_CLOSE_RECOVERY) { - //pipeline is closing - return; - } - - //get a new datanode - final DatanodeInfo[] original = nodes; - final LocatedBlock lb = dfsClient.namenode.getAdditionalDatanode( - src, fileId, block, nodes, storageIDs, - failed.toArray(new DatanodeInfo[failed.size()]), - 1, dfsClient.clientName); - setPipeline(lb); - - //find the new datanode - final int d = findNewDatanode(original); - - //transfer replica - final DatanodeInfo src = d == 0? nodes[1]: nodes[d - 1]; - final DatanodeInfo[] targets = {nodes[d]}; - final StorageType[] targetStorageTypes = {storageTypes[d]}; - transfer(src, targets, targetStorageTypes, lb.getBlockToken()); - } - - private void transfer(final DatanodeInfo src, final DatanodeInfo[] targets, - final StorageType[] targetStorageTypes, - final Token<BlockTokenIdentifier> blockToken) throws IOException { - //transfer replica to the new datanode - Socket sock = null; - DataOutputStream out = null; - DataInputStream in = null; - try { - sock = createSocketForPipeline(src, 2, dfsClient); - final long writeTimeout = dfsClient.getDatanodeWriteTimeout(2); - - OutputStream unbufOut = NetUtils.getOutputStream(sock, writeTimeout); - InputStream unbufIn = NetUtils.getInputStream(sock); - IOStreamPair saslStreams = dfsClient.saslClient.socketSend(sock, - unbufOut, unbufIn, dfsClient, blockToken, src); - unbufOut = saslStreams.out; - unbufIn = saslStreams.in; - out = new DataOutputStream(new BufferedOutputStream(unbufOut, - HdfsConstants.SMALL_BUFFER_SIZE)); - in = new DataInputStream(unbufIn); - - //send the TRANSFER_BLOCK request - new Sender(out).transferBlock(block, blockToken, dfsClient.clientName, - targets, targetStorageTypes); - out.flush(); - - //ack - BlockOpResponseProto response = - BlockOpResponseProto.parseFrom(PBHelper.vintPrefixed(in)); - if (SUCCESS != response.getStatus()) { - throw new IOException("Failed to add a datanode"); - } - } finally { - IOUtils.closeStream(in); - IOUtils.closeStream(out); - IOUtils.closeSocket(sock); - } - } - - /** - * Open a DataOutputStream to a DataNode pipeline so that - * it can be written to. - * This happens when a file is appended or data streaming fails - * It keeps on trying until a pipeline is setup - */ - private boolean setupPipelineForAppendOrRecovery() throws IOException { - // check number of datanodes - if (nodes == null || nodes.length == 0) { - String msg = "Could not get block locations. " + "Source file \"" - + src + "\" - Aborting..."; - DFSClient.LOG.warn(msg); - setLastException(new IOException(msg)); - streamerClosed = true; - return false; - } - - boolean success = false; - long newGS = 0L; - while (!success && !streamerClosed && dfsClient.clientRunning) { - // Sleep before reconnect if a dn is restarting. - // This process will be repeated until the deadline or the datanode - // starts back up. - if (restartingNodeIndex.get() >= 0) { - // 4 seconds or the configured deadline period, whichever is shorter. - // This is the retry interval and recovery will be retried in this - // interval until timeout or success. - long delay = Math.min(dfsClient.getConf().datanodeRestartTimeout, - 4000L); - try { - Thread.sleep(delay); - } catch (InterruptedException ie) { - lastException.set(new IOException("Interrupted while waiting for " + - "datanode to restart. " + nodes[restartingNodeIndex.get()])); - streamerClosed = true; - return false; - } - } - boolean isRecovery = hasError; - // remove bad datanode from list of datanodes. - // If errorIndex was not set (i.e. appends), then do not remove - // any datanodes - // - if (errorIndex >= 0) { - StringBuilder pipelineMsg = new StringBuilder(); - for (int j = 0; j < nodes.length; j++) { - pipelineMsg.append(nodes[j]); - if (j < nodes.length - 1) { - pipelineMsg.append(", "); - } - } - if (nodes.length <= 1) { - lastException.set(new IOException("All datanodes " + pipelineMsg - + " are bad. Aborting...")); - streamerClosed = true; - return false; - } - DFSClient.LOG.warn("Error Recovery for block " + block + - " in pipeline " + pipelineMsg + - ": bad datanode " + nodes[errorIndex]); - failed.add(nodes[errorIndex]); - - DatanodeInfo[] newnodes = new DatanodeInfo[nodes.length-1]; - arraycopy(nodes, newnodes, errorIndex); - - final StorageType[] newStorageTypes = new StorageType[newnodes.length]; - arraycopy(storageTypes, newStorageTypes, errorIndex); - - final String[] newStorageIDs = new String[newnodes.length]; - arraycopy(storageIDs, newStorageIDs, errorIndex); - - setPipeline(newnodes, newStorageTypes, newStorageIDs); - - // Just took care of a node error while waiting for a node restart - if (restartingNodeIndex.get() >= 0) { - // If the error came from a node further away than the restarting - // node, the restart must have been complete. - if (errorIndex > restartingNodeIndex.get()) { - restartingNodeIndex.set(-1); - } else if (errorIndex < restartingNodeIndex.get()) { - // the node index has shifted. - restartingNodeIndex.decrementAndGet(); - } else { - // this shouldn't happen... - assert false; - } - } - - if (restartingNodeIndex.get() == -1) { - hasError = false; - } - lastException.set(null); - errorIndex = -1; - } - - // Check if replace-datanode policy is satisfied. - if (dfsClient.dtpReplaceDatanodeOnFailure.satisfy(blockReplication, - nodes, isAppend, isHflushed)) { - try { - addDatanode2ExistingPipeline(); - } catch(IOException ioe) { - if (!dfsClient.dtpReplaceDatanodeOnFailure.isBestEffort()) { - throw ioe; - } - DFSClient.LOG.warn("Failed to replace datanode." - + " Continue with the remaining datanodes since " - + DFSConfigKeys.DFS_CLIENT_WRITE_REPLACE_DATANODE_ON_FAILURE_BEST_EFFORT_KEY - + " is set to true.", ioe); - } - } - - // get a new generation stamp and an access token - LocatedBlock lb = dfsClient.namenode.updateBlockForPipeline(block, dfsClient.clientName); - newGS = lb.getBlock().getGenerationStamp(); - accessToken = lb.getBlockToken(); - - // set up the pipeline again with the remaining nodes - if (failPacket) { // for testing - success = createBlockOutputStream(nodes, storageTypes, newGS, isRecovery); - failPacket = false; - try { - // Give DNs time to send in bad reports. In real situations, - // good reports should follow bad ones, if client committed - // with those nodes. - Thread.sleep(2000); - } catch (InterruptedException ie) {} - } else { - success = createBlockOutputStream(nodes, storageTypes, newGS, isRecovery); - } - - if (restartingNodeIndex.get() >= 0) { - assert hasError == true; - // check errorIndex set above - if (errorIndex == restartingNodeIndex.get()) { - // ignore, if came from the restarting node - errorIndex = -1; - } - // still within the deadline - if (Time.monotonicNow() < restartDeadline) { - continue; // with in the deadline - } - // expired. declare the restarting node dead - restartDeadline = 0; - int expiredNodeIndex = restartingNodeIndex.get(); - restartingNodeIndex.set(-1); - DFSClient.LOG.warn("Datanode did not restart in time: " + - nodes[expiredNodeIndex]); - // Mark the restarting node as failed. If there is any other failed - // node during the last pipeline construction attempt, it will not be - // overwritten/dropped. In this case, the restarting node will get - // excluded in the following attempt, if it still does not come up. - if (errorIndex == -1) { - errorIndex = expiredNodeIndex; - } - // From this point on, normal pipeline recovery applies. - } - } // while - - if (success) { - // update pipeline at the namenode - ExtendedBlock newBlock = new ExtendedBlock( - block.getBlockPoolId(), block.getBlockId(), block.getNumBytes(), newGS); - dfsClient.namenode.updatePipeline(dfsClient.clientName, block, newBlock, - nodes, storageIDs); - // update client side generation stamp - block = newBlock; - } - return false; // do not sleep, continue processing - } - - /** - * Open a DataOutputStream to a DataNode so that it can be written to. - * This happens when a file is created and each time a new block is allocated. - * Must get block ID and the IDs of the destinations from the namenode. - * Returns the list of target datanodes. - */ - private LocatedBlock nextBlockOutputStream() throws IOException { - LocatedBlock lb = null; - DatanodeInfo[] nodes = null; - StorageType[] storageTypes = null; - int count = dfsClient.getConf().nBlockWriteRetry; - boolean success = false; - ExtendedBlock oldBlock = block; - do { - hasError = false; - lastException.set(null); - errorIndex = -1; - success = false; - - DatanodeInfo[] excluded = - excludedNodes.getAllPresent(excludedNodes.asMap().keySet()) - .keySet() - .toArray(new DatanodeInfo[0]); - block = oldBlock; - lb = locateFollowingBlock(excluded.length > 0 ? excluded : null); - block = lb.getBlock(); - block.setNumBytes(0); - bytesSent = 0; - accessToken = lb.getBlockToken(); - nodes = lb.getLocations(); - storageTypes = lb.getStorageTypes(); - - // - // Connect to first DataNode in the list. - // - success = createBlockOutputStream(nodes, storageTypes, 0L, false); - - if (!success) { - DFSClient.LOG.info("Abandoning " + block); - dfsClient.namenode.abandonBlock(block, fileId, src, - dfsClient.clientName); - block = null; - DFSClient.LOG.info("Excluding datanode " + nodes[errorIndex]); - excludedNodes.put(nodes[errorIndex], nodes[errorIndex]); - } - } while (!success && --count >= 0); - - if (!success) { - throw new IOException("Unable to create new block."); - } - return lb; - } - - // connects to the first datanode in the pipeline - // Returns true if success, otherwise return failure. - // - private boolean createBlockOutputStream(DatanodeInfo[] nodes, - StorageType[] nodeStorageTypes, long newGS, boolean recoveryFlag) { - if (nodes.length == 0) { - DFSClient.LOG.info("nodes are empty for write pipeline of block " - + block); - return false; - } - Status pipelineStatus = SUCCESS; - String firstBadLink = ""; - boolean checkRestart = false; - if (DFSClient.LOG.isDebugEnabled()) { - for (int i = 0; i < nodes.length; i++) { - DFSClient.LOG.debug("pipeline = " + nodes[i]); - } - } - - // persist blocks on namenode on next flush - persistBlocks.set(true); - - int refetchEncryptionKey = 1; - while (true) { - boolean result = false; - DataOutputStream out = null; - try { - assert null == s : "Previous socket unclosed"; - assert null == blockReplyStream : "Previous blockReplyStream unclosed"; - s = createSocketForPipeline(nodes[0], nodes.length, dfsClient); - long writeTimeout = dfsClient.getDatanodeWriteTimeout(nodes.length); - - OutputStream unbufOut = NetUtils.getOutputStream(s, writeTimeout); - InputStream unbufIn = NetUtils.getInputStream(s); - IOStreamPair saslStreams = dfsClient.saslClient.socketSend(s, - unbufOut, unbufIn, dfsClient, accessToken, nodes[0]); - unbufOut = saslStreams.out; - unbufIn = saslStreams.in; - out = new DataOutputStream(new BufferedOutputStream(unbufOut, - HdfsConstants.SMALL_BUFFER_SIZE)); - blockReplyStream = new DataInputStream(unbufIn); - - // - // Xmit header info to datanode - // - - BlockConstructionStage bcs = recoveryFlag? stage.getRecoveryStage(): stage; - - // We cannot change the block length in 'block' as it counts the number - // of bytes ack'ed. - ExtendedBlock blockCopy = new ExtendedBlock(block); - blockCopy.setNumBytes(blockSize); - - boolean[] targetPinnings = getPinnings(nodes, true); - // send the request - new Sender(out).writeBlock(blockCopy, nodeStorageTypes[0], accessToken, - dfsClient.clientName, nodes, nodeStorageTypes, null, bcs, - nodes.length, block.getNumBytes(), bytesSent, newGS, - checksum4WriteBlock, cachingStrategy.get(), isLazyPersistFile, - (targetPinnings == null ? false : targetPinnings[0]), targetPinnings); - - // receive ack for connect - BlockOpResponseProto resp = BlockOpResponseProto.parseFrom( - PBHelper.vintPrefixed(blockReplyStream)); - pipelineStatus = resp.getStatus(); - firstBadLink = resp.getFirstBadLink(); - - // Got an restart OOB ack. - // If a node is already restarting, this status is not likely from - // the same node. If it is from a different node, it is not - // from the local datanode. Thus it is safe to treat this as a - // regular node error. - if (PipelineAck.isRestartOOBStatus(pipelineStatus) && - restartingNodeIndex.get() == -1) { - checkRestart = true; - throw new IOException("A datanode is restarting."); - } - - String logInfo = "ack with firstBadLink as " + firstBadLink; - DataTransferProtoUtil.checkBlockOpStatus(resp, logInfo); - - assert null == blockStream : "Previous blockStream unclosed"; - blockStream = out; - result = true; // success - restartingNodeIndex.set(-1); - hasError = false; - } catch (IOException ie) { - if (restartingNodeIndex.get() == -1) { - DFSClient.LOG.info("Exception in createBlockOutputStream", ie); - } - if (ie instanceof InvalidEncryptionKeyException && refetchEncryptionKey > 0) { - DFSClient.LOG.info("Will fetch a new encryption key and retry, " - + "encryption key was invalid when connecting to " - + nodes[0] + " : " + ie); - // The encryption key used is invalid. - refetchEncryptionKey--; - dfsClient.clearDataEncryptionKey(); - // Don't close the socket/exclude this node just yet. Try again with - // a new encryption key. - continue; - } - - // find the datanode that matches - if (firstBadLink.length() != 0) { - for (int i = 0; i < nodes.length; i++) { - // NB: Unconditionally using the xfer addr w/o hostname - if (firstBadLink.equals(nodes[i].getXferAddr())) { - errorIndex = i; - break; - } - } - } else { - assert checkRestart == false; - errorIndex = 0; - } - // Check whether there is a restart worth waiting for. - if (checkRestart && shouldWaitForRestart(errorIndex)) { - restartDeadline = dfsClient.getConf().datanodeRestartTimeout + - Time.monotonicNow(); - restartingNodeIndex.set(errorIndex); - errorIndex = -1; - DFSClient.LOG.info("Waiting for the datanode to be restarted: " + - nodes[restartingNodeIndex.get()]); - } - hasError = true; - setLastException(ie); - result = false; // error - } finally { - if (!result) { - IOUtils.closeSocket(s); - s = null; - IOUtils.closeStream(out); - out = null; - IOUtils.closeStream(blockReplyStream); - blockReplyStream = null; - } - } - return result; - } - } - - private boolean[] getPinnings(DatanodeInfo[] nodes, boolean shouldLog) { - if (favoredNodes == null) { - return null; - } else { - boolean[] pinnings = new boolean[nodes.length]; - HashSet<String> favoredSet = - new HashSet<String>(Arrays.asList(favoredNodes)); - for (int i = 0; i < nodes.length; i++) { - pinnings[i] = favoredSet.remove(nodes[i].getXferAddrWithHostname()); - if (DFSClient.LOG.isDebugEnabled()) { - DFSClient.LOG.debug(nodes[i].getXferAddrWithHostname() + - " was chosen by name node (favored=" + pinnings[i] + - ")."); - } - } - if (shouldLog && !favoredSet.isEmpty()) { - // There is one or more favored nodes that were not allocated. - DFSClient.LOG.warn( - "These favored nodes were specified but not chosen: " + - favoredSet + - " Specified favored nodes: " + Arrays.toString(favoredNodes)); - - } - return pinnings; - } - } - - private LocatedBlock locateFollowingBlock(DatanodeInfo[] excludedNodes) throws IOException { - int retries = dfsClient.getConf().nBlockWriteLocateFollowingRetry; - long sleeptime = dfsClient.getConf(). - blockWriteLocateFollowingInitialDelayMs; - while (true) { - long localstart = Time.monotonicNow(); - while (true) { - try { - return dfsClient.namenode.addBlock(src, dfsClient.clientName, - block, excludedNodes, fileId, favoredNodes); - } catch (RemoteException e) { - IOException ue = - e.unwrapRemoteException(FileNotFoundException.class, - AccessControlException.class, - NSQuotaExceededException.class, - DSQuotaExceededException.class, - UnresolvedPathException.class); - if (ue != e) { - throw ue; // no need to retry these exceptions - } - - - if (NotReplicatedYetException.class.getName(). - equals(e.getClassName())) { - if (retries == 0) { - throw e; - } else { - --retries; - DFSClient.LOG.info("Exception while adding a block", e); - long elapsed = Time.monotonicNow() - localstart; - if (elapsed > 5000) { - DFSClient.LOG.info("Waiting for replication for " - + (elapsed / 1000) + " seconds"); - } - try { - DFSClient.LOG.warn("NotReplicatedYetException sleeping " + src - + " retries left " + retries); - Thread.sleep(sleeptime); - sleeptime *= 2; - } catch (InterruptedException ie) { - DFSClient.LOG.warn("Caught exception ", ie); - } - } - } else { - throw e; - } - - } - } - } - } - - ExtendedBlock getBlock() { - return block; - } - - DatanodeInfo[] getNodes() { - return nodes; - } - - Token<BlockTokenIdentifier> getBlockToken() { - return accessToken; - } - - private void setLastException(IOException e) { - lastException.compareAndSet(null, e); - } - } - - /** - * Create a socket for a write pipeline - * @param first the first datanode - * @param length the pipeline length - * @param client client - * @return the socket connected to the first datanode - */ - static Socket createSocketForPipeline(final DatanodeInfo first, - final int length, final DFSClient client) throws IOException { - final String dnAddr = first.getXferAddr( - client.getConf().connectToDnViaHostname); - if (DFSClient.LOG.isDebugEnabled()) { - DFSClient.LOG.debug("Connecting to datanode " + dnAddr); - } - final InetSocketAddress isa = NetUtils.createSocketAddr(dnAddr); - final Socket sock = client.socketFactory.createSocket(); - final int timeout = client.getDatanodeReadTimeout(length); - NetUtils.connect(sock, isa, client.getRandomLocalInterfaceAddr(), client.getConf().socketTimeout); - sock.setSoTimeout(timeout); - sock.setSendBufferSize(HdfsConstants.DEFAULT_DATA_SOCKET_SIZE); - if(DFSClient.LOG.isDebugEnabled()) { - DFSClient.LOG.debug("Send buf size " + sock.getSendBufferSize()); - } - return sock; - } - @Override protected void checkClosed() throws IOException { if (isClosed()) { - IOException e = lastException.get(); + IOException e = streamer.getLastException().get(); throw e != null ? e : new ClosedChannelException(); } } @@ -1536,7 +148,7 @@ public class DFSOutputStream extends FSOutputSummer // @VisibleForTesting public synchronized DatanodeInfo[] getPipeline() { - if (streamer == null) { + if (streamer.streamerClosed()) { return null; } DatanodeInfo[] currentNodes = streamer.getNodes(); @@ -1556,7 +168,7 @@ public class DFSOutputStream extends FSOutputSummer */ private static DataChecksum getChecksum4Compute(DataChecksum checksum, HdfsFileStatus stat) { - if (isLazyPersist(stat) && stat.getReplication() == 1) { + if (DataStreamer.isLazyPersist(stat) && stat.getReplication() == 1) { // do not compute checksum for writing to single replica to memory return DataChecksum.newDataChecksum(Type.NULL, checksum.getBytesPerChecksum()); @@ -1573,7 +185,6 @@ public class DFSOutputStream extends FSOutputSummer this.blockSize = stat.getBlockSize(); this.blockReplication = stat.getReplication(); this.fileEncryptionInfo = stat.getFileEncryptionInfo(); - this.progress = progress; this.cachingStrategy = new AtomicReference<CachingStrategy>( dfsClient.getDefaultWriteCachingStrategy()); if ((progress != null) && DFSClient.LOG.isDebugEnabled()) { @@ -1591,10 +202,6 @@ public class DFSOutputStream extends FSOutputSummer + DFSConfigKeys.DFS_BYTES_PER_CHECKSUM_KEY + " (=" + bytesPerChecksum + ") must divide block size (=" + blockSize + ")."); } - this.checksum4WriteBlock = checksum; - - this.dfsclientSlowLogThresholdMs = - dfsClient.getConf().dfsclientSlowIoWarningThresholdMs; this.byteArrayManager = dfsClient.getClientContext().getByteArrayManager(); } @@ -1607,7 +214,8 @@ public class DFSOutputStream extends FSOutputSummer computePacketChunkSize(dfsClient.getConf().writePacketSize, bytesPerChecksum); - streamer = new DataStreamer(stat, null); + streamer = new DataStreamer(stat, null, dfsClient, src, progress, checksum, + cachingStrategy, byteArrayManager); if (favoredNodes != null && favoredNodes.length != 0) { streamer.setFavoredNodes(favoredNodes); } @@ -1676,18 +284,57 @@ public class DFSOutputStream extends FSOutputSummer this(dfsClient, src, progress, stat, checksum); initialFileSize = stat.getLen(); // length of file when opened + this.fileEncryptionInfo = stat.getFileEncryptionInfo(); + // The last partial block of the file has to be filled. if (!toNewBlock && lastBlock != null) { // indicate that we are appending to an existing block - bytesCurBlock = lastBlock.getBlockSize(); - streamer = new DataStreamer(lastBlock, stat, bytesPerChecksum); + streamer = new DataStreamer(lastBlock, stat, dfsClient, src, progress, checksum, + cachingStrategy, byteArrayManager); + streamer.setBytesCurBlock(lastBlock.getBlockSize()); + adjustPacketChunkSize(stat); + streamer.setPipelineInConstruction(lastBlock); } else { computePacketChunkSize(dfsClient.getConf().writePacketSize, bytesPerChecksum); - streamer = new DataStreamer(stat, - lastBlock != null ? lastBlock.getBlock() : null); + streamer = new DataStreamer(stat, lastBlock != null ? lastBlock.getBlock() : null, + dfsClient, src, progress, checksum, cachingStrategy, byteArrayManager); + } + } + + private void adjustPacketChunkSize(HdfsFileStatus stat) throws IOException{ + + long usedInLastBlock = stat.getLen() % blockSize; + int freeInLastBlock = (int)(blockSize - usedInLastBlock); + + // calculate the amount of free space in the pre-existing + // last crc chunk + int usedInCksum = (int)(stat.getLen() % bytesPerChecksum); + int freeInCksum = bytesPerChecksum - usedInCksum; + + // if there is space in the last block, then we have to + // append to that block + if (freeInLastBlock == blockSize) { + throw new IOException("The last block for file " + + src + " is full."); + } + + if (usedInCksum > 0 && freeInCksum > 0) { + // if there is space in the last partial chunk, then + // setup in such a way that the next packet will have only + // one chunk that fills up the partial chunk. + // + computePacketChunkSize(0, freeInCksum); + setChecksumBufSize(freeInCksum); + streamer.setAppendChunk(true); + } else { + // if the remaining space in the block is smaller than + // that expected size of of a packet, then create + // smaller size packet. + // + computePacketChunkSize(Math.min(dfsClient.getConf().writePacketSize, freeInLastBlock), + bytesPerChecksum); } - this.fileEncryptionInfo = stat.getFileEncryptionInfo(); } static DFSOutputStream newStreamForAppend(DFSClient dfsClient, String src, @@ -1708,12 +355,6 @@ public class DFSOutputStream extends FSOutputSummer scope.close(); } } - - private static boolean isLazyPersist(HdfsFileStatus stat) { - final BlockStoragePolicy p = blockStoragePolicySuite.getPolicy( - HdfsConstants.MEMORY_STORAGE_POLICY_NAME); - return p != null && stat.getStoragePolicy() == p.getId(); - } private void computePacketChunkSize(int psize, int csize) { final int bodySize = psize - PacketHeader.PKT_MAX_HEADER_LEN; @@ -1728,62 +369,6 @@ public class DFSOutputStream extends FSOutputSummer } } - private void queueCurrentPacket() { - synchronized (dataQueue) { - if (currentPacket == null) return; - currentPacket.addTraceParent(Trace.currentSpan()); - dataQueue.addLast(currentPacket); - lastQueuedSeqno = currentPacket.getSeqno(); - if (DFSClient.LOG.isDebugEnabled()) { - DFSClient.LOG.debug("Queued packet " + currentPacket.getSeqno()); - } - currentPacket = null; - dataQueue.notifyAll(); - } - } - - private void waitAndQueueCurrentPacket() throws IOException { - synchronized (dataQueue) { - try { - // If queue is full, then wait till we have enough space - boolean firstWait = true; - try { - while (!isClosed() && dataQueue.size() + ackQueue.size() > - dfsClient.getConf().writeMaxPackets) { - if (firstWait) { - Span span = Trace.currentSpan(); - if (span != null) { - span.addTimelineAnnotation("dataQueue.wait"); - } - firstWait = false; - } - try { - dataQueue.wait(); - } catch (InterruptedException e) { - // If we get interrupted while waiting to queue data, we still need to get rid - // of the current packet. This is because we have an invariant that if - // currentPacket gets full, it will get queued before the next writeChunk. - // - // Rather than wait around for space in the queue, we should instead try to - // return to the caller as soon as possible, even though we slightly overrun - // the MAX_PACKETS length. - Thread.currentThread().interrupt(); - break; - } - } - } finally { - Span span = Trace.currentSpan(); - if ((span != null) && (!firstWait)) { - span.addTimelineAnnotation("end.wait"); - } - } - checkClosed(); - queueCurrentPacket(); - } catch (ClosedChannelException e) { - } - } - } - // @see FSOutputSummer#writeChunk() @Override protected synchronized void writeChunk(byte[] b, int offset, int len, @@ -1814,57 +399,62 @@ public class DFSOutputStream extends FSOutputSummer if (currentPacket == null) { currentPacket = createPacket(packetSize, chunksPerPacket, - bytesCurBlock, currentSeqno++, false); + streamer.getBytesCurBlock(), streamer.getAndIncCurrentSeqno(), false); if (DFSClient.LOG.isDebugEnabled()) { DFSClient.LOG.debug("DFSClient writeChunk allocating new packet seqno=" + currentPacket.getSeqno() + ", src=" + src + ", packetSize=" + packetSize + ", chunksPerPacket=" + chunksPerPacket + - ", bytesCurBlock=" + bytesCurBlock); + ", bytesCurBlock=" + streamer.getBytesCurBlock()); } } currentPacket.writeChecksum(checksum, ckoff, cklen); currentPacket.writeData(b, offset, len); currentPacket.incNumChunks(); - bytesCurBlock += len; + streamer.incBytesCurBlock(len); // If packet is full, enqueue it for transmission // if (currentPacket.getNumChunks() == currentPacket.getMaxChunks() || - bytesCurBlock == blockSize) { + streamer.getBytesCurBlock() == blockSize) { if (DFSClient.LOG.isDebugEnabled()) { DFSClient.LOG.debug("DFSClient writeChunk packet full seqno=" + currentPacket.getSeqno() + ", src=" + src + - ", bytesCurBlock=" + bytesCurBlock + + ", bytesCurBlock=" + streamer.getBytesCurBlock() + ", blockSize=" + blockSize + - ", appendChunk=" + appendChunk); + ", appendChunk=" + streamer.getAppendChunk()); } - waitAndQueueCurrentPacket(); + streamer.waitAndQueuePacket(currentPacket); + currentPacket = null; // If the reopened file did not end at chunk boundary and the above // write filled up its partial chunk. Tell the summer to generate full // crc chunks from now on. - if (appendChunk && bytesCurBlock%bytesPerChecksum == 0) { - appendChunk = false; + if (streamer.getAppendChunk() && + streamer.getBytesCurBlock() % bytesPerChecksum == 0) { + streamer.setAppendChunk(false); resetChecksumBufSize(); } - if (!appendChunk) { - int psize = Math.min((int)(blockSize-bytesCurBlock), dfsClient.getConf().writePacketSize); + if (!streamer.getAppendChunk()) { + int psize = Math.min((int)(blockSize-streamer.getBytesCurBlock()), + dfsClient.getConf().writePacketSize); computePacketChunkSize(psize, bytesPerChecksum); } // // if encountering a block boundary, send an empty packet to // indicate the end of block and reset bytesCurBlock. // - if (bytesCurBlock == blockSize) { - currentPacket = createPacket(0, 0, bytesCurBlock, currentSeqno++, true); + if (streamer.getBytesCurBlock() == blockSize) { + currentPacket = createPacket(0, 0, streamer.getBytesCurBlock(), + streamer.getAndIncCurrentSeqno(), true); currentPacket.setSyncBlock(shouldSyncBlock); - waitAndQueueCurrentPacket(); - bytesCurBlock = 0; + streamer.waitAndQueuePacket(currentPacket); + currentPacket = null; + streamer.setBytesCurBlock(0); lastFlushOffset = 0; } } @@ -1954,30 +544,30 @@ public class DFSOutputStream extends FSOutputSummer if (DFSClient.LOG.isDebugEnabled()) { DFSClient.LOG.debug("DFSClient flush(): " - + " bytesCurBlock=" + bytesCurBlock + + " bytesCurBlock=" + streamer.getBytesCurBlock() + " lastFlushOffset=" + lastFlushOffset + " createNewBlock=" + endBlock); } // Flush only if we haven't already flushed till this offset. - if (lastFlushOffset != bytesCurBlock) { - assert bytesCurBlock > lastFlushOffset; + if (lastFlushOffset != streamer.getBytesCurBlock()) { + assert streamer.getBytesCurBlock() > lastFlushOffset; // record the valid offset of this flush - lastFlushOffset = bytesCurBlock; + lastFlushOffset = streamer.getBytesCurBlock(); if (isSync && currentPacket == null && !endBlock) { // Nothing to send right now, // but sync was requested. // Send an empty packet if we do not end the block right now currentPacket = createPacket(packetSize, chunksPerPacket, - bytesCurBlock, currentSeqno++, false); + streamer.getBytesCurBlock(), streamer.getAndIncCurrentSeqno(), false); } } else { - if (isSync && bytesCurBlock > 0 && !endBlock) { + if (isSync && streamer.getBytesCurBlock() > 0 && !endBlock) { // Nothing to send right now, // and the block was partially written, // and sync was requested. // So send an empty sync packet if we do not end the block right now currentPacket = createPacket(packetSize, chunksPerPacket, - bytesCurBlock, currentSeqno++, false); + streamer.getBytesCurBlock(), streamer.getAndIncCurrentSeqno(), false); } else if (currentPacket != null) { // just discard the current packet since it is already been sent. currentPacket.releaseBuffer(byteArrayManager); @@ -1986,39 +576,42 @@ public class DFSOutputStream extends FSOutputSummer } if (currentPacket != null) { currentPacket.setSyncBlock(isSync); - waitAndQueueCurrentPacket(); + streamer.waitAndQueuePacket(currentPacket); + currentPacket = null; } - if (endBlock && bytesCurBlock > 0) { + if (endBlock && streamer.getBytesCurBlock() > 0) { // Need to end the current block, thus send an empty packet to // indicate this is the end of the block and reset bytesCurBlock - currentPacket = createPacket(0, 0, bytesCurBlock, currentSeqno++, true); + currentPacket = createPacket(0, 0, streamer.getBytesCurBlock(), + streamer.getAndIncCurrentSeqno(), true); currentPacket.setSyncBlock(shouldSyncBlock || isSync); - waitAndQueueCurrentPacket(); - bytesCurBlock = 0; + streamer.waitAndQueuePacket(currentPacket); + currentPacket = null; + streamer.setBytesCurBlock(0); lastFlushOffset = 0; } else { // Restore state of stream. Record the last flush offset // of the last full chunk that was flushed. - bytesCurBlock -= numKept; + streamer.setBytesCurBlock(streamer.getBytesCurBlock() - numKept); } - toWaitFor = lastQueuedSeqno; + toWaitFor = streamer.getLastQueuedSeqno(); } // end synchronized - waitForAckedSeqno(toWaitFor); + streamer.waitForAckedSeqno(toWaitFor); // update the block length first time irrespective of flag - if (updateLength || persistBlocks.get()) { + if (updateLength || streamer.getPersistBlocks().get()) { synchronized (this) { - if (streamer != null && streamer.block != null) { - lastBlockLength = streamer.block.getNumBytes(); + if (!streamer.streamerClosed() && streamer.getBlock() != null) { + lastBlockLength = streamer.getBlock().getNumBytes(); } } } // If 1) any new blocks were allocated since the last flush, or 2) to // update length in NN is required, then persist block locations on // namenode. - if (persistBlocks.getAndSet(false) || updateLength) { + if (streamer.getPersistBlocks().getAndSet(false) || updateLength) { try { dfsClient.namenode.fsync(src, fileId, dfsClient.clientName, lastBlockLength); @@ -2035,7 +628,7 @@ public class DFSOutputStream extends FSOutputSummer } synchronized(this) { - if (streamer != null) { + if (!streamer.streamerClosed()) { streamer.setHflush(); } } @@ -2048,7 +641,7 @@ public class DFSOutputStream extends FSOutputSummer DFSClient.LOG.warn("Error while syncing", e); synchronized (this) { if (!isClosed()) { - lastException.set(new IOException("IOException flush: " + e)); + streamer.getLastException().set(new IOException("IOException flush: " + e)); closeThreads(true); } } @@ -2073,7 +666,7 @@ public class DFSOutputStream extends FSOutputSummer public synchronized int getCurrentBlockReplication() throws IOException { dfsClient.checkOpen(); checkClosed(); - if (streamer == null) { + if (streamer.streamerClosed()) { return blockReplication; // no pipeline, return repl factor of file } DatanodeInfo[] currentNodes = streamer.getNodes(); @@ -2095,47 +688,12 @@ public class DFSOutputStream extends FSOutputSummer // // If there is data in the current buffer, send it across // - queueCurrentPacket(); - toWaitFor = lastQueuedSeqno; + streamer.queuePacket(currentPacket); + currentPacket = null; + toWaitFor = streamer.getLastQueuedSeqno(); } - waitForAckedSeqno(toWaitFor); - } - - private void waitForAckedSeqno(long seqno) throws IOException { - TraceScope scope = Trace.startSpan("waitForAckedSeqno", Sampler.NEVER); - try { - if (DFSClient.LOG.isDebugEnabled()) { - DFSClient.LOG.debug("Waiting for ack for: " + seqno); - } - long begin = Time.monotonicNow(); - try { - synchronized (dataQueue) { - while (!isClosed()) { - checkClosed(); - if (lastAckedSeqno >= seqno) { - break; - } - try { - dataQueue.wait(1000); // when we receive an ack, we notify on - // dataQueue - } catch (InterruptedException ie) { - throw new InterruptedIOException( - "Interrupted while waiting for data to be acknowledged by pipeline"); - } - } - } - checkClosed(); - } catch (ClosedChannelException e) { - } - long duration = Time.monotonicNow() - begin; - if (duration > dfsclientSlowLogThresholdMs) { - DFSClient.LOG.warn("Slow waitForAckedSeqno took " + duration - + "ms (threshold=" + dfsclientSlowLogThresholdMs + "ms)"); - } - } finally { - scope.close(); - } + streamer.waitForAckedSeqno(toWaitFor); } private synchronized void start() { @@ -2157,22 +715,12 @@ public class DFSOutputStream extends FSOutputSummer } boolean isClosed() { - return closed; + return closed || streamer.streamerClosed(); } void setClosed() { closed = true; - synchronized (dataQueue) { - releaseBuffer(dataQueue, byteArrayManager); - releaseBuffer(ackQueue, byteArrayManager); - } - } - - private static void releaseBuffer(List<DFSPacket> packets, ByteArrayManager bam) { - for (DFSPacket p : packets) { - p.releaseBuffer(bam); - } - packets.clear(); + streamer.release(); } // shutdown datastreamer and responseprocessor threads. @@ -2181,14 +729,11 @@ public class DFSOutputStream extends FSOutputSummer try { streamer.close(force); streamer.join(); - if (s != null) { - s.close(); - } + streamer.closeSocket(); } catch (InterruptedException e) { throw new IOException("Failed to shutdown streamer"); } finally { - streamer = null; - s = null; + streamer.setSocketToNull(); setClosed(); } } @@ -2210,7 +755,7 @@ public class DFSOutputStream extends FSOutputSummer private synchronized void closeImpl() throws IOException { if (isClosed()) { - IOException e = lastException.getAndSet(null); + IOException e = streamer.getLastException().getAndSet(null); if (e == null) return; else @@ -2221,12 +766,14 @@ public class DFSOutputStream extends FSOutputSummer flushBuffer(); // flush from all upper layers if (currentPacket != null) { - waitAndQueueCurrentPacket(); + streamer.waitAndQueuePacket(currentPacket); + currentPacket = null; } - if (bytesCurBlock != 0) { + if (streamer.getBytesCurBlock() != 0) { // send an empty packet to mark the end of the block - currentPacket = createPacket(0, 0, bytesCurBlock, currentSeqno++, true); + currentPacket = createPacket(0, 0, streamer.getBytesCurBlock(), + streamer.getAndIncCurrentSeqno(), true); currentPacket.setSyncBlock(shouldSyncBlock); } @@ -2261,7 +808,7 @@ public class DFSOutputStream extends FSOutputSummer if (!fileComplete) { final int hdfsTimeout = dfsClient.getHdfsTimeout(); if (!dfsClient.clientRunning - || (hdfsTimeout > 0 + || (hdfsTimeout > 0 && localstart + hdfsTimeout < Time.monotonicNow())) { String msg = "Unable to close file because dfsclient " + " was unable to contact the HDFS servers." + @@ -2290,7 +837,7 @@ public class DFSOutputStream extends FSOutputSummer @VisibleForTesting public void setArtificialSlowdown(long period) { - artificialSlowdown = period; + streamer.setArtificialSlowdown(period); } @VisibleForTesting @@ -2299,10 +846,6 @@ public class DFSOutputStream extends FSOutputSummer packetSize = (bytesPerChecksum + getChecksumSize()) * chunksPerPacket; } - synchronized void setTestFilename(String newname) { - src = newname; - } - /** * Returns the size of a file as it was when this stream was opened */ @@ -2345,9 +888,4 @@ public class DFSOutputStream extends FSOutputSummer public long getFileId() { return fileId; } - - private static <T> void arraycopy(T[] srcs, T[] dsts, int skipIndex) { - System.arraycopy(srcs, 0, dsts, 0, skipIndex); - System.arraycopy(srcs, skipIndex+1, dsts, skipIndex, dsts.length-skipIndex); - } }