@Test public void copyToLocalBeforeWrite() throws Exception { GarbageCollectableBlobStore blobStore = new MemoryBlobStore(); LuceneIndexDefinition def = new LuceneIndexDefinition(ns.getRoot(), ns.getRoot(), "/foo"); NodeBuilder builder = ns.getRoot().builder(); Directory dir = new DefaultDirectoryFactory(copier, blobStore).newInstance(def, builder.child("foo"), INDEX_DATA_CHILD_NAME, false); addFiles(dir); dir.close(); File dirName = copier.getIndexDir(def, def.getIndexPath(), INDEX_DATA_CHILD_NAME); long oldSize = FileUtils.sizeOfDirectory(dirName.getParentFile()); FileUtils.deleteQuietly(dirName); assertEquals(0, FileUtils.sizeOfDirectory(dirName.getParentFile())); dir = new DefaultDirectoryFactory(copier, blobStore).newInstance(def, builder.child("foo"), INDEX_DATA_CHILD_NAME, false); long newSize = FileUtils.sizeOfDirectory(dirName.getParentFile()); assertEquals(oldSize, newSize); dir.close(); }
@Override public void close() throws IOException { if (isClosed()) { return; } int pendingCopies = queue.size(); addTask(STOP); try { long start = PERF_LOGGER.start(); while (!copyDone.await(10, TimeUnit.SECONDS)) { if (indexCopier.isClosed()) { throw new IndexCopierClosedException("IndexCopier found to be closed " + "while processing copy task for" + remote.toString()); } } PERF_LOGGER.end(start, -1, "[COW][{}] Completed pending copying task {}", indexPath, pendingCopies); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new IOException(e); } Throwable t = errorInCopy.get(); if (t != null){ throw new IOException("Error occurred while copying files for " + indexPath, t); } checkArgument(queue.isEmpty(), "Copy queue still " + "has pending task left [%d]. %s", queue.size(), queue); long skippedFilesSize = getSkippedFilesSize(); for (String fileName : deletedFilesLocal){ deleteLocalFile(fileName); } indexCopier.skippedUpload(skippedFilesSize); String msg = "[COW][{}] CopyOnWrite stats : Skipped copying {} files with total size {}"; if ((reindexMode && skippedFilesSize > 0) || skippedFilesSize > 10 * FileUtils.ONE_MB){ log.info(msg, indexPath, skippedFiles.size(), humanReadableByteCount(skippedFilesSize)); } else { log.debug(msg, indexPath, skippedFiles.size(), humanReadableByteCount(skippedFilesSize)); } if (log.isTraceEnabled()){ log.trace("[COW][{}] File listing - Upon completion {}", indexPath, Arrays.toString(remote.listAll())); } local.close(); remote.close(); closed = true; }
CopyOnWriteDirectory extends FilterDirectory { @Override public void close() throws IOException { if (isClosed()) { return; } int pendingCopies = queue.size(); addTask(STOP); try { long start = PERF_LOGGER.start(); while (!copyDone.await(10, TimeUnit.SECONDS)) { if (indexCopier.isClosed()) { throw new IndexCopierClosedException("IndexCopier found to be closed " + "while processing copy task for" + remote.toString()); } } PERF_LOGGER.end(start, -1, "[COW][{}] Completed pending copying task {}", indexPath, pendingCopies); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new IOException(e); } Throwable t = errorInCopy.get(); if (t != null){ throw new IOException("Error occurred while copying files for " + indexPath, t); } checkArgument(queue.isEmpty(), "Copy queue still " + "has pending task left [%d]. %s", queue.size(), queue); long skippedFilesSize = getSkippedFilesSize(); for (String fileName : deletedFilesLocal){ deleteLocalFile(fileName); } indexCopier.skippedUpload(skippedFilesSize); String msg = "[COW][{}] CopyOnWrite stats : Skipped copying {} files with total size {}"; if ((reindexMode && skippedFilesSize > 0) || skippedFilesSize > 10 * FileUtils.ONE_MB){ log.info(msg, indexPath, skippedFiles.size(), humanReadableByteCount(skippedFilesSize)); } else { log.debug(msg, indexPath, skippedFiles.size(), humanReadableByteCount(skippedFilesSize)); } if (log.isTraceEnabled()){ log.trace("[COW][{}] File listing - Upon completion {}", indexPath, Arrays.toString(remote.listAll())); } local.close(); remote.close(); closed = true; } }
CopyOnWriteDirectory extends FilterDirectory { @Override public void close() throws IOException { if (isClosed()) { return; } int pendingCopies = queue.size(); addTask(STOP); try { long start = PERF_LOGGER.start(); while (!copyDone.await(10, TimeUnit.SECONDS)) { if (indexCopier.isClosed()) { throw new IndexCopierClosedException("IndexCopier found to be closed " + "while processing copy task for" + remote.toString()); } } PERF_LOGGER.end(start, -1, "[COW][{}] Completed pending copying task {}", indexPath, pendingCopies); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new IOException(e); } Throwable t = errorInCopy.get(); if (t != null){ throw new IOException("Error occurred while copying files for " + indexPath, t); } checkArgument(queue.isEmpty(), "Copy queue still " + "has pending task left [%d]. %s", queue.size(), queue); long skippedFilesSize = getSkippedFilesSize(); for (String fileName : deletedFilesLocal){ deleteLocalFile(fileName); } indexCopier.skippedUpload(skippedFilesSize); String msg = "[COW][{}] CopyOnWrite stats : Skipped copying {} files with total size {}"; if ((reindexMode && skippedFilesSize > 0) || skippedFilesSize > 10 * FileUtils.ONE_MB){ log.info(msg, indexPath, skippedFiles.size(), humanReadableByteCount(skippedFilesSize)); } else { log.debug(msg, indexPath, skippedFiles.size(), humanReadableByteCount(skippedFilesSize)); } if (log.isTraceEnabled()){ log.trace("[COW][{}] File listing - Upon completion {}", indexPath, Arrays.toString(remote.listAll())); } local.close(); remote.close(); closed = true; } CopyOnWriteDirectory(IndexCopier indexCopier, Directory remote, Directory local, boolean reindexMode,
                                String indexPath, Executor executor); }
