flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManagerTest.java
/**
	 * Tests that the {@link BlobLibraryCacheManager} cleans up after calling {@link
	 * BlobLibraryCacheManager#unregisterJob(JobID)}.
	 */
Tests that the  BlobLibraryCacheManager cleans up after calling {@link
BlobLibraryCacheManager#unregisterJob(JobID)}.
Tests that the BlobLibraryCacheManager cleans up after calling { @link BlobLibraryCacheManager#unregisterJob ( JobID ) } .
@Test
	public void testLibraryCacheManagerJobCleanup() throws IOException, InterruptedException {
		JobID jobId1 = new JobID();
		JobID jobId2 = new JobID();
		List<PermanentBlobKey> keys1 = new ArrayList<>();
		List<PermanentBlobKey> keys2 = new ArrayList<>();
		BlobServer server = null;
		PermanentBlobCache cache = null;
		BlobLibraryCacheManager libCache = null;
		final byte[] buf = new byte[128];
		try {
			Configuration config = new Configuration();
			config.setString(BlobServerOptions.STORAGE_DIRECTORY,
				temporaryFolder.newFolder().getAbsolutePath());
			config.setLong(BlobServerOptions.CLEANUP_INTERVAL, 1L);
			server = new BlobServer(config, new VoidBlobStore());
			server.start();
			InetSocketAddress serverAddress = new InetSocketAddress("localhost", server.getPort());
			cache = new PermanentBlobCache(serverAddress, config, new VoidBlobStore());
			keys1.add(server.putPermanent(jobId1, buf));
			buf[0] += 1;
			keys1.add(server.putPermanent(jobId1, buf));
			keys2.add(server.putPermanent(jobId2, buf));
			libCache = new BlobLibraryCacheManager(cache, FlinkUserCodeClassLoaders.ResolveOrder.CHILD_FIRST, new String[0]);
			cache.registerJob(jobId1);
			cache.registerJob(jobId2);
			assertEquals(0, libCache.getNumberOfManagedJobs());
			assertEquals(0, libCache.getNumberOfReferenceHolders(jobId1));
			checkFileCountForJob(2, jobId1, server);
			checkFileCountForJob(0, jobId1, cache);
			checkFileCountForJob(1, jobId2, server);
			checkFileCountForJob(0, jobId2, cache);
			libCache.registerJob(jobId1, keys1, Collections.<URL>emptyList());
			ClassLoader classLoader1 = libCache.getClassLoader(jobId1);
			assertEquals(1, libCache.getNumberOfManagedJobs());
			assertEquals(1, libCache.getNumberOfReferenceHolders(jobId1));
			assertEquals(0, libCache.getNumberOfReferenceHolders(jobId2));
			assertEquals(2, checkFilesExist(jobId1, keys1, cache, true));
			checkFileCountForJob(2, jobId1, server);
			checkFileCountForJob(2, jobId1, cache);
			assertEquals(0, checkFilesExist(jobId2, keys2, cache, false));
			checkFileCountForJob(1, jobId2, server);
			checkFileCountForJob(0, jobId2, cache);
			libCache.registerJob(jobId2, keys2, Collections.<URL>emptyList());
			ClassLoader classLoader2 = libCache.getClassLoader(jobId2);
			assertNotEquals(classLoader1, classLoader2);
			try {
				libCache.registerJob(jobId2, keys1, Collections.<URL>emptyList());
				fail("Should fail with an IllegalStateException");
			}
			catch (IllegalStateException e) {
			}
			try {
				libCache.registerJob(
					jobId2, keys2,
					Collections.singletonList(new URL("file:///tmp/does-not-exist")));
				fail("Should fail with an IllegalStateException");
			}
			catch (IllegalStateException e) {
			}
			assertEquals(2, libCache.getNumberOfManagedJobs());
			assertEquals(1, libCache.getNumberOfReferenceHolders(jobId1));
			assertEquals(1, libCache.getNumberOfReferenceHolders(jobId2));
			assertEquals(2, checkFilesExist(jobId1, keys1, cache, true));
			checkFileCountForJob(2, jobId1, server);
			checkFileCountForJob(2, jobId1, cache);
			assertEquals(1, checkFilesExist(jobId2, keys2, cache, true));
			checkFileCountForJob(1, jobId2, server);
			checkFileCountForJob(1, jobId2, cache);
			libCache.unregisterJob(jobId1);
			assertEquals(1, libCache.getNumberOfManagedJobs());
			assertEquals(0, libCache.getNumberOfReferenceHolders(jobId1));
			assertEquals(1, libCache.getNumberOfReferenceHolders(jobId2));
			assertEquals(2, checkFilesExist(jobId1, keys1, cache, true));
			checkFileCountForJob(2, jobId1, server);
			checkFileCountForJob(2, jobId1, cache);
			assertEquals(1, checkFilesExist(jobId2, keys2, cache, true));
			checkFileCountForJob(1, jobId2, server);
			checkFileCountForJob(1, jobId2, cache);
			libCache.unregisterJob(jobId2);
			assertEquals(0, libCache.getNumberOfManagedJobs());
			assertEquals(0, libCache.getNumberOfReferenceHolders(jobId1));
			assertEquals(0, libCache.getNumberOfReferenceHolders(jobId2));
			assertEquals(2, checkFilesExist(jobId1, keys1, cache, true));
			checkFileCountForJob(2, jobId1, server);
			checkFileCountForJob(2, jobId1, cache);
			assertEquals(1, checkFilesExist(jobId2, keys2, cache, true));
			checkFileCountForJob(1, jobId2, server);
			checkFileCountForJob(1, jobId2, cache);
		}
		finally {
			if (libCache != null) {
				libCache.shutdown();
			}
			if (cache != null) {
				cache.close();
			}
			if (server != null) {
				server.close();
			}
		}
	}
@ Test public void testLibraryCacheManagerJobCleanup ( ) throws IOException , InterruptedException { JobID jobId1 = new JobID ( ) ; JobID jobId2 = new JobID ( ) ; List < PermanentBlobKey > keys1 = new ArrayList < > ( ) ; List < PermanentBlobKey > keys2 = new ArrayList < > ( ) ; BlobServer server = null ; PermanentBlobCache cache = null ; BlobLibraryCacheManager libCache = null ; final byte [ ] buf = new byte [ 128 ] ; try { Configuration config = new Configuration ( ) ; config . setString ( BlobServerOptions . STORAGE_DIRECTORY , temporaryFolder . newFolder ( ) . getAbsolutePath ( ) ) ; config . setLong ( BlobServerOptions . CLEANUP_INTERVAL , 1L ) ; server = new BlobServer ( config , new VoidBlobStore ( ) ) ; server . start ( ) ; InetSocketAddress serverAddress = new InetSocketAddress ( "localhost" , server . getPort ( ) ) ; cache = new PermanentBlobCache ( serverAddress , config , new VoidBlobStore ( ) ) ; keys1 . add ( server . putPermanent ( jobId1 , buf ) ) ; buf [ 0 ] += 1 ; keys1 . add ( server . putPermanent ( jobId1 , buf ) ) ; keys2 . add ( server . putPermanent ( jobId2 , buf ) ) ; libCache = new BlobLibraryCacheManager ( cache , FlinkUserCodeClassLoaders . ResolveOrder . CHILD_FIRST , new String [ 0 ] ) ; cache . registerJob ( jobId1 ) ; cache . registerJob ( jobId2 ) ; assertEquals ( 0 , libCache . getNumberOfManagedJobs ( ) ) ; assertEquals ( 0 , libCache . getNumberOfReferenceHolders ( jobId1 ) ) ; checkFileCountForJob ( 2 , jobId1 , server ) ; checkFileCountForJob ( 0 , jobId1 , cache ) ; checkFileCountForJob ( 1 , jobId2 , server ) ; checkFileCountForJob ( 0 , jobId2 , cache ) ; libCache . registerJob ( jobId1 , keys1 , Collections . < URL > emptyList ( ) ) ; ClassLoader classLoader1 = libCache . getClassLoader ( jobId1 ) ; assertEquals ( 1 , libCache . getNumberOfManagedJobs ( ) ) ; assertEquals ( 1 , libCache . getNumberOfReferenceHolders ( jobId1 ) ) ; assertEquals ( 0 , libCache . getNumberOfReferenceHolders ( jobId2 ) ) ; assertEquals ( 2 , checkFilesExist ( jobId1 , keys1 , cache , true ) ) ; checkFileCountForJob ( 2 , jobId1 , server ) ; checkFileCountForJob ( 2 , jobId1 , cache ) ; assertEquals ( 0 , checkFilesExist ( jobId2 , keys2 , cache , false ) ) ; checkFileCountForJob ( 1 , jobId2 , server ) ; checkFileCountForJob ( 0 , jobId2 , cache ) ; libCache . registerJob ( jobId2 , keys2 , Collections . < URL > emptyList ( ) ) ; ClassLoader classLoader2 = libCache . getClassLoader ( jobId2 ) ; assertNotEquals ( classLoader1 , classLoader2 ) ; try { libCache . registerJob ( jobId2 , keys1 , Collections . < URL > emptyList ( ) ) ; fail ( "Should fail with an IllegalStateException" ) ; } catch ( IllegalStateException e ) { } try { libCache . registerJob ( jobId2 , keys2 , Collections . singletonList ( new URL ( "file:///tmp/does-not-exist" ) ) ) ; fail ( "Should fail with an IllegalStateException" ) ; } catch ( IllegalStateException e ) { } assertEquals ( 2 , libCache . getNumberOfManagedJobs ( ) ) ; assertEquals ( 1 , libCache . getNumberOfReferenceHolders ( jobId1 ) ) ; assertEquals ( 1 , libCache . getNumberOfReferenceHolders ( jobId2 ) ) ; assertEquals ( 2 , checkFilesExist ( jobId1 , keys1 , cache , true ) ) ; checkFileCountForJob ( 2 , jobId1 , server ) ; checkFileCountForJob ( 2 , jobId1 , cache ) ; assertEquals ( 1 , checkFilesExist ( jobId2 , keys2 , cache , true ) ) ; checkFileCountForJob ( 1 , jobId2 , server ) ; checkFileCountForJob ( 1 , jobId2 , cache ) ; libCache . unregisterJob ( jobId1 ) ; assertEquals ( 1 , libCache . getNumberOfManagedJobs ( ) ) ; assertEquals ( 0 , libCache . getNumberOfReferenceHolders ( jobId1 ) ) ; assertEquals ( 1 , libCache . getNumberOfReferenceHolders ( jobId2 ) ) ; assertEquals ( 2 , checkFilesExist ( jobId1 , keys1 , cache , true ) ) ; checkFileCountForJob ( 2 , jobId1 , server ) ; checkFileCountForJob ( 2 , jobId1 , cache ) ; assertEquals ( 1 , checkFilesExist ( jobId2 , keys2 , cache , true ) ) ; checkFileCountForJob ( 1 , jobId2 , server ) ; checkFileCountForJob ( 1 , jobId2 , cache ) ; libCache . unregisterJob ( jobId2 ) ; assertEquals ( 0 , libCache . getNumberOfManagedJobs ( ) ) ; assertEquals ( 0 , libCache . getNumberOfReferenceHolders ( jobId1 ) ) ; assertEquals ( 0 , libCache . getNumberOfReferenceHolders ( jobId2 ) ) ; assertEquals ( 2 , checkFilesExist ( jobId1 , keys1 , cache , true ) ) ; checkFileCountForJob ( 2 , jobId1 , server ) ; checkFileCountForJob ( 2 , jobId1 , cache ) ; assertEquals ( 1 , checkFilesExist ( jobId2 , keys2 , cache , true ) ) ; checkFileCountForJob ( 1 , jobId2 , server ) ; checkFileCountForJob ( 1 , jobId2 , cache ) ; } finally { if ( libCache != null ) { libCache . shutdown ( ) ; } if ( cache != null ) { cache . close ( ) ; } if ( server != null ) { server . close ( ) ; } } }
Tests that the {@link BlobLibraryCacheManager} cleans up after calling {@link
 BlobLibraryCacheManager#unregisterJob(JobID)}.
Tests that the { @link BlobLibraryCacheManager } cleans up after calling { @link BlobLibraryCacheManager#unregisterJob ( JobID ) } .
// that's what we want // that's what we want // only PermanentBlobCache#releaseJob() calls clean up files (tested in BlobCacheCleanupTest etc. // should have been closed by the libraryCacheManager, but just in case