DocumentTextCacheStorageUtils { public static void storeInCache(SparkAvroSaver avroSaver, JavaRDD<DocumentText> toBeStoredEntities, JavaRDD<Fault> toBeStoredFaults, Path cacheRootDir, LockManager lockManager, CacheMetadataManagingProcess cacheManager, Configuration hadoopConf, int numberOfEmittedFiles) throws Exception { lockManager.obtain(cacheRootDir.toString()); try { String newCacheId = cacheManager.generateNewCacheId(hadoopConf, cacheRootDir); try { avroSaver.saveJavaRDD(toBeStoredEntities.repartition(numberOfEmittedFiles), DocumentText.SCHEMA$, getCacheLocation(cacheRootDir, newCacheId, CacheRecordType.text).toString()); avroSaver.saveJavaRDD(toBeStoredFaults.repartition(numberOfEmittedFiles), Fault.SCHEMA$, getCacheLocation(cacheRootDir, newCacheId, CacheRecordType.fault).toString()); cacheManager.writeCacheId(hadoopConf, cacheRootDir, newCacheId); } catch(Exception e) { FileSystem.get(hadoopConf).delete(new Path(cacheRootDir, newCacheId), true); throw e; } } finally { lockManager.release(cacheRootDir.toString()); } } private  DocumentTextCacheStorageUtils(); static Path getCacheLocation(Path cacheRootDir, String cacheId, CacheRecordType cacheRecordType); static JavaRDD<T> getRddOrEmpty(JavaSparkContext sc, SparkAvroLoader avroLoader,
            Path cacheRootDir, String existingCacheId, CacheRecordType cacheRecordType, Class<T> avroRecordClass); static void storeInCache(SparkAvroSaver avroSaver, JavaRDD<DocumentText> toBeStoredEntities, JavaRDD<Fault> toBeStoredFaults, 
            Path cacheRootDir, LockManager lockManager, CacheMetadataManagingProcess cacheManager, 
            Configuration hadoopConf, int numberOfEmittedFiles);  }
@Test(expected = IOException.class) public void testStoreInCacheByCheckingProperLocksHandlingWhenWriteCacheThrowsException() throws Exception { Path cacheRootDir = new Path("some/root/dir"); int numberOfEmittedFiles = 2; String predefinedCacheId = "someCacheId"; doReturn(predefinedCacheId).when(cacheManager).generateNewCacheId(hadoopConf, cacheRootDir); doReturn(FileSystem.DEFAULT_FS).when(hadoopConf).get(FileSystem.FS_DEFAULT_NAME_KEY, FileSystem.DEFAULT_FS); doReturn(toBeStoredRepartitionedEntities).when(toBeStoredEntities).repartition(numberOfEmittedFiles); doReturn(toBeStoredRepartitionedFaults).when(toBeStoredFaults).repartition(numberOfEmittedFiles); doThrow(IOException.class).when(cacheManager).writeCacheId(hadoopConf, cacheRootDir, predefinedCacheId); try { DocumentTextCacheStorageUtils.storeInCache(avroSaver, toBeStoredEntities, toBeStoredFaults, cacheRootDir, lockManager, cacheManager, hadoopConf, numberOfEmittedFiles); } catch (Exception e) { verify(lockManager, times(1)).obtain(cacheRootDir.toString()); verify(lockManager, times(1)).release(cacheRootDir.toString()); throw e; } }