IndexCopier implements CopyOnReadStatsMBean, Closeable { public Directory wrapForRead(String indexPath, LuceneIndexDefinition definition, Directory remote, String dirName) throws IOException { Directory local = createLocalDirForIndexReader(indexPath, definition, dirName); checkIntegrity(indexPath, local, remote); return new CopyOnReadDirectory(this, remote, local, prefetchEnabled, indexPath, executor); } IndexCopier(Executor executor, File indexRootDir);  IndexCopier(Executor executor, File indexRootDir, boolean prefetchEnabled); Directory wrapForRead(String indexPath, LuceneIndexDefinition definition,
                                 Directory remote, String dirName); Directory wrapForWrite(LuceneIndexDefinition definition, Directory remote,
                                  boolean reindexMode, String dirName,
                                  COWDirectoryTracker cowDirectoryTracker); @Override void close(); boolean isClosed(); File getIndexDir(IndexDefinition definition, String indexPath, String dirName); boolean deleteFile(Directory dir, String fileName, boolean copiedFromRemote); static long getNewestLocalFSTimestampFor(Set<String> names, Directory localDir); static boolean isFileModifiedBefore(String name, Directory localDir, long millis); long startCopy(LocalIndexFile file); boolean isCopyInProgress(LocalIndexFile file); void waitForCopyCompletion(LocalIndexFile file, long timeoutMillis); void doneCopy(LocalIndexFile file, long start); void skippedUpload(long skippedFilesSize); void scheduledForCopy(); void copyDone(); void readFromRemote(boolean reader); void readFromLocal(boolean reader); void foundInvalidFile(); @Override TabularData getIndexPathMapping(); @Override boolean isPrefetchEnabled(); @Override int getReaderLocalReadCount(); @Override int getReaderRemoteReadCount(); @Override int getWriterLocalReadCount(); @Override int getWriterRemoteReadCount(); int getInvalidFileCount(); @Override String getDownloadSize(); @Override long getDownloadTime(); @Override int getDownloadCount(); @Override int getUploadCount(); @Override String getUploadSize(); @Override long getUploadTime(); @Override String getLocalIndexSize(); @Override long getLocalIndexDirSize(); @Override String[] getGarbageDetails(); @Override String getGarbageSize(); @Override int getScheduledForCopyCount(); @Override int getCopyInProgressCount(); @Override String getCopyInProgressSize(); @Override int getMaxCopyInProgressCount(); @Override int getMaxScheduledForCopyCount(); String getSkippedFromUploadSize(); @Override String[] getCopyInProgressDetails(); @Override int getDeletedFilesCount(); @Override String getGarbageCollectedSize(); static final Set<String> REMOTE_ONLY; }
@Test public void basicTest() throws Exception{ Directory baseDir = new RAMDirectory(); LuceneIndexDefinition defn = new LuceneIndexDefinition(root, builder.getNodeState(), "/foo"); IndexCopier c1 = new RAMIndexCopier(baseDir, sameThreadExecutor(), getWorkDir()); Directory remote = new RAMDirectory(); Directory wrapped = c1.wrapForRead("/foo", defn, remote, INDEX_DATA_CHILD_NAME); byte[] t1 = writeFile(remote , "t1"); byte[] t2 = writeFile(remote , "t2"); assertEquals(2, wrapped.listAll().length); assertTrue(wrapped.fileExists("t1")); assertTrue(wrapped.fileExists("t2")); assertEquals(t1.length, wrapped.fileLength("t1")); assertEquals(t2.length, wrapped.fileLength("t2")); readAndAssert(wrapped, "t1", t1); assertTrue(baseDir.fileExists("t1")); }
@Test public void basicTestWithPrefetch() throws Exception{ final List<String> syncedFiles = Lists.newArrayList(); Directory baseDir = new RAMDirectory(){ @Override public void sync(Collection<String> names) throws IOException { syncedFiles.addAll(names); super.sync(names); } }; LuceneIndexDefinition defn = new LuceneIndexDefinition(root, builder.getNodeState(), "/foo"); IndexCopier c1 = new RAMIndexCopier(baseDir, sameThreadExecutor(), getWorkDir(), true); Directory remote = new RAMDirectory(); byte[] t1 = writeFile(remote, "t1"); byte[] t2 = writeFile(remote , "t2"); Directory wrapped = c1.wrapForRead("/foo", defn, remote, INDEX_DATA_CHILD_NAME); assertEquals(2, wrapped.listAll().length); assertThat(syncedFiles, containsInAnyOrder("t1", "t2")); assertTrue(wrapped.fileExists("t1")); assertTrue(wrapped.fileExists("t2")); assertTrue(baseDir.fileExists("t1")); assertTrue(baseDir.fileExists("t2")); assertEquals(t1.length, wrapped.fileLength("t1")); assertEquals(t2.length, wrapped.fileLength("t2")); readAndAssert(wrapped, "t1", t1); }
@Test public void nonExistentFile() throws Exception{ Directory baseDir = new RAMDirectory(); LuceneIndexDefinition defn = new LuceneIndexDefinition(root, builder.getNodeState(), "/foo"); CollectingExecutor executor = new CollectingExecutor(); IndexCopier c1 = new RAMIndexCopier(baseDir, executor, getWorkDir(), true); Directory remote = new RAMDirectory(); Directory wrapped = c1.wrapForRead("/foo", defn, remote, INDEX_DATA_CHILD_NAME); try { wrapped.openInput("foo.txt", IOContext.DEFAULT); fail(); } catch(FileNotFoundException ignore){ } assertEquals(0, executor.commands.size()); }
@Test public void reuseLocalDir() throws Exception{ Directory baseDir = new RAMDirectory(); LuceneIndexDefinition defn = new LuceneIndexDefinition(root, builder.getNodeState(), "/foo"); IndexCopier c1 = new RAMIndexCopier(baseDir, sameThreadExecutor(), getWorkDir()); FileTrackingDirectory remote = new FileTrackingDirectory(); Directory wrapped = c1.wrapForRead("/foo", defn, remote, INDEX_DATA_CHILD_NAME); byte[] t1 = writeFile(remote, "t1"); readAndAssert(wrapped, "t1", t1); assertEquals(1, remote.openedFiles.size()); Directory wrapped2 = c1.wrapForRead("/foo", defn, remote, INDEX_DATA_CHILD_NAME); remote.reset(); readAndAssert(wrapped2, "t1", t1); assertEquals(0, remote.openedFiles.size()); Directory wrapped3 = c1.wrapForRead("/foo", defn, remote, INDEX_DATA_CHILD_NAME); remote.reset(); writeFile(baseDir, "t1"); readAndAssert(wrapped3, "t1", t1); assertEquals(1, remote.openedFiles.size()); }
@Test public void deleteCorruptedFile() throws Exception{ Directory baseDir = new RAMDirectory(); LuceneIndexDefinition defn = new LuceneIndexDefinition(root, builder.getNodeState(), "/foo"); RAMIndexCopier c1 = new RAMIndexCopier(baseDir, sameThreadExecutor(), getWorkDir()); Directory remote = new RAMDirectory(){ @Override public IndexInput openInput(String name, IOContext context) throws IOException { throw new IllegalStateException("boom"); } }; String fileName = "failed.txt"; Directory wrapped = c1.wrapForRead("/foo", defn, remote, INDEX_DATA_CHILD_NAME); byte[] t1 = writeFile(remote , fileName); try { readAndAssert(wrapped, fileName, t1); fail("Read of file should have failed"); } catch (IllegalStateException ignore){ } assertFalse(c1.baseDir.fileExists(fileName)); }