ActiveDeletedBlobCollectorMBeanImpl implements ActiveDeletedBlobCollectorMBean { @Override public void flagActiveDeletionUnsafeForCurrentState() { activeDeletedBlobCollector.flagActiveDeletionUnsafe(true); if (!waitForRunningIndexCycles()) { LOG.warn("Some indexers were still found running. Resume and quit gracefully"); activeDeletedBlobCollector.flagActiveDeletionUnsafe(false); } try { markCurrentIndexFilesUnsafeForActiveDeletion(); } catch (CommitFailedException e) { LOG.warn("Could not set current index files unsafe for active deletion. Resume and quit gracefully", e); activeDeletedBlobCollector.flagActiveDeletionUnsafe(false); } } ActiveDeletedBlobCollectorMBeanImpl(
            @NotNull ActiveDeletedBlobCollector activeDeletedBlobCollector,
            @NotNull Whiteboard whiteboard,
            @NotNull NodeStore store,
            @NotNull IndexPathService indexPathService,
            @NotNull AsyncIndexInfoService asyncIndexInfoService,
            @NotNull GarbageCollectableBlobStore blobStore,
            @NotNull Executor executor); @Override boolean isDisabled(); @NotNull @Override CompositeData startActiveCollection(); @NotNull @Override CompositeData cancelActiveCollection(); @NotNull @Override CompositeData getActiveCollectionStatus(); @Override boolean isActiveDeletionUnsafe(); @Override void flagActiveDeletionUnsafeForCurrentState(); @Override void flagActiveDeletionSafe();  }
