trunk/oak-jcr/src/test/java/org/apache/jackrabbit/oak/jcr/ConcurrentAddNodesClusterIT.java - jackrabbit-oak - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
  * the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package org.apache.jackrabbit.oak.jcr;

 import static java.lang.Thread.UncaughtExceptionHandler;
 import static org.apache.jackrabbit.oak.jcr.AbstractRepositoryTest.dispose;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;

 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;

 import javax.jcr.Node;
 import javax.jcr.PropertyType;
 import javax.jcr.Repository;
 import javax.jcr.RepositoryException;
 import javax.jcr.Session;
 import javax.jcr.SimpleCredentials;

 import com.google.common.util.concurrent.Uninterruptibles;
 import org.apache.jackrabbit.oak.plugins.document.DocumentMK;
 import org.apache.jackrabbit.oak.plugins.document.util.MongoConnection;
 import org.apache.jackrabbit.oak.plugins.index.IndexConstants;
 import org.apache.jackrabbit.oak.plugins.index.property.PropertyIndexEditorProvider;
 import org.junit.After;
 import org.junit.Assume;
 import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Ignore;
 import org.junit.Test;

 /**
  * Concurrently add nodes with multiple sessions on multiple cluster nodes.
  */
 public class ConcurrentAddNodesClusterIT {

     private static final int NUM_CLUSTER_NODES = 5;
     private static final int NODE_COUNT = 100;
     private static final int LOOP_COUNT = 10;
     private static final int WORKER_COUNT = 20;
     private static final String PROP_NAME = "testcount";

     private List<Repository> repos = new ArrayList<Repository>();
     private List<DocumentMK> mks = new ArrayList<DocumentMK>();
     private List<Thread> workers = new ArrayList<Thread>();

     @BeforeClass
     public static void mongoDBAvailable() {
         Assume.assumeTrue(OakMongoNSRepositoryStub.isMongoDBAvailable());
     }

     @Before
     public void before() throws Exception {
         dropDB();
         initRepository();
     }

     @After
     public void after() throws Exception {
         workers.clear();
         for (Repository repo : repos) {
             dispose(repo);
         }
         repos.clear();
         for (DocumentMK mk : mks) {
             mk.dispose();
         }
         mks.clear();
         dropDB();
     }

     @Test
     public void addNodesConcurrent() throws Exception {
         for (int i = 0; i < NUM_CLUSTER_NODES; i++) {
             MongoConnection c = createConnection();
             DocumentMK mk = new DocumentMK.Builder()
                     .setMongoDB(c.getMongoClient(), c.getDBName())
                     .setClusterId(i + 1).open();
             mks.add(mk);
         }
         Map<String, Exception> exceptions = Collections.synchronizedMap(
                 new HashMap<String, Exception>());
         for (int i = 0; i < mks.size(); i++) {
             DocumentMK mk = mks.get(i);
             Repository repo = new Jcr(mk.getNodeStore()).createRepository();
             repos.add(repo);
             workers.add(new Thread(new Worker(repo, exceptions), "Worker-" + (i + 1)));
         }
         for (Thread t : workers) {
             t.start();
         }
         for (Thread t : workers) {
             t.join();
         }
         for (Map.Entry<String, Exception> entry : exceptions.entrySet()) {
             // System.out.println("exception in thread " + entry.getKey());
             throw entry.getValue();
         }
     }

     @Ignore("OAK-1807")
     @Test
     public void addNodesConcurrent2() throws Exception {
         final Thread mainThread = Thread.currentThread();
         for (int i = 0; i < NUM_CLUSTER_NODES; i++) {
             MongoConnection c = createConnection();
             DocumentMK mk = new DocumentMK.Builder()
                     .setMongoDB(c.getMongoClient(), c.getDBName())
                     .setClusterId(i + 1).open();
             mks.add(mk);
         }
         final Map<String, Exception> exceptions = Collections.synchronizedMap(
                 new HashMap<String, Exception>());
         final CountDownLatch latch = new CountDownLatch(1);
         final AtomicBoolean stop = new AtomicBoolean();
         final UncaughtExceptionHandler ueh = new UncaughtExceptionHandler() {
             @Override
             public void uncaughtException(Thread t, Throwable e) {
                 RuntimeException r = new RuntimeException("Exception in thread "+t.getName(), e);
                 r.printStackTrace();
             }
         };
         for (int i = 0; i < mks.size(); i++) {
             DocumentMK mk = mks.get(i);
             final Repository repo = new Jcr(mk.getNodeStore()).createRepository();
             repos.add(repo);
             for (int w = 0; w <= WORKER_COUNT; w++) {
                 final String name = "Worker-" + (i + 1) + "-" + (w + 1);
                 final Runnable r = new Runnable() {
                     final Session session = createAdminSession(repo);
                     int count = 0;
                     @Override
                     public void run() {
                         try {
                             Uninterruptibles.awaitUninterruptibly(latch);
                             session.refresh(false);
                             Node node = session.getRootNode().addNode(name+count++, "oak:Unstructured");
                             for (int j = 0; j < NODE_COUNT && !stop.get() ; j++) {
                                 node.addNode("node" + j);
                                 session.save();
                             }
                         } catch (RepositoryException e) {
                             RuntimeException r = new RuntimeException("Exception in thread "+name, e);
                             r.printStackTrace();
                             exceptions.put(Thread.currentThread().getName(), r);
                             stop.set(true);
                             mainThread.interrupt();
                         } finally {
                             session.logout();
                         }
                     }
                 };

                 //Last runnable would be a long running one
                 Runnable runnable = r;
                 if(w == WORKER_COUNT){
                     runnable = new Runnable() {
                         @Override
                         public void run() {
                             while(!stop.get()){
                                 r.run();
                             }
                         }
                     };
                 }

                 Thread t = new Thread(runnable);
                 t.setName(name);
                 t.setUncaughtExceptionHandler(ueh);
                 workers.add(t);
             }
         }
         for (Thread t : workers) {
             t.start();
         }
         latch.countDown();

         TimeUnit.MINUTES.sleep(10);
         stop.set(true);

         for (Thread t : workers) {
             t.join();
         }
         for (Map.Entry<String, Exception> entry : exceptions.entrySet()) {
             // System.out.println("exception in thread " + entry.getKey());
             throw entry.getValue();
         }
     }

     @Test
     public void addNodes() throws Exception {
         for (int i = 0; i < 2; i++) {
             MongoConnection c = createConnection();
             DocumentMK mk = new DocumentMK.Builder()
                     .setMongoDB(c.getMongoClient(), c.getDBName())
                     .setClusterId(i + 1).open();
             mks.add(mk);
         }
         final DocumentMK mk1 = mks.get(0);
         final DocumentMK mk2 = mks.get(1);
         Repository r1 = new Jcr(mk1.getNodeStore()).createRepository();
         repos.add(r1);
         Repository r2 = new Jcr(mk2.getNodeStore()).createRepository();
         repos.add(r2);

         Session s1 = r1.login(new SimpleCredentials("admin", "admin".toCharArray()));
         Session s2 = r2.login(new SimpleCredentials("admin", "admin".toCharArray()));

         ensureIndex(s1.getRootNode(), PROP_NAME);
         ensureIndex(s2.getRootNode(), PROP_NAME);

         Map<String, Exception> exceptions = Collections.synchronizedMap(
                 new HashMap<String, Exception>());
         createNodes(s1, "testroot-1", 1, 1, exceptions);
         createNodes(s2, "testroot-2", 1, 1, exceptions);

         for (Map.Entry<String, Exception> entry : exceptions.entrySet()) {
             throw entry.getValue();
         }

         s1.logout();
         s2.logout();
     }

     @Test
     public void addNodes2() throws Exception {
         for (int i = 0; i < 3; i++) {
             MongoConnection c = createConnection();
             DocumentMK mk = new DocumentMK.Builder()
                     .setMongoDB(c.getMongoClient(), c.getDBName())
                     .setAsyncDelay(0)
                     .setClusterId(i + 1).open();
             mks.add(mk);
         }
         final DocumentMK mk1 = mks.get(0);
         final DocumentMK mk2 = mks.get(1);
         final DocumentMK mk3 = mks.get(2);
         Repository r1 = new Jcr(mk1.getNodeStore()).createRepository();
         repos.add(r1);
         Repository r2 = new Jcr(mk2.getNodeStore()).createRepository();
         repos.add(r2);
         Repository r3 = new Jcr(mk3.getNodeStore()).createRepository();
         repos.add(r3);

         Session s1 = r1.login(new SimpleCredentials("admin", "admin".toCharArray()));
         Session s2 = r2.login(new SimpleCredentials("admin", "admin".toCharArray()));
         Session s3 = r3.login(new SimpleCredentials("admin", "admin".toCharArray()));

         ensureIndex(s1.getRootNode(), PROP_NAME);
         runBackgroundOps(mk1);
         runBackgroundOps(mk2);
         runBackgroundOps(mk3);

         // begin test

         Node root2 = s2.getRootNode().addNode("testroot-Worker-2", "nt:unstructured");
         createNodes(root2, "testnode0");
         s2.save();

         createNodes(root2, "testnode1");

         runBackgroundOps(mk1);
         runBackgroundOps(mk3);
         runBackgroundOps(mk2); // publish 'testroot-Worker-2/testnode0'

         Node root3 = s3.getRootNode().addNode("testroot-Worker-3", "nt:unstructured");
         createNodes(root3, "testnode0");

         s2.save();
         createNodes(root2, "testnode2");

         runBackgroundOps(mk1); // sees 'testroot-Worker-2/testnode0'
         runBackgroundOps(mk3); // sees 'testroot-Worker-2/testnode0'
         runBackgroundOps(mk2); // publish 'testroot-Worker-2/testnode1'

         // subsequent read on mk3 will read already published docs from mk2
         s3.save();
         createNodes(root3, "testnode1");

         Node root1 = s1.getRootNode().addNode("testroot-Worker-1", "nt:unstructured");
         createNodes(root1, "testnode0");

         s2.save();
         createNodes(root2, "testnode3");

         runBackgroundOps(mk1);
         runBackgroundOps(mk3);
         runBackgroundOps(mk2);

         s1.save();
         createNodes(root1, "testnode1");

         s3.save();
         createNodes(root3, "testnode2");

         runBackgroundOps(mk1);

         s1.save();

         s1.logout();
         s2.logout();
         s3.logout();
     }

     @Test
     public void rebaseVisibility() throws Exception {
         for (int i = 0; i < 2; i++) {
             MongoConnection c = createConnection();
             DocumentMK mk = new DocumentMK.Builder()
                     .setMongoDB(c.getMongoClient(), c.getDBName())
                     .setAsyncDelay(0)
                     .setClusterId(i + 1).open();
             mks.add(mk);
         }
         final DocumentMK mk1 = mks.get(0);
         final DocumentMK mk2 = mks.get(1);
         Repository r1 = new Jcr(mk1.getNodeStore()).createRepository();
         repos.add(r1);
         Repository r2 = new Jcr(mk2.getNodeStore()).createRepository();
         repos.add(r2);

         Session s1 = r1.login(new SimpleCredentials("admin", "admin".toCharArray()));
         Session s2 = r2.login(new SimpleCredentials("admin", "admin".toCharArray()));

         Node root1 = s1.getRootNode().addNode("session-1");
         s1.save();
         Node root2 = s2.getRootNode().addNode("session-2");
         s2.save();

         runBackgroundOps(mk1);
         runBackgroundOps(mk2);
         runBackgroundOps(mk1);

         createNodes(root1, "nodes");

         createNodes(root2, "nodes");
         s2.save();

         runBackgroundOps(mk2);
         runBackgroundOps(mk1);

         assertFalse(s1.getRootNode().hasNode("session-2/nodes"));

         s1.refresh(true);
         assertTrue(s1.getRootNode().hasNode("session-2/nodes"));

         s1.logout();
         s2.logout();
     }

     private static MongoConnection createConnection() throws Exception {
         return OakMongoNSRepositoryStub.createConnection(
                 ConcurrentAddNodesClusterIT.class.getSimpleName());
     }

     private static void dropDB() throws Exception {
         MongoConnection con = createConnection();
         try {
             con.getDatabase().drop();
         } finally {
             con.close();
         }
     }

     private static void initRepository() throws Exception {
         MongoConnection con = createConnection();
         DocumentMK mk = new DocumentMK.Builder()
                 .setMongoDB(con.getMongoClient(), con.getDBName())
                 .setClusterId(1).open();
         Repository repository = new Jcr(mk.getNodeStore()).createRepository();
         Session session = repository.login(
                 new SimpleCredentials("admin", "admin".toCharArray()));
         session.logout();
         dispose(repository);
         mk.dispose(); // closes connection as well
     }


     private static void ensureIndex(Node root, String propertyName)
             throws RepositoryException {
         Node indexDef = root.getNode(IndexConstants.INDEX_DEFINITIONS_NAME);
         if (indexDef.hasNode(propertyName)) {
             return;
         }
         Node index = indexDef.addNode(propertyName, IndexConstants.INDEX_DEFINITIONS_NODE_TYPE);
         index.setProperty(IndexConstants.TYPE_PROPERTY_NAME,
                 PropertyIndexEditorProvider.TYPE);
         index.setProperty(IndexConstants.REINDEX_PROPERTY_NAME,
                 true);
         index.setProperty(IndexConstants.PROPERTY_NAMES,
                 new String[] { propertyName }, PropertyType.NAME);
         try {
             root.getSession().save();
         } catch (RepositoryException e) {
             // created by other thread -> ignore
             root.getSession().refresh(false);
         }
     }

     private static void runBackgroundOps(DocumentMK mk) throws Exception {
         mk.getNodeStore().runBackgroundOperations();
     }

     private final class Worker implements Runnable {

         private final Repository repo;
         private final Map<String, Exception> exceptions;

         Worker(Repository repo, Map<String, Exception> exceptions) {
             this.repo = repo;
             this.exceptions = exceptions;
         }

         @Override
         public void run() {
             try {
                 Session session = createAdminSession(repo);
                 try {
                     ensureIndex(session.getRootNode(), PROP_NAME);
                     String nodeName = "testroot-" + Thread.currentThread().getName();
                     createNodes(session, nodeName, LOOP_COUNT, NODE_COUNT, exceptions);
                 } finally {
                     session.logout();
                 }
             } catch (Exception e) {
                 exceptions.put(Thread.currentThread().getName(), e);
             }
         }

     }

     private Session createAdminSession(Repository repository) throws RepositoryException {
         return repository.login(new SimpleCredentials("admin", "admin".toCharArray()));
     }

     private void createNodes(Session session,
                              String nodeName,
                              int loopCount,
                              int nodeCount,
                              Map<String, Exception> exceptions)
             throws RepositoryException {
         Node root = session.getRootNode().addNode(nodeName, "nt:unstructured");
         for (int i = 0; i < loopCount; i++) {
             Node node = root.addNode("testnode" + i, "nt:unstructured");
             for (int j = 0; j < nodeCount; j++) {
                 Node child = node.addNode("node" + j, "nt:unstructured");
                 child.setProperty(PROP_NAME, j);
             }
             if (!exceptions.isEmpty()) {
                 break;
             }
             session.save();
         }
     }

     private void createNodes(Node parent, String child)
             throws RepositoryException {
         Node node = parent.addNode(child, "nt:unstructured");
         for (int i = 0; i < NODE_COUNT; i++) {
             Node c = node.addNode("node" + i, "nt:unstructured");
             c.setProperty(PROP_NAME, i);
         }
     }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You under the Apache License, Version 2.0
	* (the "License"); you may not use this file except in compliance with
	* the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package org.apache.jackrabbit.oak.jcr;

	import static java.lang.Thread.UncaughtExceptionHandler;
	import static org.apache.jackrabbit.oak.jcr.AbstractRepositoryTest.dispose;
	import static org.junit.Assert.assertFalse;
	import static org.junit.Assert.assertTrue;

	import java.util.ArrayList;
	import java.util.Collections;
	import java.util.HashMap;
	import java.util.List;
	import java.util.Map;
	import java.util.concurrent.CountDownLatch;
	import java.util.concurrent.TimeUnit;
	import java.util.concurrent.atomic.AtomicBoolean;

	import javax.jcr.Node;
	import javax.jcr.PropertyType;
	import javax.jcr.Repository;
	import javax.jcr.RepositoryException;
	import javax.jcr.Session;
	import javax.jcr.SimpleCredentials;

	import com.google.common.util.concurrent.Uninterruptibles;
	import org.apache.jackrabbit.oak.plugins.document.DocumentMK;
	import org.apache.jackrabbit.oak.plugins.document.util.MongoConnection;
	import org.apache.jackrabbit.oak.plugins.index.IndexConstants;
	import org.apache.jackrabbit.oak.plugins.index.property.PropertyIndexEditorProvider;
	import org.junit.After;
	import org.junit.Assume;
	import org.junit.Before;
	import org.junit.BeforeClass;
	import org.junit.Ignore;
	import org.junit.Test;

	/**
	* Concurrently add nodes with multiple sessions on multiple cluster nodes.
	*/
	public class ConcurrentAddNodesClusterIT {

	private static final int NUM_CLUSTER_NODES = 5;
	private static final int NODE_COUNT = 100;
	private static final int LOOP_COUNT = 10;
	private static final int WORKER_COUNT = 20;
	private static final String PROP_NAME = "testcount";

	private List<Repository> repos = new ArrayList<Repository>();
	private List<DocumentMK> mks = new ArrayList<DocumentMK>();
	private List<Thread> workers = new ArrayList<Thread>();

	@BeforeClass
	public static void mongoDBAvailable() {
	Assume.assumeTrue(OakMongoNSRepositoryStub.isMongoDBAvailable());
	}

	@Before
	public void before() throws Exception {
	dropDB();
	initRepository();
	}

	@After
	public void after() throws Exception {
	workers.clear();
	for (Repository repo : repos) {
	dispose(repo);
	}
	repos.clear();
	for (DocumentMK mk : mks) {
	mk.dispose();
	}
	mks.clear();
	dropDB();
	}

	@Test
	public void addNodesConcurrent() throws Exception {
	for (int i = 0; i < NUM_CLUSTER_NODES; i++) {
	MongoConnection c = createConnection();
	DocumentMK mk = new DocumentMK.Builder()
	.setMongoDB(c.getMongoClient(), c.getDBName())
	.setClusterId(i + 1).open();
	mks.add(mk);
	}
	Map<String, Exception> exceptions = Collections.synchronizedMap(
	new HashMap<String, Exception>());
	for (int i = 0; i < mks.size(); i++) {
	DocumentMK mk = mks.get(i);
	Repository repo = new Jcr(mk.getNodeStore()).createRepository();
	repos.add(repo);
	workers.add(new Thread(new Worker(repo, exceptions), "Worker-" + (i + 1)));
	}
	for (Thread t : workers) {
	t.start();
	}
	for (Thread t : workers) {
	t.join();
	}
	for (Map.Entry<String, Exception> entry : exceptions.entrySet()) {
	// System.out.println("exception in thread " + entry.getKey());
	throw entry.getValue();
	}
	}

	@Ignore("OAK-1807")
	@Test
	public void addNodesConcurrent2() throws Exception {
	final Thread mainThread = Thread.currentThread();
	for (int i = 0; i < NUM_CLUSTER_NODES; i++) {
	MongoConnection c = createConnection();
	DocumentMK mk = new DocumentMK.Builder()
	.setMongoDB(c.getMongoClient(), c.getDBName())
	.setClusterId(i + 1).open();
	mks.add(mk);
	}
	final Map<String, Exception> exceptions = Collections.synchronizedMap(
	new HashMap<String, Exception>());
	final CountDownLatch latch = new CountDownLatch(1);
	final AtomicBoolean stop = new AtomicBoolean();
	final UncaughtExceptionHandler ueh = new UncaughtExceptionHandler() {
	@Override
	public void uncaughtException(Thread t, Throwable e) {
	RuntimeException r = new RuntimeException("Exception in thread "+t.getName(), e);
	r.printStackTrace();
	}
	};
	for (int i = 0; i < mks.size(); i++) {
	DocumentMK mk = mks.get(i);
	final Repository repo = new Jcr(mk.getNodeStore()).createRepository();
	repos.add(repo);
	for (int w = 0; w <= WORKER_COUNT; w++) {
	final String name = "Worker-" + (i + 1) + "-" + (w + 1);
	final Runnable r = new Runnable() {
	final Session session = createAdminSession(repo);
	int count = 0;
	@Override
	public void run() {
	try {
	Uninterruptibles.awaitUninterruptibly(latch);
	session.refresh(false);
	Node node = session.getRootNode().addNode(name+count++, "oak:Unstructured");
	for (int j = 0; j < NODE_COUNT && !stop.get() ; j++) {
	node.addNode("node" + j);
	session.save();
	}
	} catch (RepositoryException e) {
	RuntimeException r = new RuntimeException("Exception in thread "+name, e);
	r.printStackTrace();
	exceptions.put(Thread.currentThread().getName(), r);
	stop.set(true);
	mainThread.interrupt();
	} finally {
	session.logout();
	}
	}
	};

	//Last runnable would be a long running one
	Runnable runnable = r;
	if(w == WORKER_COUNT){
	runnable = new Runnable() {
	@Override
	public void run() {
	while(!stop.get()){
	r.run();
	}
	}
	};
	}

	Thread t = new Thread(runnable);
	t.setName(name);
	t.setUncaughtExceptionHandler(ueh);
	workers.add(t);
	}
	}
	for (Thread t : workers) {
	t.start();
	}
	latch.countDown();

	TimeUnit.MINUTES.sleep(10);
	stop.set(true);

	for (Thread t : workers) {
	t.join();
	}
	for (Map.Entry<String, Exception> entry : exceptions.entrySet()) {
	// System.out.println("exception in thread " + entry.getKey());
	throw entry.getValue();
	}
	}

	@Test
	public void addNodes() throws Exception {
	for (int i = 0; i < 2; i++) {
	MongoConnection c = createConnection();
	DocumentMK mk = new DocumentMK.Builder()
	.setMongoDB(c.getMongoClient(), c.getDBName())
	.setClusterId(i + 1).open();
	mks.add(mk);
	}
	final DocumentMK mk1 = mks.get(0);
	final DocumentMK mk2 = mks.get(1);
	Repository r1 = new Jcr(mk1.getNodeStore()).createRepository();
	repos.add(r1);
	Repository r2 = new Jcr(mk2.getNodeStore()).createRepository();
	repos.add(r2);

	Session s1 = r1.login(new SimpleCredentials("admin", "admin".toCharArray()));
	Session s2 = r2.login(new SimpleCredentials("admin", "admin".toCharArray()));

	ensureIndex(s1.getRootNode(), PROP_NAME);
	ensureIndex(s2.getRootNode(), PROP_NAME);

	Map<String, Exception> exceptions = Collections.synchronizedMap(
	new HashMap<String, Exception>());
	createNodes(s1, "testroot-1", 1, 1, exceptions);
	createNodes(s2, "testroot-2", 1, 1, exceptions);

	for (Map.Entry<String, Exception> entry : exceptions.entrySet()) {
	throw entry.getValue();
	}

	s1.logout();
	s2.logout();
	}

	@Test
	public void addNodes2() throws Exception {
	for (int i = 0; i < 3; i++) {
	MongoConnection c = createConnection();
	DocumentMK mk = new DocumentMK.Builder()
	.setMongoDB(c.getMongoClient(), c.getDBName())
	.setAsyncDelay(0)
	.setClusterId(i + 1).open();
	mks.add(mk);
	}
	final DocumentMK mk1 = mks.get(0);
	final DocumentMK mk2 = mks.get(1);
	final DocumentMK mk3 = mks.get(2);
	Repository r1 = new Jcr(mk1.getNodeStore()).createRepository();
	repos.add(r1);
	Repository r2 = new Jcr(mk2.getNodeStore()).createRepository();
	repos.add(r2);
	Repository r3 = new Jcr(mk3.getNodeStore()).createRepository();
	repos.add(r3);

	Session s1 = r1.login(new SimpleCredentials("admin", "admin".toCharArray()));
	Session s2 = r2.login(new SimpleCredentials("admin", "admin".toCharArray()));
	Session s3 = r3.login(new SimpleCredentials("admin", "admin".toCharArray()));

	ensureIndex(s1.getRootNode(), PROP_NAME);
	runBackgroundOps(mk1);
	runBackgroundOps(mk2);
	runBackgroundOps(mk3);

	// begin test

	Node root2 = s2.getRootNode().addNode("testroot-Worker-2", "nt:unstructured");
	createNodes(root2, "testnode0");
	s2.save();

	createNodes(root2, "testnode1");

	runBackgroundOps(mk1);
	runBackgroundOps(mk3);
	runBackgroundOps(mk2); // publish 'testroot-Worker-2/testnode0'

	Node root3 = s3.getRootNode().addNode("testroot-Worker-3", "nt:unstructured");
	createNodes(root3, "testnode0");

	s2.save();
	createNodes(root2, "testnode2");

	runBackgroundOps(mk1); // sees 'testroot-Worker-2/testnode0'
	runBackgroundOps(mk3); // sees 'testroot-Worker-2/testnode0'
	runBackgroundOps(mk2); // publish 'testroot-Worker-2/testnode1'

	// subsequent read on mk3 will read already published docs from mk2
	s3.save();
	createNodes(root3, "testnode1");

	Node root1 = s1.getRootNode().addNode("testroot-Worker-1", "nt:unstructured");
	createNodes(root1, "testnode0");

	s2.save();
	createNodes(root2, "testnode3");

	runBackgroundOps(mk1);
	runBackgroundOps(mk3);
	runBackgroundOps(mk2);

	s1.save();
	createNodes(root1, "testnode1");

	s3.save();
	createNodes(root3, "testnode2");

	runBackgroundOps(mk1);

	s1.save();

	s1.logout();
	s2.logout();
	s3.logout();
	}

	@Test
	public void rebaseVisibility() throws Exception {
	for (int i = 0; i < 2; i++) {
	MongoConnection c = createConnection();
	DocumentMK mk = new DocumentMK.Builder()
	.setMongoDB(c.getMongoClient(), c.getDBName())
	.setAsyncDelay(0)
	.setClusterId(i + 1).open();
	mks.add(mk);
	}
	final DocumentMK mk1 = mks.get(0);
	final DocumentMK mk2 = mks.get(1);
	Repository r1 = new Jcr(mk1.getNodeStore()).createRepository();
	repos.add(r1);
	Repository r2 = new Jcr(mk2.getNodeStore()).createRepository();
	repos.add(r2);

	Session s1 = r1.login(new SimpleCredentials("admin", "admin".toCharArray()));
	Session s2 = r2.login(new SimpleCredentials("admin", "admin".toCharArray()));

	Node root1 = s1.getRootNode().addNode("session-1");
	s1.save();
	Node root2 = s2.getRootNode().addNode("session-2");
	s2.save();

	runBackgroundOps(mk1);
	runBackgroundOps(mk2);
	runBackgroundOps(mk1);

	createNodes(root1, "nodes");

	createNodes(root2, "nodes");
	s2.save();

	runBackgroundOps(mk2);
	runBackgroundOps(mk1);

	assertFalse(s1.getRootNode().hasNode("session-2/nodes"));

	s1.refresh(true);
	assertTrue(s1.getRootNode().hasNode("session-2/nodes"));

	s1.logout();
	s2.logout();
	}

	private static MongoConnection createConnection() throws Exception {
	return OakMongoNSRepositoryStub.createConnection(
	ConcurrentAddNodesClusterIT.class.getSimpleName());
	}

	private static void dropDB() throws Exception {
	MongoConnection con = createConnection();
	try {
	con.getDatabase().drop();
	} finally {
	con.close();
	}
	}

	private static void initRepository() throws Exception {
	MongoConnection con = createConnection();
	DocumentMK mk = new DocumentMK.Builder()
	.setMongoDB(con.getMongoClient(), con.getDBName())
	.setClusterId(1).open();
	Repository repository = new Jcr(mk.getNodeStore()).createRepository();
	Session session = repository.login(
	new SimpleCredentials("admin", "admin".toCharArray()));
	session.logout();
	dispose(repository);
	mk.dispose(); // closes connection as well
	}


	private static void ensureIndex(Node root, String propertyName)
	throws RepositoryException {
	Node indexDef = root.getNode(IndexConstants.INDEX_DEFINITIONS_NAME);
	if (indexDef.hasNode(propertyName)) {
	return;
	}
	Node index = indexDef.addNode(propertyName, IndexConstants.INDEX_DEFINITIONS_NODE_TYPE);
	index.setProperty(IndexConstants.TYPE_PROPERTY_NAME,
	PropertyIndexEditorProvider.TYPE);
	index.setProperty(IndexConstants.REINDEX_PROPERTY_NAME,
	true);
	index.setProperty(IndexConstants.PROPERTY_NAMES,
	new String[] { propertyName }, PropertyType.NAME);
	try {
	root.getSession().save();
	} catch (RepositoryException e) {
	// created by other thread -> ignore
	root.getSession().refresh(false);
	}
	}

	private static void runBackgroundOps(DocumentMK mk) throws Exception {
	mk.getNodeStore().runBackgroundOperations();
	}

	private final class Worker implements Runnable {

	private final Repository repo;
	private final Map<String, Exception> exceptions;

	Worker(Repository repo, Map<String, Exception> exceptions) {
	this.repo = repo;
	this.exceptions = exceptions;
	}

	@Override
	public void run() {
	try {
	Session session = createAdminSession(repo);
	try {
	ensureIndex(session.getRootNode(), PROP_NAME);
	String nodeName = "testroot-" + Thread.currentThread().getName();
	createNodes(session, nodeName, LOOP_COUNT, NODE_COUNT, exceptions);
	} finally {
	session.logout();
	}
	} catch (Exception e) {
	exceptions.put(Thread.currentThread().getName(), e);
	}
	}

	}

	private Session createAdminSession(Repository repository) throws RepositoryException {
	return repository.login(new SimpleCredentials("admin", "admin".toCharArray()));
	}

	private void createNodes(Session session,
	String nodeName,
	int loopCount,
	int nodeCount,
	Map<String, Exception> exceptions)
	throws RepositoryException {
	Node root = session.getRootNode().addNode(nodeName, "nt:unstructured");
	for (int i = 0; i < loopCount; i++) {
	Node node = root.addNode("testnode" + i, "nt:unstructured");
	for (int j = 0; j < nodeCount; j++) {
	Node child = node.addNode("node" + j, "nt:unstructured");
	child.setProperty(PROP_NAME, j);
	}
	if (!exceptions.isEmpty()) {
	break;
	}
	session.save();
	}
	}

	private void createNodes(Node parent, String child)
	throws RepositoryException {
	Node node = parent.addNode(child, "nt:unstructured");
	for (int i = 0; i < NODE_COUNT; i++) {
	Node c = node.addNode("node" + i, "nt:unstructured");
	c.setProperty(PROP_NAME, i);
	}
	}
	}