DocumentNodeStore implements NodeStore, RevisionContext, Observable, Clusterable, NodeStateDiffer { public void dispose() { LOG.info("Starting disposal of DocumentNodeStore with clusterNodeId: {} ({})", clusterId, getClusterNodeInfoDisplayString()); if (isDisposed.getAndSet(true)) { return; } synchronized (isDisposed) { isDisposed.notifyAll(); } Utils.joinQuietly(backgroundReadThread, backgroundUpdateThread, backgroundSweepThread); DocumentStoreException ex = null; if (!readOnlyMode) { try { Revision tombstone = commitQueue.createRevision(); commitQueue.done(tombstone, new CommitQueue.Callback() { @Override public void headOfQueue(@NotNull Revision revision) { setRoot(getHeadRevision().update(revision)); unsavedLastRevisions.put(ROOT, revision); } }); } catch (DocumentStoreException e) { LOG.error("dispose: a DocumentStoreException happened during dispose's attempt to commit a tombstone: " + e, e); ex = e; } } try { bundlingConfigHandler.close(); } catch (IOException e) { LOG.warn("Error closing bundlingConfigHandler", bundlingConfigHandler, e); } if (!readOnlyMode) { try { internalRunBackgroundSweepOperation(); internalRunBackgroundUpdateOperations(); } catch (DocumentStoreException e) { LOG.error("dispose: a DocumentStoreException happened during dispose's last background ops: " + e, e); ex = e; } } Utils.joinQuietly(clusterUpdateThread); LOG.debug("Stopping LeaseUpdate thread..."); stopLeaseUpdateThread.set(true); synchronized (stopLeaseUpdateThread) { stopLeaseUpdateThread.notifyAll(); } Utils.joinQuietly(leaseUpdateThread); LOG.debug("Stopped LeaseUpdate thread"); if (ex == null) { clusterNodeInfo.dispose(); } store.dispose(); try { blobStore.close(); } catch (Exception e) { LOG.debug("Error closing blob store " + blobStore, e); } if (persistentCache != null) { persistentCache.close(); } if (journalCache != null) { journalCache.close(); } String result = "(successful)"; if (ex != null) { result = "(with exception: " + ex.toString() + ")"; } LOG.info("Disposed DocumentNodeStore with clusterNodeId: {}, {}", clusterId, result); if (ex != null) { throw ex; } } DocumentNodeStore(DocumentNodeStoreBuilder<?> builder); void dispose(); @NotNull DocumentStore getDocumentStore(); void setAsyncDelay(int delay); int getAsyncDelay(); void setMaxBackOffMillis(int time); int getMaxBackOffMillis(); int getChangeSetMaxItems(); void setChangeSetMaxItems(int changeSetMaxItems); int getChangeSetMaxDepth(); void setChangeSetMaxDepth(int changeSetMaxDepth); @NotNull ClusterNodeInfo getClusterInfo(); CacheStats getNodeCacheStats(); CacheStats getNodeChildrenCacheStats(); @NotNull Iterable<CacheStats> getDiffCacheStats(); Cache<PathRev, DocumentNodeState> getNodeCache(); Cache<NamePathRev, DocumentNodeState.Children> getNodeChildrenCache(); Predicate<Path> getNodeCachePredicate(); int getPendingWriteCount(); boolean isDisableBranches(); @NotNull PropertyState createPropertyState(String name, String value); @Nullable DocumentNodeState getNode(@NotNull final Path path,
                                     @NotNull final RevisionVector rev); BundlingConfigHandler getBundlingConfigHandler(); @Override boolean compare(@NotNull final AbstractDocumentNodeState node,
                           @NotNull final AbstractDocumentNodeState base,
                           @NotNull NodeStateDiff diff); @Override Closeable addObserver(Observer observer); @NotNull @Override DocumentNodeState getRoot(); @NotNull @Override NodeState merge(@NotNull NodeBuilder builder,
                           @NotNull CommitHook commitHook,
                           @NotNull CommitInfo info); @NotNull @Override NodeState rebase(@NotNull NodeBuilder builder); @Override NodeState reset(@NotNull NodeBuilder builder); @Override @NotNull BlobStoreBlob createBlob(InputStream inputStream); @Override Blob getBlob(@NotNull String reference); Blob getBlobFromBlobId(String blobId); @NotNull @Override String checkpoint(long lifetime, @NotNull Map<String, String> properties); @NotNull @Override String checkpoint(long lifetime); @NotNull @Override Map<String, String> checkpointInfo(@NotNull String checkpoint); @NotNull @Override Iterable<String> checkpoints(); @Nullable @Override NodeState retrieve(@NotNull String checkpoint); @Override boolean release(@NotNull String checkpoint); @Override UnmergedBranches getBranches(); @Override UnsavedModifications getPendingModifications(); @Override int getClusterId(); @Override @NotNull RevisionVector getHeadRevision(); @Override @NotNull Revision newRevision(); @Override @NotNull Clock getClock(); @Override String getCommitValue(@NotNull Revision changeRevision,
                                 @NotNull NodeDocument doc); void runBackgroundOperations(); @Nullable MarkSweepGarbageCollector createBlobGarbageCollector(long blobGcMaxAgeInSecs, String repositoryId,
        Whiteboard whiteboard, StatisticsProvider statisticsProvider); DocumentNodeStoreMBean getMBean(); BlobStore getBlobStore(); Iterator<ReferencedBlob> getReferencedBlobsIterator(); DiffCache getDiffCache(); Checkpoints getCheckpoints(); @NotNull VersionGarbageCollector getVersionGarbageCollector(); @NotNull JournalGarbageCollector getJournalGarbageCollector(); @NotNull LastRevRecoveryAgent getLastRevRecoveryAgent(); @Override String getInstanceId(); @Override String getVisibilityToken(); @Override boolean isVisible(@NotNull String visibilityToken, long maxWaitMillis); DocumentNodeStoreStatsCollector getStatsCollector(); DocumentNodeStateCache getNodeStateCache(); void setNodeStateCache(DocumentNodeStateCache nodeStateCache); JournalPropertyHandlerFactory getJournalPropertyHandlerFactory(); static final FormatVersion VERSION; static final List<String> META_PROP_NAMES; static final String SYS_PROP_DISABLE_JOURNAL; }