@Test public void onlyRunningIndexesRequireToBeWaitedOn() { IndexPathService indexPathService = MockRegistrar.getIndexPathsService(indexPaths); final StatusSupplier statusSupplier = new StatusSupplier(); final AtomicLong returnExecCount = new AtomicLong(2L); AsyncIndexInfoService asyncIndexInfoService = MockRegistrar.getAsyncIndexInfoService(newArrayList( new IndexMBeanInfoSupplier("foo-async", statusSupplier, returnExecCount::get) )); ActiveDeletedBlobCollectorMBean bean = getTestBean(indexPathService, asyncIndexInfoService); long start = clock.getTime(); bean.flagActiveDeletionUnsafeForCurrentState(); long elapsed = clock.getTime() - start; assertTrue("Non running index lane was polled for " + TimeUnit.MILLISECONDS.toSeconds(elapsed) + " seconds.", elapsed < TimeUnit.SECONDS.toMillis(5)); statusSupplier.status = STATUS_RUNNING; start = clock.getTime(); bean.flagActiveDeletionUnsafeForCurrentState(); elapsed = clock.getTime() - start; assertTrue("Running index lane without changing execCnt was polled for " + TimeUnit.MILLISECONDS.toSeconds(elapsed) + " seconds.", elapsed > TimeUnit.SECONDS.toMillis(120) && elapsed < TimeUnit.SECONDS.toMillis(125)); statusSupplier.status = STATUS_RUNNING; asyncIndexInfoService = MockRegistrar.getAsyncIndexInfoService(newArrayList( new IndexMBeanInfoSupplier("foo-async", statusSupplier, returnExecCount::incrementAndGet) )); bean = getTestBean(indexPathService, asyncIndexInfoService); start = clock.getTime(); bean.flagActiveDeletionUnsafeForCurrentState(); elapsed = clock.getTime() - start; assertTrue("Running index lane without changing execCnt was polled for " + TimeUnit.MILLISECONDS.toSeconds(elapsed) + " seconds.", elapsed < TimeUnit.SECONDS.toMillis(5)); }
@Test public void headIndexFilesGetMarkedUnsafe() throws Exception { String indexPath = "/fooIndex"; createFakeIndex(indexPath); IndexPathService indexPathService = MockRegistrar.getIndexPathsService(indexPaths); AsyncIndexInfoService asyncIndexInfoService = MockRegistrar.getAsyncIndexInfoService(newArrayList( new IndexMBeanInfoSupplier("foo-async", () -> STATUS_DONE, () -> 2L) )); ActiveDeletedBlobCollectorMBean bean = getTestBean(indexPathService, asyncIndexInfoService); bean.flagActiveDeletionUnsafeForCurrentState(); NodeState indexFile = getFakeIndexFile(indexPath); assertTrue(indexFile.getBoolean(PROP_UNSAFE_FOR_ACTIVE_DELETION)); }
@Test public void orderOfFlaggingWaitForIndexersAndUpdateIndexFiles() { final AtomicBoolean isPaused = new AtomicBoolean(); final AtomicBoolean hadWaitedForIndex = new AtomicBoolean(); IndexPathService indexPathService = mock(IndexPathService.class); when(indexPathService.getIndexPaths()).then(mockObj -> { assertTrue("Must wait for indexers before going to update index files", hadWaitedForIndex.get()); return indexPaths; }); AsyncIndexInfoService asyncIndexInfoService = MockRegistrar.getAsyncIndexInfoService(newArrayList( new IndexMBeanInfoSupplier("foo-async", () -> { assertTrue("Must pause before waiting for indexers", isPaused.get()); hadWaitedForIndex.set(true); return STATUS_DONE; }, () -> 2L) )); ActiveDeletedBlobCollectorMBeanImpl bean = new ActiveDeletedBlobCollectorMBeanImpl(new PauseNotifyingActiveDeletedBlobCollector(() -> { isPaused.set(true); return null; }), wb, nodeStore, indexPathService, asyncIndexInfoService, new MemoryBlobStore(), sameThreadExecutor()); bean.clock = clock; bean.flagActiveDeletionUnsafeForCurrentState(); }
@Test public void clonedNSWithSharedDS() throws Exception { MemoryBlobStore bs = new MemoryBlobStore(); bs.setBlockSizeMin(48); MemoryDocumentStore mds1 = new MemoryDocumentStore(); DocumentNodeStore dns1 = builderProvider.newBuilder() .setDocumentStore(mds1).setBlobStore(bs).build(); LuceneIndexEditorProvider editorProvider = new LuceneIndexEditorProvider(); ContentRepository repository = new Oak(dns1) .with(new InitialContent()) .with(new OpenSecurityProvider()) .with(editorProvider) .with(new PropertyIndexEditorProvider()) .with(new NodeTypeIndexProvider()) .createContentRepository(); ContentSession session = repository.login(null, null); Root root = session.getLatestRoot(); TestUtil.createFulltextIndex(root.getTree("/"), "testIndex"); root.commit(); IndexPathService indexPathService = new IndexPathServiceImpl(dns1); AsyncIndexInfoService asyncIndexInfoService = MockRegistrar.getAsyncIndexInfoService(newArrayList( new IndexMBeanInfoSupplier("foo-async", () -> STATUS_DONE, () -> 2L) )); ActiveDeletedBlobCollectorMBeanImpl bean = new ActiveDeletedBlobCollectorMBeanImpl(ActiveDeletedBlobCollectorFactory.NOOP, wb, dns1, indexPathService, asyncIndexInfoService, new MemoryBlobStore(), sameThreadExecutor()); bean.clock = clock; bean.flagActiveDeletionUnsafeForCurrentState(); root = session.getLatestRoot(); assertTrue("First pass indexing should generate segments_1", root.getTree("/oak:index/testIndex/:data/segments_1").exists()); dns1.dispose(); MemoryDocumentStore mds2 = mds1.copy(); DocumentNodeStore dns2 = builderProvider.newBuilder().setDocumentStore(mds2).setBlobStore(bs).build(); DeletedFileTrackingADBC deletedFileTrackingADBC = new DeletedFileTrackingADBC( new File(temporaryFolder.getRoot(), "adbc-workdir")); editorProvider = new LuceneIndexEditorProvider(null, null, new ExtractedTextCache(0, 0), null, Mounts.defaultMountInfoProvider(), deletedFileTrackingADBC, null, null); repository = new Oak(dns2) .with(new OpenSecurityProvider()) .with(editorProvider) .with(new PropertyIndexEditorProvider()) .with(new NodeTypeIndexProvider()) .createContentRepository(); session = repository.login(null, null); root = session.getLatestRoot(); Tree rootTree = root.getTree("/"); for (int i = 0; i < 20; i++) { Tree child = rootTree.addChild("a" + i); for (int j = 0; j < 20; j++) { child.setProperty("foo" + j, "bar" + j); } } root.commit(); deletedFileTrackingADBC.blobDeletionCallback.commitProgress(COMMIT_SUCCEDED); deletedFileTrackingADBC.purgeBlobsDeleted(Clock.SIMPLE.getTime() + TimeUnit.SECONDS.toMillis(1), bs); root = session.getLatestRoot(); assertFalse("Churn created via dns2 should delete segments_1", root.getTree("/oak:index/testIndex/:data/segments_1").exists()); dns2.dispose(); dns1 = builderProvider.newBuilder().setDocumentStore(mds1).setBlobStore(bs).build(); IndexConsistencyChecker checker = new IndexConsistencyChecker(dns1.getRoot(), "/oak:index/testIndex", new File(temporaryFolder.getRoot(), "checker-workdir")); IndexConsistencyChecker.Result result = checker.check(IndexConsistencyChecker.Level.BLOBS_ONLY); assertFalse("Nodestore1 can't read blobs: " + result.missingBlobIds + " while reading index", result.missingBlobs); }