AsyncIndexUpdate implements Runnable, Closeable { @Override public synchronized void run() { boolean permitAcquired = false; try{ if (runPermit.tryAcquire()){ permitAcquired = true; runWhenPermitted(); } else { log.warn("[{}] Could not acquire run permit. Stop flag set to [{}] Skipping the run", name, forcedStopFlag); } } finally { if (permitAcquired){ runPermit.release(); } } } AsyncIndexUpdate(@NotNull String name, @NotNull NodeStore store,
                            @NotNull IndexEditorProvider provider, boolean switchOnSync);  AsyncIndexUpdate(@NotNull String name, @NotNull NodeStore store,
                            @NotNull IndexEditorProvider provider, StatisticsProvider statsProvider, boolean switchOnSync);  AsyncIndexUpdate(@NotNull String name, @NotNull NodeStore store,
                            @NotNull IndexEditorProvider provider); static String checkValidName(String asyncName); static boolean isAsyncLaneName(String asyncName); @Override synchronized void run(); @Override void close(); static String leasify(String name); void setValidatorProviders(List<ValidatorProvider> validatorProviders); void setCorruptIndexHandler(TrackingCorruptIndexHandler corruptIndexHandler); boolean isClosed(); AsyncIndexStats getIndexStats(); boolean isFinished(); boolean isFailing(); void setIndexMBeanRegistration(IndexMBeanRegistration mbeanRegistration); String getName(); static final String PROP_ASYNC_NAME; }
