OpenFileCtxCache { @VisibleForTesting void scan(long streamTimeout) { ArrayList<OpenFileCtx> ctxToRemove = new ArrayList<OpenFileCtx>(); Iterator<Entry<FileHandle, OpenFileCtx>> it = openFileMap.entrySet() .iterator(); if (LOG.isTraceEnabled()) { LOG.trace("openFileMap size:" + size()); } while (it.hasNext()) { Entry<FileHandle, OpenFileCtx> pairs = it.next(); FileHandle handle = pairs.getKey(); OpenFileCtx ctx = pairs.getValue(); if (!ctx.streamCleanup(handle, streamTimeout)) { continue; } synchronized (this) { OpenFileCtx ctx2 = openFileMap.get(handle); if (ctx2 != null) { if (ctx2.streamCleanup(handle, streamTimeout)) { openFileMap.remove(handle); if (LOG.isDebugEnabled()) { LOG.debug("After remove stream " + handle.dumpFileHandle() + ", the stream number:" + size()); } ctxToRemove.add(ctx2); } } } } for (OpenFileCtx ofc : ctxToRemove) { ofc.cleanup(); } } OpenFileCtxCache(NfsConfiguration config, long streamTimeout);   }
@Test public void testScan() throws IOException, InterruptedException { NfsConfiguration conf = new NfsConfiguration(); conf.setInt(NfsConfigKeys.DFS_NFS_MAX_OPEN_FILES_KEY, 2); DFSClient dfsClient = Mockito.mock(DFSClient.class); Nfs3FileAttributes attr = new Nfs3FileAttributes(); HdfsDataOutputStream fos = Mockito.mock(HdfsDataOutputStream.class); Mockito.when(fos.getPos()).thenReturn((long) 0); OpenFileCtx context1 = new OpenFileCtx(fos, attr, "/dumpFilePath", dfsClient, new ShellBasedIdMapping(new NfsConfiguration())); OpenFileCtx context2 = new OpenFileCtx(fos, attr, "/dumpFilePath", dfsClient, new ShellBasedIdMapping(new NfsConfiguration())); OpenFileCtx context3 = new OpenFileCtx(fos, attr, "/dumpFilePath", dfsClient, new ShellBasedIdMapping(new NfsConfiguration())); OpenFileCtx context4 = new OpenFileCtx(fos, attr, "/dumpFilePath", dfsClient, new ShellBasedIdMapping(new NfsConfiguration())); OpenFileCtxCache cache = new OpenFileCtxCache(conf, 10 * 60 * 100); boolean ret = cache.put(new FileHandle(1), context1); assertTrue(ret); ret = cache.put(new FileHandle(2), context2); assertTrue(ret); Thread.sleep(NfsConfigKeys.DFS_NFS_STREAM_TIMEOUT_MIN_DEFAULT + 1); cache.scan(NfsConfigKeys.DFS_NFS_STREAM_TIMEOUT_MIN_DEFAULT); assertTrue(cache.size() == 0); ret = cache.put(new FileHandle(3), context3); assertTrue(ret); ret = cache.put(new FileHandle(4), context4); assertTrue(ret); context3.setActiveStatusForTest(false); cache.scan(NfsConfigKeys.DFS_NFS_STREAM_TIMEOUT_DEFAULT); assertTrue(cache.size() == 1); assertTrue(cache.get(new FileHandle(3)) == null); assertTrue(cache.get(new FileHandle(4)) != null); }