flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobClient.java
/**
	 * Uploads data from the given input stream to the BLOB server.
	 *
	 * @param jobId
	 * 		the ID of the job the BLOB belongs to (or <tt>null</tt> if job-unrelated)
	 * @param inputStream
	 * 		the input stream to read the data from
	 * @param blobType
	 * 		whether the BLOB should become permanent or transient
	 *
	 * @return the computed BLOB key of the uploaded BLOB
	 *
	 * @throws IOException
	 * 		thrown if an I/O error occurs while uploading the data to the BLOB server
	 */
Uploads data from the given input stream to the BLOB server.
@param jobId
the ID of the job the BLOB belongs to (or null if job-unrelated)
@param inputStream
the input stream to read the data from
@param blobType
whether the BLOB should become permanent or transient

@return the computed BLOB key of the uploaded BLOB

@throws IOException
thrown if an I/O error occurs while uploading the data to the BLOB server
Uploads data from the given input stream to the BLOB server . @param jobId the ID of the job the BLOB belongs to ( or null if job - unrelated ) @param inputStream the input stream to read the data from @param blobType whether the BLOB should become permanent or transient @return the computed BLOB key of the uploaded BLOB @throws IOException thrown if an I / O error occurs while uploading the data to the BLOB server
BlobKey putInputStream(@Nullable JobID jobId, InputStream inputStream, BlobKey.BlobType blobType)
			throws IOException {
		if (this.socket.isClosed()) {
			throw new IllegalStateException("BLOB Client is not connected. " +
					"Client has been shut down or encountered an error before.");
		}
		checkNotNull(inputStream);
		if (LOG.isDebugEnabled()) {
			LOG.debug("PUT BLOB stream to {}.", socket.getLocalSocketAddress());
		}
		try {
			final OutputStream os = this.socket.getOutputStream();
			final MessageDigest md = BlobUtils.createMessageDigest();
			final byte[] xferBuf = new byte[BUFFER_SIZE];
			sendPutHeader(os, jobId, blobType);
			while (true) {
				final int read = inputStream.read(xferBuf);
				if (read < 0) {
					writeLength(-1, os);
					break;
				}
				if (read > 0) {
					writeLength(read, os);
					os.write(xferBuf, 0, read);
					md.update(xferBuf, 0, read);
				}
			}
			final InputStream is = this.socket.getInputStream();
			return receiveAndCheckPutResponse(is, md, blobType);
		}
		catch (Throwable t) {
			BlobUtils.closeSilently(socket, LOG);
			throw new IOException("PUT operation failed: " + t.getMessage(), t);
		}
	}
BlobKey putInputStream ( @ Nullable JobID jobId , InputStream inputStream , BlobKey . BlobType blobType ) throws IOException { if ( this . socket . isClosed ( ) ) { throw new IllegalStateException ( "BLOB Client is not connected. " + "Client has been shut down or encountered an error before." ) ; } checkNotNull ( inputStream ) ; if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "PUT BLOB stream to {}." , socket . getLocalSocketAddress ( ) ) ; } try { final OutputStream os = this . socket . getOutputStream ( ) ; final MessageDigest md = BlobUtils . createMessageDigest ( ) ; final byte [ ] xferBuf = new byte [ BUFFER_SIZE ] ; sendPutHeader ( os , jobId , blobType ) ; while ( true ) { final int read = inputStream . read ( xferBuf ) ; if ( read < 0 ) { writeLength ( - 1 , os ) ; break ; } if ( read > 0 ) { writeLength ( read , os ) ; os . write ( xferBuf , 0 , read ) ; md . update ( xferBuf , 0 , read ) ; } } final InputStream is = this . socket . getInputStream ( ) ; return receiveAndCheckPutResponse ( is , md , blobType ) ; } catch ( Throwable t ) { BlobUtils . closeSilently ( socket , LOG ) ; throw new IOException ( "PUT operation failed: " + t . getMessage ( ) , t ) ; } }
Uploads data from the given input stream to the BLOB server.
Uploads data from the given input stream to the BLOB server .
// Send the PUT header // since we don't know a total size here, send lengths iteratively // we are done. send a -1 and be done // Receive blob key and compare