CopyOnWriteDirectory extends FilterDirectory { @Override public void close() throws IOException { if (isClosed()) { return; } int pendingCopies = queue.size(); addTask(STOP); try { long start = PERF_LOGGER.start(); while (!copyDone.await(10, TimeUnit.SECONDS)) { if (indexCopier.isClosed()) { throw new IndexCopierClosedException("IndexCopier found to be closed " + "while processing copy task for" + remote.toString()); } } PERF_LOGGER.end(start, -1, "[COW][{}] Completed pending copying task {}", indexPath, pendingCopies); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new IOException(e); } Throwable t = errorInCopy.get(); if (t != null){ throw new IOException("Error occurred while copying files for " + indexPath, t); } checkArgument(queue.isEmpty(), "Copy queue still " + "has pending task left [%d]. %s", queue.size(), queue); long skippedFilesSize = getSkippedFilesSize(); for (String fileName : deletedFilesLocal){ deleteLocalFile(fileName); } indexCopier.skippedUpload(skippedFilesSize); String msg = "[COW][{}] CopyOnWrite stats : Skipped copying {} files with total size {}"; if ((reindexMode && skippedFilesSize > 0) || skippedFilesSize > 10 * FileUtils.ONE_MB){ log.info(msg, indexPath, skippedFiles.size(), humanReadableByteCount(skippedFilesSize)); } else { log.debug(msg, indexPath, skippedFiles.size(), humanReadableByteCount(skippedFilesSize)); } if (log.isTraceEnabled()){ log.trace("[COW][{}] File listing - Upon completion {}", indexPath, Arrays.toString(remote.listAll())); } local.close(); remote.close(); closed = true; } CopyOnWriteDirectory(IndexCopier indexCopier, Directory remote, Directory local, boolean reindexMode,
                                String indexPath, Executor executor); @Override String[] listAll(); @Override boolean fileExists(String name); @Override void deleteFile(String name); @Override long fileLength(String name); @Override IndexOutput createOutput(String name, IOContext context); @Override void sync(Collection<String> names); @Override IndexInput openInput(String name, IOContext context); boolean isClosed(); @Override void close(); @Override String toString(); }
CopyOnWriteDirectory extends FilterDirectory { @Override public void close() throws IOException { if (isClosed()) { return; } int pendingCopies = queue.size(); addTask(STOP); try { long start = PERF_LOGGER.start(); while (!copyDone.await(10, TimeUnit.SECONDS)) { if (indexCopier.isClosed()) { throw new IndexCopierClosedException("IndexCopier found to be closed " + "while processing copy task for" + remote.toString()); } } PERF_LOGGER.end(start, -1, "[COW][{}] Completed pending copying task {}", indexPath, pendingCopies); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new IOException(e); } Throwable t = errorInCopy.get(); if (t != null){ throw new IOException("Error occurred while copying files for " + indexPath, t); } checkArgument(queue.isEmpty(), "Copy queue still " + "has pending task left [%d]. %s", queue.size(), queue); long skippedFilesSize = getSkippedFilesSize(); for (String fileName : deletedFilesLocal){ deleteLocalFile(fileName); } indexCopier.skippedUpload(skippedFilesSize); String msg = "[COW][{}] CopyOnWrite stats : Skipped copying {} files with total size {}"; if ((reindexMode && skippedFilesSize > 0) || skippedFilesSize > 10 * FileUtils.ONE_MB){ log.info(msg, indexPath, skippedFiles.size(), humanReadableByteCount(skippedFilesSize)); } else { log.debug(msg, indexPath, skippedFiles.size(), humanReadableByteCount(skippedFilesSize)); } if (log.isTraceEnabled()){ log.trace("[COW][{}] File listing - Upon completion {}", indexPath, Arrays.toString(remote.listAll())); } local.close(); remote.close(); closed = true; } CopyOnWriteDirectory(IndexCopier indexCopier, Directory remote, Directory local, boolean reindexMode,
                                String indexPath, Executor executor); @Override String[] listAll(); @Override boolean fileExists(String name); @Override void deleteFile(String name); @Override long fileLength(String name); @Override IndexOutput createOutput(String name, IOContext context); @Override void sync(Collection<String> names); @Override IndexInput openInput(String name, IOContext context); boolean isClosed(); @Override void close(); @Override String toString();  }