flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobClientTest.java
/**
	 * Checks the correct result if a GET operation fails during the file download.
	 *
	 * @param jobId
	 * 		job ID or <tt>null</tt> if job-unrelated
	 * @param blobType
	 * 		whether the BLOB should become permanent or transient
	 */
Checks the correct result if a GET operation fails during the file download.
@param jobId
job ID or null if job-unrelated
@param blobType
whether the BLOB should become permanent or transient
Checks the correct result if a GET operation fails during the file download . @param jobId job ID or null if job - unrelated @param blobType whether the BLOB should become permanent or transient
private void testGetFailsDuringStreaming(@Nullable final JobID jobId, BlobKey.BlobType blobType)
			throws IOException {
		try (BlobClient client = new BlobClient(
			new InetSocketAddress("localhost", getBlobServer().getPort()), getBlobClientConfig())) {
			byte[] data = new byte[5000000];
			Random rnd = new Random();
			rnd.nextBytes(data);
			BlobKey key = client.putBuffer(jobId, data, 0, data.length, blobType);
			assertNotNull(key);
			InputStream is = client.getInternal(jobId, key);
			byte[] receiveBuffer = new byte[data.length];
			int firstChunkLen = 50000;
			BlobUtils.readFully(is, receiveBuffer, 0, firstChunkLen, null);
			BlobUtils.readFully(is, receiveBuffer, firstChunkLen, firstChunkLen, null);
			for (BlobServerConnection conn : getBlobServer().getCurrentActiveConnections()) {
				conn.close();
			}
			try {
				BlobUtils.readFully(is, receiveBuffer, 2 * firstChunkLen, data.length - 2 * firstChunkLen, null);
				assertArrayEquals(data, receiveBuffer);
			}
			catch (IOException e) {
			}
		}
	}
private void testGetFailsDuringStreaming ( @ Nullable final JobID jobId , BlobKey . BlobType blobType ) throws IOException { try ( BlobClient client = new BlobClient ( new InetSocketAddress ( "localhost" , getBlobServer ( ) . getPort ( ) ) , getBlobClientConfig ( ) ) ) { byte [ ] data = new byte [ 5000000 ] ; Random rnd = new Random ( ) ; rnd . nextBytes ( data ) ; BlobKey key = client . putBuffer ( jobId , data , 0 , data . length , blobType ) ; assertNotNull ( key ) ; InputStream is = client . getInternal ( jobId , key ) ; byte [ ] receiveBuffer = new byte [ data . length ] ; int firstChunkLen = 50000 ; BlobUtils . readFully ( is , receiveBuffer , 0 , firstChunkLen , null ) ; BlobUtils . readFully ( is , receiveBuffer , firstChunkLen , firstChunkLen , null ) ; for ( BlobServerConnection conn : getBlobServer ( ) . getCurrentActiveConnections ( ) ) { conn . close ( ) ; } try { BlobUtils . readFully ( is , receiveBuffer , 2 * firstChunkLen , data . length - 2 * firstChunkLen , null ) ; assertArrayEquals ( data , receiveBuffer ) ; } catch ( IOException e ) { } } }
Checks the correct result if a GET operation fails during the file download.
Checks the correct result if a GET operation fails during the file download .
// put content addressable (like libraries) // issue a GET request that succeeds // shut down the server // we tolerate that this succeeds, as the receiver socket may have buffered // everything already, but in this case, also verify the contents // expected