flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobClientTest.java
/**
	 * Tests the PUT/GET operations for content-addressable buffers.
	 *
	 * @param blobType
	 * 		whether the BLOB should become permanent or transient
	 */
Tests the PUT/GET operations for content-addressable buffers.
@param blobType
whether the BLOB should become permanent or transient
Tests the PUT / GET operations for content - addressable buffers . @param blobType whether the BLOB should become permanent or transient
private void testContentAddressableBuffer(BlobKey.BlobType blobType)
			throws IOException, InterruptedException {
		BlobClient client = null;
		try {
			byte[] testBuffer = createTestBuffer();
			MessageDigest md = BlobUtils.createMessageDigest();
			md.update(testBuffer);
			byte[] digest = md.digest();
			InetSocketAddress serverAddress = new InetSocketAddress("localhost", getBlobServer().getPort());
			client = new BlobClient(serverAddress, getBlobClientConfig());
			JobID jobId = new JobID();
			BlobKey receivedKey1 = null;
			if (blobType == TRANSIENT_BLOB) {
				receivedKey1 = client.putBuffer(null, testBuffer, 0, testBuffer.length, blobType);
				assertArrayEquals(digest, receivedKey1.getHash());
			}
			BlobKey receivedKey2 = client.putBuffer(jobId, testBuffer, 0, testBuffer.length, blobType);
			assertArrayEquals(digest, receivedKey2.getHash());
			if (blobType == TRANSIENT_BLOB) {
				verifyKeyDifferentHashEquals(receivedKey1, receivedKey2);
			}
			if (blobType == TRANSIENT_BLOB) {
				validateGetAndClose(client.getInternal(null, receivedKey1), testBuffer);
				verifyDeletedEventually(getBlobServer(), null, receivedKey1);
			}
			validateGetAndClose(client.getInternal(jobId, receivedKey2), testBuffer);
			if (blobType == TRANSIENT_BLOB) {
				verifyDeletedEventually(getBlobServer(), jobId, receivedKey2);
			}
			try (InputStream ignored = client.getInternal(null, BlobKey.createKey(blobType))) {
				fail("Expected IOException did not occur");
			}
			catch (IOException fnfe) {
			}
			client = new BlobClient(serverAddress, getBlobClientConfig());
			try (InputStream ignored = client.getInternal(jobId, BlobKey.createKey(blobType))) {
				fail("Expected IOException did not occur");
			}
			catch (IOException fnfe) {
			}
		}
		finally {
			if (client != null) {
				try {
					client.close();
				} catch (Throwable ignored) {}
			}
		}
	}
private void testContentAddressableBuffer ( BlobKey . BlobType blobType ) throws IOException , InterruptedException { BlobClient client = null ; try { byte [ ] testBuffer = createTestBuffer ( ) ; MessageDigest md = BlobUtils . createMessageDigest ( ) ; md . update ( testBuffer ) ; byte [ ] digest = md . digest ( ) ; InetSocketAddress serverAddress = new InetSocketAddress ( "localhost" , getBlobServer ( ) . getPort ( ) ) ; client = new BlobClient ( serverAddress , getBlobClientConfig ( ) ) ; JobID jobId = new JobID ( ) ; BlobKey receivedKey1 = null ; if ( blobType == TRANSIENT_BLOB ) { receivedKey1 = client . putBuffer ( null , testBuffer , 0 , testBuffer . length , blobType ) ; assertArrayEquals ( digest , receivedKey1 . getHash ( ) ) ; } BlobKey receivedKey2 = client . putBuffer ( jobId , testBuffer , 0 , testBuffer . length , blobType ) ; assertArrayEquals ( digest , receivedKey2 . getHash ( ) ) ; if ( blobType == TRANSIENT_BLOB ) { verifyKeyDifferentHashEquals ( receivedKey1 , receivedKey2 ) ; } if ( blobType == TRANSIENT_BLOB ) { validateGetAndClose ( client . getInternal ( null , receivedKey1 ) , testBuffer ) ; verifyDeletedEventually ( getBlobServer ( ) , null , receivedKey1 ) ; } validateGetAndClose ( client . getInternal ( jobId , receivedKey2 ) , testBuffer ) ; if ( blobType == TRANSIENT_BLOB ) { verifyDeletedEventually ( getBlobServer ( ) , jobId , receivedKey2 ) ; } try ( InputStream ignored = client . getInternal ( null , BlobKey . createKey ( blobType ) ) ) { fail ( "Expected IOException did not occur" ) ; } catch ( IOException fnfe ) { } client = new BlobClient ( serverAddress , getBlobClientConfig ( ) ) ; try ( InputStream ignored = client . getInternal ( jobId , BlobKey . createKey ( blobType ) ) ) { fail ( "Expected IOException did not occur" ) ; } catch ( IOException fnfe ) { } } finally { if ( client != null ) { try { client . close ( ) ; } catch ( Throwable ignored ) { } } } }
Tests the PUT/GET operations for content-addressable buffers.
Tests the PUT / GET operations for content - addressable buffers .
// 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 // Check reaction to invalid keys for job-unrelated blobs // expected // Check reaction to invalid keys for job-related blobs // new client needed (closed from failure above) // expected