flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobClientTest.java
/**
	 * Tests the PUT/GET operations for content-addressable streams.
	 *
	 * @param blobType
	 * 		whether the BLOB should become permanent or transient
	 */
Tests the PUT/GET operations for content-addressable streams.
@param blobType
whether the BLOB should become permanent or transient
Tests the PUT / GET operations for content - addressable streams . @param blobType whether the BLOB should become permanent or transient
private void testContentAddressableStream(BlobKey.BlobType blobType)
			throws IOException, InterruptedException {
		File testFile = temporaryFolder.newFile();
		byte[] digest = prepareTestFile(testFile);
		InputStream is = null;
		try (BlobClient client = new BlobClient(new InetSocketAddress("localhost", getBlobServer().getPort()), getBlobClientConfig())) {
			JobID jobId = new JobID();
			BlobKey receivedKey1 = null;
			if (blobType == TRANSIENT_BLOB) {
				is = new FileInputStream(testFile);
				receivedKey1 = client.putInputStream(null, is, blobType);
				assertArrayEquals(digest, receivedKey1.getHash());
			}
			is = new FileInputStream(testFile);
			BlobKey receivedKey2 = client.putInputStream(jobId, is, blobType);
			is.close();
			is = null;
			if (blobType == TRANSIENT_BLOB) {
				verifyKeyDifferentHashEquals(receivedKey1, receivedKey2);
				validateGetAndClose(client.getInternal(null, receivedKey1), testFile);
				verifyDeletedEventually(getBlobServer(), null, receivedKey1);
			}
			validateGetAndClose(client.getInternal(jobId, receivedKey2), testFile);
			if (blobType == TRANSIENT_BLOB) {
				verifyDeletedEventually(getBlobServer(), jobId, receivedKey2);
			}
		} finally {
			if (is != null) {
				try {
					is.close();
				} catch (Throwable ignored) {}
			}
		}
	}
private void testContentAddressableStream ( BlobKey . BlobType blobType ) throws IOException , InterruptedException { File testFile = temporaryFolder . newFile ( ) ; byte [ ] digest = prepareTestFile ( testFile ) ; InputStream is = null ; try ( BlobClient client = new BlobClient ( new InetSocketAddress ( "localhost" , getBlobServer ( ) . getPort ( ) ) , getBlobClientConfig ( ) ) ) { JobID jobId = new JobID ( ) ; BlobKey receivedKey1 = null ; if ( blobType == TRANSIENT_BLOB ) { is = new FileInputStream ( testFile ) ; receivedKey1 = client . putInputStream ( null , is , blobType ) ; assertArrayEquals ( digest , receivedKey1 . getHash ( ) ) ; } is = new FileInputStream ( testFile ) ; BlobKey receivedKey2 = client . putInputStream ( jobId , is , blobType ) ; is . close ( ) ; is = null ; if ( blobType == TRANSIENT_BLOB ) { verifyKeyDifferentHashEquals ( receivedKey1 , receivedKey2 ) ; validateGetAndClose ( client . getInternal ( null , receivedKey1 ) , testFile ) ; verifyDeletedEventually ( getBlobServer ( ) , null , receivedKey1 ) ; } validateGetAndClose ( client . getInternal ( jobId , receivedKey2 ) , testFile ) ; if ( blobType == TRANSIENT_BLOB ) { verifyDeletedEventually ( getBlobServer ( ) , jobId , receivedKey2 ) ; } } finally { if ( is != null ) { try { is . close ( ) ; } catch ( Throwable ignored ) { } } } }
Tests the PUT/GET operations for content-addressable streams.
Tests the PUT / GET operations for content - addressable streams .
// Store the data (job-unrelated) // try again with a job-related BLOB: // Retrieve the data (job-unrelated) // transient BLOBs should be deleted from the server, eventually // job-related // transient BLOBs should be deleted from the server, eventually