FSDataOutputStream out = fs.create(outFile);

/***************************************************************** Implementation of the abstract FileSystem for the DFS system.* This object is the way end-user code interacts with a Hadoop* DistributedFileSystem.******************************************************************/

 @Overridepublic FSDataOutputStream create(final Path f, final FsPermission permission,final EnumSet<CreateFlag> cflags, final int bufferSize,final short replication, final long blockSize, final Progressable progress,final ChecksumOpt checksumOpt) throws IOException {statistics.incrementWriteOps(1);Path absF = fixRelativePart(f);return new FileSystemLinkResolver<FSDataOutputStream>() {@Overridepublic FSDataOutputStream doCall(final Path p)throws IOException, UnresolvedLinkException {final DFSOutputStream dfsos = dfs.create(getPathName(p), permission,cflags, replication, blockSize, progress, bufferSize,checksumOpt);return dfs.createWrappedOutputStream(dfsos, statistics);}@Overridepublic FSDataOutputStream next(final FileSystem fs, final Path p)throws IOException {return fs.create(p, permission, cflags, bufferSize,replication, blockSize, progress, checksumOpt);}}.resolve(this, absF);}

 /*** Same as {@link #create(String, FsPermission, EnumSet, boolean, short, long,* Progressable, int, ChecksumOpt)} with the addition of favoredNodes that is* a hint to where the namenode should place the file blocks.* The favored nodes hint is not persisted in HDFS. Hence it may be honored* at the creation time only. HDFS could move the blocks during balancing or* replication, to move the blocks from favored nodes. A value of null means* no favored nodes for this create*/public DFSOutputStream create(String src, FsPermission permission,EnumSet<CreateFlag> flag, boolean createParent,short replication,long blockSize,Progressable progress,int buffersize,ChecksumOpt checksumOpt,InetSocketAddress[] favoredNodes) throws IOException {checkOpen();if (permission == null) {permission = FsPermission.getFileDefault();}FsPermission masked = permission.applyUMask(dfsClientConf.uMask);if(LOG.isDebugEnabled()) {LOG.debug(src + ": masked=" + masked);}String[] favoredNodeStrs = null;if (favoredNodes != null) {favoredNodeStrs = new String[favoredNodes.length];for (int i = 0; i < favoredNodes.length; i++) {favoredNodeStrs[i] = favoredNodes[i].getHostName() + ":" + favoredNodes[i].getPort();}}final DFSOutputStream result = DFSOutputStream.newStreamForCreate(this,src, masked, flag, createParent, replication, blockSize, progress,buffersize, dfsClientConf.createChecksum(checksumOpt),favoredNodeStrs);beginFileLease(result.getFileId(), result);return result;}

