flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobClient.java
/**
	 * Reads the response from the input stream and throws in case of errors.
	 *
	 * @param is
	 * 		stream to read from
	 * @param md
	 * 		message digest to check the response against
	 * @param blobType
	 * 		whether the BLOB should be permanent or transient
	 *
	 * @throws IOException
	 * 		if the response is an error, the message digest does not match or reading the response
	 * 		failed
	 */
Reads the response from the input stream and throws in case of errors.
@param is
stream to read from
@param md
message digest to check the response against
@param blobType
whether the BLOB should be permanent or transient

@throws IOException
if the response is an error, the message digest does not match or reading the response
failed
Reads the response from the input stream and throws in case of errors . @param is stream to read from @param md message digest to check the response against @param blobType whether the BLOB should be permanent or transient @throws IOException if the response is an error the message digest does not match or reading the response failed
private static BlobKey receiveAndCheckPutResponse(
			InputStream is, MessageDigest md, BlobKey.BlobType blobType)
			throws IOException {
		int response = is.read();
		if (response < 0) {
			throw new EOFException("Premature end of response");
		}
		else if (response == RETURN_OKAY) {
			BlobKey remoteKey = BlobKey.readFromInputStream(is);
			byte[] localHash = md.digest();
			if (blobType != remoteKey.getType()) {
				throw new IOException("Detected data corruption during transfer");
			}
			if (!Arrays.equals(localHash, remoteKey.getHash())) {
				throw new IOException("Detected data corruption during transfer");
			}
			return remoteKey;
		}
		else if (response == RETURN_ERROR) {
			Throwable cause = readExceptionFromStream(is);
			throw new IOException("Server side error: " + cause.getMessage(), cause);
		}
		else {
			throw new IOException("Unrecognized response: " + response + '.');
		}
	}
private static BlobKey receiveAndCheckPutResponse ( InputStream is , MessageDigest md , BlobKey . BlobType blobType ) throws IOException { int response = is . read ( ) ; if ( response < 0 ) { throw new EOFException ( "Premature end of response" ) ; } else if ( response == RETURN_OKAY ) { BlobKey remoteKey = BlobKey . readFromInputStream ( is ) ; byte [ ] localHash = md . digest ( ) ; if ( blobType != remoteKey . getType ( ) ) { throw new IOException ( "Detected data corruption during transfer" ) ; } if ( ! Arrays . equals ( localHash , remoteKey . getHash ( ) ) ) { throw new IOException ( "Detected data corruption during transfer" ) ; } return remoteKey ; } else if ( response == RETURN_ERROR ) { Throwable cause = readExceptionFromStream ( is ) ; throw new IOException ( "Server side error: " + cause . getMessage ( ) , cause ) ; } else { throw new IOException ( "Unrecognized response: " + response + '.' ) ; } }
Reads the response from the input stream and throws in case of errors.
Reads the response from the input stream and throws in case of errors .