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 byte buffer to the BLOB server.
	 *
	 * @param jobId
	 * 		the ID of the job the BLOB belongs to (or <tt>null</tt> if job-unrelated)
	 * @param value
	 * 		the buffer to read the data from
	 * @param offset
	 * 		the read offset within the buffer
	 * @param len
	 * 		the number of bytes to read from the buffer
	 * @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 byte buffer to the BLOB server.

@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 byte buffer to the BLOB server . @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 putBuffer(
			@Nullable JobID jobId, byte[] value, int offset, int len, 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(value);
		if (LOG.isDebugEnabled()) {
			LOG.debug("PUT BLOB buffer (" + len + " bytes) to " + socket.getLocalSocketAddress() + ".");
		}
		try {
			final OutputStream os = this.socket.getOutputStream();
			final MessageDigest md = BlobUtils.createMessageDigest();
			sendPutHeader(os, jobId, blobType);
			int remainingBytes = len;
			while (remainingBytes > 0) {
				final int bytesToSend = Math.min(BUFFER_SIZE, remainingBytes);
				writeLength(bytesToSend, os);
				os.write(value, offset, bytesToSend);
				md.update(value, offset, bytesToSend);
				remainingBytes -= bytesToSend;
				offset += bytesToSend;
			}
			writeLength(-1, os);
			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 putBuffer ( @ Nullable JobID jobId , byte [ ] value , int offset , int len , 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 ( value ) ; if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "PUT BLOB buffer (" + len + " bytes) to " + socket . getLocalSocketAddress ( ) + "." ) ; } try { final OutputStream os = this . socket . getOutputStream ( ) ; final MessageDigest md = BlobUtils . createMessageDigest ( ) ; sendPutHeader ( os , jobId , blobType ) ; int remainingBytes = len ; while ( remainingBytes > 0 ) { final int bytesToSend = Math . min ( BUFFER_SIZE , remainingBytes ) ; writeLength ( bytesToSend , os ) ; os . write ( value , offset , bytesToSend ) ; md . update ( value , offset , bytesToSend ) ; remainingBytes -= bytesToSend ; offset += bytesToSend ; } writeLength ( - 1 , os ) ; 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 byte buffer to the BLOB server.
Uploads data from the given byte buffer to the BLOB server .
// Send the PUT header // Send the value in iterations of BUFFER_SIZE // want a common code path for byte[] and InputStream at the BlobServer // -> since for InputStream we don't know a total size beforehand, send lengths iteratively // Update the message digest // send -1 as the stream end // Receive blob key and compare