@Test public void dispose() throws CommitFailedException, InterruptedException { final BlockingQueue<String> updates = new ArrayBlockingQueue<String>(1); final AtomicBoolean throttleUpdates = new AtomicBoolean(true); MemoryDocumentStore docStore = new MemoryDocumentStore() { @Override public <T extends Document> T createOrUpdate(Collection<T> collection, UpdateOp update) { if (throttleUpdates.get() && TestUtils.isLastRevUpdate(update)) { try { updates.put(update.getId()); } catch (InterruptedException e) { throw new RuntimeException(e); } } return super.createOrUpdate(collection, update); } }; final DocumentNodeStore store = builderProvider.newBuilder() .setClusterId(1).setAsyncDelay(0) .setDocumentStore(docStore).getNodeStore(); updates.clear(); NodeBuilder builder = store.getRoot().builder(); builder.child("test").child("node"); merge(store, builder); builder = store.getRoot().builder(); builder.child("test").child("node").child("child-1"); merge(store, builder); Thread t = new Thread(new Runnable() { @Override public void run() { store.dispose(); } }); t.start(); String p = updates.take(); assertEquals("2:/test/node", p); assertTrue(t.isAlive()); for (;;) { if (updates.peek() != null) { break; } } try { builder = store.getRoot().builder(); builder.child("test").child("node").child("child-2"); merge(store, builder); fail("Merge must fail with CommitFailedException"); } catch (CommitFailedException e) { } while (t.isAlive()) { updates.poll(10, TimeUnit.MILLISECONDS); } updates.clear(); throttleUpdates.set(false); DocumentNodeStore store2 = builderProvider.newBuilder() .setClusterId(2).setAsyncDelay(0) .setDocumentStore(docStore).getNodeStore(); LastRevRecoveryAgent agent = new LastRevRecoveryAgent(docStore, store2); if (agent.isRecoveryNeeded()) { agent.recover(1); } builder = store2.getRoot().builder(); NodeBuilder test = builder.getChildNode("test"); assertTrue(test.exists()); NodeBuilder node = test.getChildNode("node"); assertTrue(node.exists()); if (!node.hasChildNode("child-2")) { node.child("child-2"); merge(store2, builder); } }
@Test public void preventCommitInPast() throws Exception { Clock clock = new Clock.Virtual(); clock.waitUntil(System.currentTimeMillis()); Revision.setClock(clock); ClusterNodeInfo.setClock(clock); DocumentStore store = new MemoryDocumentStore(); DocumentNodeStore ns = builderProvider.newBuilder() .setLeaseCheckMode(LeaseCheckMode.LENIENT) .setClusterId(1).clock(clock) .setDocumentStore(store).build(); clock.waitUntil(clock.getTime() + TimeUnit.MINUTES.toMillis(2)); doSomeChange(ns); ns.dispose(); long now = clock.getTime(); clock = new Clock.Virtual(); clock.waitUntil(now - TimeUnit.MINUTES.toMillis(1)); Revision.setClock(clock); ClusterNodeInfo.setClock(clock); try { builderProvider.newBuilder() .setClusterId(1).clock(clock) .setDocumentStore(store).build(); fail("must fail with DocumentStoreException"); } catch (DocumentStoreException e) { assertThat(e.getMessage(), containsString("newer than current time")); } ClusterNodeInfoDocument info = store.find(CLUSTER_NODES, "1"); assertNotNull(info); assertFalse(info.isActive()); }