@Test public void testAsync() throws Exception { NodeStore store = new MemoryNodeStore(); IndexEditorProvider provider = new PropertyIndexEditorProvider(); NodeBuilder builder = store.getRoot().builder(); createIndexDefinition(builder.child(INDEX_DEFINITIONS_NAME), "rootIndex", true, false, ImmutableSet.of("foo"), null) .setProperty(ASYNC_PROPERTY_NAME, "async"); builder.child("testRoot").setProperty("foo", "abc"); store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); AsyncIndexUpdate async = new AsyncIndexUpdate("async", store, provider); async.run(); NodeState root = store.getRoot(); checkPathExists(root, INDEX_DEFINITIONS_NAME, "rootIndex", INDEX_CONTENT_NODE_NAME); assertFalse(root.getChildNode(INDEX_DEFINITIONS_NAME).hasChildNode( ":conflict")); PropertyIndexLookup lookup = new PropertyIndexLookup(root); assertEquals(ImmutableSet.of("testRoot"), find(lookup, "foo", "abc")); }
@Test public void testAsyncDouble() throws Exception { NodeStore store = new MemoryNodeStore(); IndexEditorProvider provider = new PropertyIndexEditorProvider(); NodeBuilder builder = store.getRoot().builder(); createIndexDefinition(builder.child(INDEX_DEFINITIONS_NAME), "rootIndex", true, false, ImmutableSet.of("foo"), null) .setProperty(ASYNC_PROPERTY_NAME, "async"); createIndexDefinition(builder.child(INDEX_DEFINITIONS_NAME), "rootIndexSecond", true, false, ImmutableSet.of("bar"), null) .setProperty(ASYNC_PROPERTY_NAME, "async"); builder.child("testRoot").setProperty("foo", "abc") .setProperty("bar", "def"); builder.child("testSecond").setProperty("bar", "ghi"); store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); AsyncIndexUpdate async = new AsyncIndexUpdate("async", store, provider); async.run(); NodeState root = store.getRoot(); checkPathExists(root, INDEX_DEFINITIONS_NAME, "rootIndex", INDEX_CONTENT_NODE_NAME); checkPathExists(root, INDEX_DEFINITIONS_NAME, "rootIndexSecond", INDEX_CONTENT_NODE_NAME); PropertyIndexLookup lookup = new PropertyIndexLookup(root); assertEquals(ImmutableSet.of("testRoot"), find(lookup, "foo", "abc")); assertEquals(ImmutableSet.<String> of(), find(lookup, "foo", "def")); assertEquals(ImmutableSet.<String> of(), find(lookup, "foo", "ghi")); assertEquals(ImmutableSet.<String> of(), find(lookup, "bar", "abc")); assertEquals(ImmutableSet.of("testRoot"), find(lookup, "bar", "def")); assertEquals(ImmutableSet.of("testSecond"), find(lookup, "bar", "ghi")); }
@Test public void testAsyncDoubleSubtree() throws Exception { NodeStore store = new MemoryNodeStore(); IndexEditorProvider provider = new PropertyIndexEditorProvider(); NodeBuilder builder = store.getRoot().builder(); createIndexDefinition(builder.child(INDEX_DEFINITIONS_NAME), "rootIndex", true, false, ImmutableSet.of("foo"), null) .setProperty(ASYNC_PROPERTY_NAME, "async"); createIndexDefinition( builder.child("newchild").child("other") .child(INDEX_DEFINITIONS_NAME), "subIndex", true, false, ImmutableSet.of("foo"), null).setProperty( ASYNC_PROPERTY_NAME, "async"); builder.child("testRoot").setProperty("foo", "abc"); builder.child("newchild").child("other").child("testChild") .setProperty("foo", "xyz"); store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); AsyncIndexUpdate async = new AsyncIndexUpdate("async", store, provider); async.run(); NodeState root = store.getRoot(); checkPathExists(root, INDEX_DEFINITIONS_NAME, "rootIndex", INDEX_CONTENT_NODE_NAME); checkPathExists(root, "newchild", "other", INDEX_DEFINITIONS_NAME, "subIndex", INDEX_CONTENT_NODE_NAME); PropertyIndexLookup lookup = new PropertyIndexLookup(root); assertEquals(ImmutableSet.of("testRoot"), find(lookup, "foo", "abc")); PropertyIndexLookup lookupChild = new PropertyIndexLookup(root .getChildNode("newchild").getChildNode("other")); assertEquals(ImmutableSet.of("testChild"), find(lookupChild, "foo", "xyz")); assertEquals(ImmutableSet.<String> of(), find(lookupChild, "foo", "abc")); }
@Test public void branchBaseOnCheckpoint() throws Exception { final Semaphore retrieve = new Semaphore(1); final Semaphore checkpoint = new Semaphore(0); NodeStore store = new MemoryNodeStore() { @Nullable @Override public NodeState retrieve(@NotNull String checkpoint) { retrieve.acquireUninterruptibly(); try { return super.retrieve(checkpoint); } finally { retrieve.release(); } } @NotNull @Override public String checkpoint(long lifetime, @NotNull Map<String, String> properties) { try { return super.checkpoint(lifetime, properties); } finally { checkpoint.release(); } } }; IndexEditorProvider provider = new PropertyIndexEditorProvider(); NodeBuilder builder = store.getRoot().builder(); createIndexDefinition(builder.child(INDEX_DEFINITIONS_NAME), "foo", false, ImmutableSet.of("foo"), null, TYPE, Collections.singletonMap(ASYNC_PROPERTY_NAME, "async")); builder.child("test").setProperty("foo", "a"); builder.child("child"); store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); final AsyncIndexUpdate async = new AsyncIndexUpdate("async", store, provider); async.run(); builder = store.getRoot().builder(); builder.child("test").setProperty("foo", "b"); builder.child("child").setProperty("prop", "value"); store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); Thread t = new Thread(new Runnable() { @Override public void run() { async.run(); } }); checkpoint.acquireUninterruptibly(checkpoint.availablePermits()); retrieve.acquireUninterruptibly(); t.start(); retrieve.release(); checkpoint.acquireUninterruptibly(); builder = store.getRoot().builder(); builder.child("child").remove(); store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); retrieve.release(); t.join(); assertFalse(store.getRoot().hasChildNode("child")); }
@Test public void failOnConflict() throws Exception { final Map<Thread, Semaphore> locks = Maps.newIdentityHashMap(); NodeStore store = new MemoryNodeStore() { @NotNull @Override public NodeState merge(@NotNull NodeBuilder builder, @NotNull CommitHook commitHook, @NotNull CommitInfo info) throws CommitFailedException { Semaphore s = locks.get(Thread.currentThread()); if (s != null) { s.acquireUninterruptibly(); } return super.merge(builder, commitHook, info); } }; IndexEditorProvider provider = new PropertyIndexEditorProvider(); NodeBuilder builder = store.getRoot().builder(); createIndexDefinition(builder.child(INDEX_DEFINITIONS_NAME), "foo", false, ImmutableSet.of("foo"), null, TYPE, Collections.singletonMap(ASYNC_PROPERTY_NAME, "async")); builder.child("test").setProperty("foo", "a"); store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); final AsyncIndexUpdate async = new AsyncIndexUpdate("async", store, provider); async.run(); builder = store.getRoot().builder(); builder.child("test").setProperty("foo", "b"); store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); Thread t = new Thread(new Runnable() { @Override public void run() { async.run(); } }); Semaphore s = new Semaphore(0); locks.put(t, s); t.start(); builder = store.getRoot().builder(); builder.setChildNode("dummy").setProperty("foo", "bar"); store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); while (!s.hasQueuedThreads()) { Thread.yield(); } builder = store.getRoot().builder(); builder.getChildNode(INDEX_DEFINITIONS_NAME).getChildNode("foo") .getChildNode(":index").child("a").remove(); store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); s.release(100); t.join(); builder = store.getRoot().builder(); assertNoConflictMarker(builder); }
@Test public void recoverFromMissingCpRef() throws Exception { MemoryNodeStore store = new MemoryNodeStore(); IndexEditorProvider provider = new PropertyIndexEditorProvider(); NodeBuilder builder = store.getRoot().builder(); createIndexDefinition(builder.child(INDEX_DEFINITIONS_NAME), "rootIndex", true, false, ImmutableSet.of("foo"), null) .setProperty(ASYNC_PROPERTY_NAME, "async"); builder.child("testRoot").setProperty("foo", "abc"); store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); new AsyncIndexUpdate("async", store, provider).run(); checkPathExists(store.getRoot(), INDEX_DEFINITIONS_NAME, "rootIndex", INDEX_CONTENT_NODE_NAME, "abc", "testRoot"); builder = store.getRoot().builder(); builder.child(ASYNC).setProperty("async", "faulty"); builder.child("testAnother").setProperty("foo", "def"); store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); new AsyncIndexUpdate("async", store, provider).run(); checkPathExists(store.getRoot(), INDEX_DEFINITIONS_NAME, "rootIndex", INDEX_CONTENT_NODE_NAME, "def", "testAnother"); }
@Test public void cpCleanupWChanges() throws Exception { MemoryNodeStore store = new MemoryNodeStore(); IndexEditorProvider provider = new PropertyIndexEditorProvider(); NodeBuilder builder = store.getRoot().builder(); createIndexDefinition(builder.child(INDEX_DEFINITIONS_NAME), "rootIndex", true, false, ImmutableSet.of("foo"), null) .setProperty(ASYNC_PROPERTY_NAME, "async"); builder.child("testRoot").setProperty("foo", "abc"); store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); assertTrue("Expecting no checkpoints", store.listCheckpoints().size() == 0); AsyncIndexUpdate async = new AsyncIndexUpdate("async", store, provider); async.run(); assertTrue("Expecting one checkpoint", store.listCheckpoints().size() == 1); String firstCp = store.listCheckpoints().iterator().next(); builder = store.getRoot().builder(); builder.child("testRoot").setProperty("foo", "def"); store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); async.run(); assertTrue("Expecting one checkpoint", store.listCheckpoints().size() == 1); String secondCp = store.listCheckpoints().iterator().next(); assertFalse("Store should keep only second checkpoint", secondCp.equals(firstCp)); assertEquals( secondCp, store.getRoot().getChildNode(ASYNC) .getString("async")); }
@Test public void cpCleanupWUnrelatedChanges() throws Exception { MemoryNodeStore store = new MemoryNodeStore(); IndexEditorProvider provider = new PropertyIndexEditorProvider(); NodeBuilder builder = store.getRoot().builder(); createIndexDefinition(builder.child(INDEX_DEFINITIONS_NAME), "rootIndex", true, false, ImmutableSet.of("foo"), null) .setProperty(ASYNC_PROPERTY_NAME, "async"); builder.child("testRoot").setProperty("foo", "abc"); store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); assertTrue("Expecting no checkpoints", store.listCheckpoints().size() == 0); AsyncIndexUpdate async = new AsyncIndexUpdate("async", store, provider); async.run(); assertTrue("Expecting one checkpoint", store.listCheckpoints().size() == 1); String firstCp = store.listCheckpoints().iterator().next(); builder = store.getRoot().builder(); builder.child("testRoot").child(":hidden"); store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); async.run(); assertTrue("Expecting one checkpoint", store.listCheckpoints().size() == 1); String secondCp = store.listCheckpoints().iterator().next(); assertFalse("Store should keep only second checkpoint", secondCp.equals(firstCp)); assertEquals( secondCp, store.getRoot().getChildNode(ASYNC) .getString("async")); }
@Test public void commitContext() throws Exception{ MemoryNodeStore store = new MemoryNodeStore(); IndexEditorProvider provider = new PropertyIndexEditorProvider(); NodeBuilder builder = store.getRoot().builder(); createIndexDefinition(builder.child(INDEX_DEFINITIONS_NAME), "rootIndex", true, false, ImmutableSet.of("foo"), null) .setProperty(ASYNC_PROPERTY_NAME, "async"); builder.child("testRoot").setProperty("foo", "abc"); store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); AsyncIndexUpdate async = new AsyncIndexUpdate("async", store, provider); CommitInfoCollector infoCollector = new CommitInfoCollector(); store.addObserver(infoCollector); async.run(); assertFalse(infoCollector.infos.isEmpty()); assertNotNull(infoCollector.infos.get(0).getInfo().get(CommitContext.NAME)); }
@Test public void startTimePresentInCommitInfo() throws Exception{ MemoryNodeStore store = new MemoryNodeStore(); NodeBuilder builder = store.getRoot().builder(); createIndexDefinition(builder.child(INDEX_DEFINITIONS_NAME), "fooIndex", true, false, ImmutableSet.of("foo"), null) .setProperty(ASYNC_PROPERTY_NAME, "async"); builder.child("testRoot1").setProperty("foo", "abc"); store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); IndexingContextCapturingProvider provider = new IndexingContextCapturingProvider(); AsyncIndexUpdate async = new AsyncIndexUpdate("async", store, provider); async.run(); assertNotNull(provider.lastIndexingContext); CommitInfo info = provider.lastIndexingContext.getCommitInfo(); String indexStartTime = (String) info.getInfo().get(IndexConstants.CHECKPOINT_CREATION_TIME); assertNotNull(indexStartTime); }
@Test public void disableSupersededIndex() throws Exception { IndexEditorProvider propIdxEditorProvider = new PropertyIndexEditorProvider(); EditorHook propIdxHook = new EditorHook(new IndexUpdateProvider(propIdxEditorProvider)); MemoryNodeStore store = new MemoryNodeStore(); String supersededIndexName = "supersededIndex"; String supersedingIndexName = "supersedingIndex"; AsyncIndexUpdate async = new AsyncIndexUpdate("async", store, propIdxEditorProvider); NodeBuilder builder = store.getRoot().builder(); NodeBuilder oakIndex = builder.child(INDEX_DEFINITIONS_NAME); createIndexDefinition(oakIndex, supersededIndexName, true, false, ImmutableSet.of("foo"), null); store.merge(builder, propIdxHook, CommitInfo.EMPTY); builder = store.getRoot().builder(); oakIndex = builder.child(INDEX_DEFINITIONS_NAME); createIndexDefinition(oakIndex, supersedingIndexName, true, false, ImmutableSet.of("foo"), null) .setProperty(ASYNC_PROPERTY_NAME, ImmutableSet.of("async", "nrt"), Type.STRINGS) .setProperty(SUPERSEDED_INDEX_PATHS, INDEX_DEFINITIONS_NAME + "/" + supersededIndexName) ; store.merge(builder, propIdxHook, CommitInfo.EMPTY); builder = store.getRoot().builder(); builder.child("testNode1").setProperty("foo", "bar"); store.merge(builder, propIdxHook, CommitInfo.EMPTY); NodeState supersededIndex = store.getRoot().getChildNode(INDEX_DEFINITIONS_NAME).getChildNode(supersededIndexName); assertEquals("Index disabled too early", "property", supersededIndex.getString("type")); assertFalse("Don't set :disableIndexesOnNextCycle on superseded index", supersededIndex.hasProperty(DISABLE_INDEXES_ON_NEXT_CYCLE)); NodeState supersedingIndex = store.getRoot().getChildNode(INDEX_DEFINITIONS_NAME).getChildNode(supersedingIndexName); assertFalse("Don't set :disableIndexesOnNextCycle on superseding index just yet", supersedingIndex.hasProperty(DISABLE_INDEXES_ON_NEXT_CYCLE)); async.run(); supersededIndex = store.getRoot().getChildNode(INDEX_DEFINITIONS_NAME).getChildNode(supersededIndexName); assertEquals("Index disabled too early", "property", supersededIndex.getString("type")); assertFalse("Don't set :disableIndexesOnNextCycle on superseded index", supersededIndex.hasProperty(DISABLE_INDEXES_ON_NEXT_CYCLE)); supersedingIndex = store.getRoot().getChildNode(INDEX_DEFINITIONS_NAME).getChildNode(supersedingIndexName); assertTrue(":disableIndexesOnNextCycle not set on superseding index after reindexing run", supersedingIndex.hasProperty(DISABLE_INDEXES_ON_NEXT_CYCLE)); builder = store.getRoot().builder(); store.getRoot().builder().child("testNode2").setProperty("foo", "bar"); store.merge(builder, propIdxHook, CommitInfo.EMPTY); supersededIndex = store.getRoot().getChildNode(INDEX_DEFINITIONS_NAME).getChildNode(supersededIndexName); assertEquals("Index disabled too early", "property", supersededIndex.getString("type")); assertFalse("Don't set :disableIndexesOnNextCycle on superseded index", supersededIndex.hasProperty(DISABLE_INDEXES_ON_NEXT_CYCLE)); supersedingIndex = store.getRoot().getChildNode(INDEX_DEFINITIONS_NAME).getChildNode(supersedingIndexName); assertTrue(":disableIndexesOnNextCycle not set on superseding index after reindexing run", supersedingIndex.getBoolean(DISABLE_INDEXES_ON_NEXT_CYCLE)); async.run(); supersededIndex = store.getRoot().getChildNode(INDEX_DEFINITIONS_NAME).getChildNode(supersededIndexName); assertEquals("Index yet not disabled", "disabled", supersededIndex.getString("type")); assertFalse("Don't set :disableIndexesOnNextCycle on superseded index", supersededIndex.hasProperty(DISABLE_INDEXES_ON_NEXT_CYCLE)); supersedingIndex = store.getRoot().getChildNode(INDEX_DEFINITIONS_NAME).getChildNode(supersedingIndexName); assertFalse("Don't keep :disableIndexesOnNextCycle on superseding index after disabling", supersedingIndex.hasProperty(DISABLE_INDEXES_ON_NEXT_CYCLE)); }
@Test public void indexCommitCallback() throws Exception { AtomicBoolean gotFailedCommit = new AtomicBoolean(); AtomicBoolean gotSuccessfulCommit = new AtomicBoolean(); AtomicBoolean shouldFail = new AtomicBoolean(); MemoryNodeStore store = new MemoryNodeStore(); AsyncIndexUpdate async = new AsyncIndexUpdate("async", store, (type, definition, root, callback) -> { IndexingContext indexingContext = ((ContextAwareCallback)callback).getIndexingContext(); indexingContext.registerIndexCommitCallback(indexProgress -> { switch (indexProgress) { case COMMIT_FAILED: gotFailedCommit.set(true); break; case COMMIT_SUCCEDED: gotSuccessfulCommit.set(true); break; } }); if (shouldFail.get()) { throw new CommitFailedException("indexer-fail", 1, "Explicitly failing while indexing"); } return new DefaultEditor(); }); NodeBuilder builder = store.getRoot().builder(); builder.child("oak:index").child("fooIndex").setProperty("async", "async").setProperty("type", "foo"); store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); gotFailedCommit.set(false); gotSuccessfulCommit.set(false); builder = store.getRoot().builder(); builder.setProperty("foo", "bar"); store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); async.run(); assertTrue("Successful indexing commit must report success", gotSuccessfulCommit.get()); assertFalse("Successful indexing commit must not report failure", gotFailedCommit.get()); gotFailedCommit.set(false); gotSuccessfulCommit.set(false); shouldFail.set(true); builder = store.getRoot().builder(); builder.setProperty("foo", "bar1"); store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); async.run(); assertFalse("Failing indexing must not report success", gotSuccessfulCommit.get()); assertTrue("Failing indexing must report failure", gotFailedCommit.get()); }