/***************************************************************** DFSOutputStream creates files from a stream of bytes.** The client application writes data that is cached internally by* this stream. Data is broken up into packets, each packet is* typically 64K in size. A packet comprises of chunks. Each chunk* 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.** 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.
****************************************************************/

 static DFSOutputStream newStreamForCreate(DFSClient dfsClient, String src,FsPermission masked, EnumSet<CreateFlag> flag, boolean createParent,short replication, long blockSize, Progressable progress, int buffersize,DataChecksum checksum, String[] favoredNodes) throws IOException {HdfsFileStatus stat = null;// Retry the create if we get a RetryStartFileException up to a maximum// number of timesboolean shouldRetry = true;int retryCount = CREATE_RETRY_COUNT;while (shouldRetry) {shouldRetry = false;try {stat = dfsClient.namenode.create(src, masked, dfsClient.clientName,new EnumSetWritable<CreateFlag>(flag), createParent, replication,blockSize, SUPPORTED_CRYPTO_VERSIONS);break;} catch (RemoteException re) {IOException e = re.unwrapRemoteException(AccessControlException.class,DSQuotaExceededException.class,FileAlreadyExistsException.class,FileNotFoundException.class,ParentNotDirectoryException.class,NSQuotaExceededException.class,RetryStartFileException.class,SafeModeException.class,UnresolvedPathException.class,SnapshotAccessControlException.class,UnknownCryptoProtocolVersionException.class);if (e instanceof RetryStartFileException) {if (retryCount > 0) {shouldRetry = true;retryCount--;} else {throw new IOException("Too many retries because of encryption" +" zone operations", e);}} else {throw e;}}}Preconditions.checkNotNull(stat, "HdfsFileStatus should not be null!");final DFSOutputStream out = new DFSOutputStream(dfsClient, src, stat,flag, progress, checksum, favoredNodes);out.start();return out;}

 private static class Packet {private static final long HEART_BEAT_SEQNO = -1L;long seqno; // sequencenumber of buffer in blockfinal long offsetInBlock; // offset in blockboolean syncBlock; // this packet forces the current block to diskint numChunks; // number of chunks currently in packetfinal int maxChunks; // max chunks in packetprivate byte[] buf;private boolean lastPacketInBlock; // is this the last packet in block?/*** buf is pointed into like follows:*  (C is checksum data, D is payload data)** [_________CCCCCCCCC________________DDDDDDDDDDDDDDDD___]*           ^        ^               ^               ^*           |        checksumPos     dataStart       dataPos*           checksumStart* * Right before sending, we move the checksum data to immediately precede* the actual data, and then insert the header into the buffer immediately* preceding the checksum data, so we make sure to keep enough space in* front of the checksum data to support the largest conceivable header. */int checksumStart;int checksumPos;final int dataStart;int dataPos;/*** Create a new packet.* * @param pktSize maximum size of the packet, *                including checksum data and actual data.* @param chunksPerPkt maximum number of chunks per packet.* @param offsetInBlock offset in bytes into the HDFS block.*/private Packet(byte[] buf, int chunksPerPkt, long offsetInBlock, long seqno,int checksumSize) {this.lastPacketInBlock = false;this.numChunks = 0;this.offsetInBlock = offsetInBlock;this.seqno = seqno;this.buf = buf;checksumStart = PacketHeader.PKT_MAX_HEADER_LEN;checksumPos = checksumStart;dataStart = checksumStart + (chunksPerPkt * checksumSize);dataPos = dataStart;maxChunks = chunksPerPkt;}
}

  private synchronized void start() {streamer.start();}

  //// 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.//

 /** streamer thread is the only thread that opens streams to datanode, * and closes them. Any error recovery is also done by this thread.*/@Overridepublic void run() {long lastPacket = Time.now();TraceScope traceScope = null;if (traceSpan != null) {traceScope = Trace.continueSpan(traceSpan);}while (!streamerClosed && dfsClient.clientRunning) {// if the Responder encountered an error, shutdown Responderif (hasError && response != null) {try {response.close();response.join();response = null;} catch (InterruptedException  e) {DFSClient.LOG.warn("Caught exception ", e);}}Packet one;try {// process datanode IO errors if anyboolean doSleep = false;if (hasError && (errorIndex >= 0 || restartingNodeIndex >= 0)) {doSleep = processDatanodeError();}synchronized (dataQueue) {// wait for a packet to be sent.long now = Time.now();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.now();}if (streamerClosed || hasError || !dfsClient.clientRunning) {continue;}// get packet to be sent.if (dataQueue.isEmpty()) {one = createHeartbeatPacket();} else {one = dataQueue.getFirst(); // regular data packet
            }}assert one != null;// 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.lastPacketInBlock) {// wait for all data packets have been successfully ackedsynchronized (dataQueue) {while (!streamerClosed && !hasError && ackQueue.size() != 0 && dfsClient.clientRunning) {try {// wait for acks to arrive from datanodesdataQueue.wait(1000);} catch (InterruptedException  e) {DFSClient.LOG.warn("Caught exception ", e);}}}if (streamerClosed || hasError || !dfsClient.clientRunning) {continue;}stage = BlockConstructionStage.PIPELINE_CLOSE;}// send the packetsynchronized (dataQueue) {// move packet from dataQueue to ackQueueif (!one.isHeartbeatPacket()) {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 datanodetry {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;}lastPacket = Time.now();// update bytesSentlong tmpBytesSent = one.getLastByteOffsetBlock();if (bytesSent < tmpBytesSent) {bytesSent = tmpBytesSent;}if (streamerClosed || hasError || !dfsClient.clientRunning) {continue;}// Is this block full?if (one.lastPacketInBlock) {// wait for the close packet has been ackedsynchronized (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 == -1) {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 == -1) {// Not a datanode issuestreamerClosed = true;}}}if (traceScope != null) {traceScope.close();}closeInternal();}

    /*** 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;}

    //// Processes responses from the datanodes.  A packet is removed// from the ackQueue when its response arrives.//

 public void run() {setName("ResponseProcessor for block " + block);
        PipelineAck ack = new PipelineAck();
while (!responderClosed && dfsClient.clientRunning && !isLastPacketInBlock) {// process responses from datanodes.try {// read an ack from the pipelinelong begin = Time.monotonicNow();
            ack.readFields(blockReplyStream);
            long duration = Time.monotonicNow() - begin;
            if (duration > dfsclientSlowLogThresholdMs&& ack.getSeqno() != Packet.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 = 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.now();
                setRestartingNodeIndex(i);
                String message = "A datanode is restarting: " + targets[i];
                DFSClient.LOG.info(message);
               throw new IOException(message);
              }// node errorif (reply != SUCCESS) {setErrorIndex(i); // first bad datanodethrow 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 == Packet.HEART_BEAT_SEQNO) {  // a heartbeat ackcontinue;
            }// a success ack for a data packetPacket one;
            synchronized (dataQueue) {one = ackQueue.getFirst();
            }if (one.seqno != seqno) {throw new IOException("ResponseProcessor: Expecting seqno " +" for block " + block +one.seqno + " but received " + seqno);
            }isLastPacketInBlock = one.lastPacketInBlock;
// 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 bytesAckedblock.setNumBytes(one.getLastByteOffsetBlock());
synchronized (dataQueue) {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 == -1) {DFSClient.LOG.warn("DFSOutputStream ResponseProcessor exception "+ " for block " + block, e);
              }responderClosed = true;
            }}}}