Missing matched documents on searches and updates reproduction

This blog recently exposed an interesting concurrency caveat related to MongoDB where matching documents won't be found (or updated) if they are being reindexes.

The only part in the entry I was missing is a way how to reproduce this issue. So I decided to create a test which you can test against your version of MongoDB to check if it is still a problem.

Here it is:

package co.uk.matejtymes.mongodb;

import com.mongodb.*;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;

import static com.mongodb.BasicDBObjectBuilder.start;
import static java.util.Arrays.asList;
import static java.util.UUID.randomUUID;
import static java.util.concurrent.Executors.newFixedThreadPool;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.Matchers.hasSize;
import static org.junit.Assert.assertThat;

public class ReindexFailureTest {

    private static final String STATE_FIELD = "state";

    private static final Random RANDOM = new Random();

    private DBCollection coll;

    @Before
    public void setUp() throws Exception {
        // todo: provide connection details for your mongoDB instance
        MongoClient mongo = new MongoClient("localhost", 27017);
        DB db = mongo.getDB("testDb");

        coll = db.getCollection("indexTest");
    }

    @After
    public void tearDown() throws Exception {
        coll.drop();
    }

    @Test
    public void shouldFindAllMatchingItemsEvenWhenRecalculatingIndex()throws Exception {
        int docCount = 250;
        int concurrentUpdates = 40;
        int attemptsCount = 1_000;

        List<String> stateValues = asList("Active", "Inactive");

        coll.createIndex(new BasicDBObject(STATE_FIELD, 1));

        List<String> allIds = createNDocumentsWithState(docCount, stateValues);

        ExecutorService executor = newFixedThreadPool(concurrentUpdates);
        for (int attempt = 1; attempt <= attemptsCount; attempt++) {
            System.out.println(attempt + ". attempt");

            CountDownLatch beginLatch = new CountDownLatch(concurrentUpdates + 1);
            CountDownLatch endLatch = new CountDownLatch(concurrentUpdates + 1);

            for (int update = 0; update < concurrentUpdates; update++) {
                executor.submit(() -> updateState(pickRandomItem(allIds), stateValues, beginLatch, endLatch));
            }

            List<String> foundIds = findDocumentsInState(stateValues, beginLatch, endLatch);

            Set<String> uniqueIds = new HashSet<>();
            Set<String> duplicateIds = new HashSet<>();
            Set<String> missingIds = new HashSet<>(allIds);

            for (String foundId : foundIds) {
                if (uniqueIds.contains(foundId)) {
                    duplicateIds.add(foundId);
                }
                uniqueIds.add(foundId);
                missingIds.remove(foundId);
            }

            if (!missingIds.isEmpty()) {
                System.err.println(missingIds.size() + ". missingIds: " + missingIds);
            }
            if (!duplicateIds.isEmpty()) {
                System.err.println(duplicateIds.size() + ". duplicateIds: " + duplicateIds);
            }

            assertThat(foundIds, hasSize(allIds.size()));
            assertThat(missingIds, hasSize(0));
            assertThat(duplicateIds, hasSize(0));
        }

        executor.shutdown();
        executor.awaitTermination(3, SECONDS);
    }

    @Test
    public void shouldUpdateAllMatchingItemsEvenWhenRecalculatingIndex()throws Exception {
        int docCount = 250;
        int concurrentUpdates = 40;
        int attemptsCount = 1_000;

        List<String> stateValues = asList("Active", "Inactive");

        coll.createIndex(new BasicDBObject(STATE_FIELD, 1));

        List<String> allIds = createNDocumentsWithState(docCount, stateValues);

        ExecutorService executor = newFixedThreadPool(concurrentUpdates);
        for (int attempt = 1; attempt <= attemptsCount; attempt++) {
            System.out.println(attempt + ". attempt");

            String fieldToUpdate = "field" + attempt;
            Object valueToSet = true;

            CountDownLatch beginLatch = new CountDownLatch(concurrentUpdates + 1);
            CountDownLatch endLatch = new CountDownLatch(concurrentUpdates + 1);

            for (int update = 0; update < concurrentUpdates; update++) {
                executor.submit(() -> updateState(pickRandomItem(allIds), stateValues, beginLatch, endLatch));
            }

            BasicDBObject query = new BasicDBObject(STATE_FIELD, new BasicDBObject("$in", stateValues));
            BasicDBObject update = new BasicDBObject("$set", new BasicDBObject(fieldToUpdate, valueToSet));

            beginLatch.countDown();
            int n = coll.updateMulti(query, update).getN();
            endLatch.countDown();


            List<String> updatedIds = new ArrayList<>();
            coll.find(new BasicDBObject(fieldToUpdate, valueToSet)).forEach(
                    dbObject -> updatedIds.add((String) dbObject.get("_id"))
            );

            Set<String> missingIds = new HashSet<>(allIds);
            missingIds.removeAll(updatedIds);


            if (!missingIds.isEmpty()) {
                System.err.println(missingIds.size() + ". missingIds: " + missingIds);
            }
            if (n != allIds.size()) {
                System.err.println("n = " + n);
            }
            if (updatedIds.size() != allIds.size()) {
                System.err.println("updateIds = " + updatedIds.size());
            }

            assertThat(n, equalTo(allIds.size()));
            assertThat(updatedIds, hasSize(allIds.size()));
            assertThat(missingIds, hasSize(0));
        }

        executor.shutdown();
        executor.awaitTermination(3, SECONDS);
    }

    /* ====================== */
    /* --- helper methods --- */
    /* ====================== */

    private List<String> createNDocumentsWithState(int docCount, List<String> stateValues) {
        List<String> ids = new ArrayList<>();

        for (int i = 0; i < docCount; i++) {
            String id = randomUUID().toString();
            String state = stateValues.get(i % stateValues.size());

            DBObject dbObject = start()
                    .add("_id", id)
                    .add(STATE_FIELD, state)
                    .get();
            coll.insert(dbObject);

            ids.add(id);
        }
        return ids;
    }

    private void updateState(String id, List<String> stateValues, CountDownLatch beginLatch, CountDownLatch endLatch) {
        BasicDBObject query = new BasicDBObject("_id", id);

        String oldStateValue = (String) coll.find(query).next().get(STATE_FIELD);
        String newStateValue = stateValues.stream().filter(state -> !state.equals(oldStateValue)).findFirst().get();

        BasicDBObject update = new BasicDBObject("$set", new BasicDBObject(STATE_FIELD, newStateValue));

        beginLatch.countDown();
        coll.update(query, update);
        endLatch.countDown();
    }

    private List<String> findDocumentsInState(List<String> stateValues, CountDownLatch beginLatch, CountDownLatch endLatch) {
        BasicDBObject query = new BasicDBObject(STATE_FIELD, new BasicDBObject("$in", stateValues));
        Iterator<DBObject> dbObjects = coll.find(query).iterator();

        List<String> foundIds = new ArrayList<>();

        beginLatch.countDown();;
        while (dbObjects.hasNext()) {
            foundIds.add((String) dbObjects.next().get("_id"));
        }
        endLatch.countDown();

        return foundIds;
    }

    private static <T> T pickRandomItem(List<T> values) {
        return values.get(RANDOM.nextInt(values.size()));
    }
}

Executor that notifies you when task finish

Java Executors don't let you know when all tasks are finished or to be more precise, don't block you until the tasks are finished. You could call shutdown() on them and then awaitTermination(), but this way you can't reuse the executor anymore, which is not great. This is why I create a class Runner that can accomplish this. It's used like this:

Runner runner = Runner.runner(10);

runner.runIn(2, SECONDS, runnable);
runner.run(runnable);


runner.waitTillDone(); // blocks until all tasks are finished (or failed)


// and reuse it

runner.runIn(500, MILLISECONDS, callable);

runner.waitTillDone();

runner.shutdownAndAwaitTermination();

The code for it can be found here:

https://github.com/MatejTymes/JavaFixes

Hope this will help

Testing multithreaded code

Sometimes you needed to test that your code is thread-safe and can be run from multiple threads at the same time. To help with doing this I wrote an utility class that can make your test code simpler and more readable.

Here is the usage of it. I was testing that SimpleDateFormat is thread safe (actually it is not so this test will fail):

public class SimpleDateFormatTest {

    // this test will fail as SimpleDateFormat is not thread safe
    @Test
    public void shouldWorkConcurrently() throws ExecutionException, InterruptedException {
        // Given
        final DateFormat dateFormat = new SimpleDateFormat();

        final Date date1 = newRandomDate();
        final Date date2 = newRandomDate();
        final Date date3 = newRandomDate();

        ConcurrentExecutor executor = new ConcurrentExecutor();

        Future<String> result1 = executor.addAction(new Callable<String>() {
            public String call() {

                return dateFormat.format(date1);
            }
        });
        Future<String> result2 = executor.addAction(new Callable<String>() {
            public String call() {

                return dateFormat.format(date2);
            }
        });
        Future<String> result3 = executor.addAction(new Callable<String>() {
            public String call() {

                return dateFormat.format(date3);
            }
        });

        // When
        executor.executeAtTheSameTime(); // this will block until all actions are finished

        // Then
        assertThat(result1.get(), is(new SimpleDateFormat().format(date1)));
        assertThat(result2.get(), is(new SimpleDateFormat().format(date2)));
        assertThat(result3.get(), is(new SimpleDateFormat().format(date3)));
    }

    private Date newRandomDate() {
        return new Date(new Random().nextLong());
    }
}

In this test I created a ConcurrentExecutor to which I added 3 actions. Each of those actions calls the format method for one of 3 dates. Then I call the method executeAtTheSameTime which will make sure that all actions will start at the same time. Once the execution is finished I retrieve each result and verify that it is the same as proper single threaded conversion.

One nice thing about this utility is that it is extremely simple to retry the execution. Just rerun the executeAtTheSameTime and the Future results will hold a new value. This way you can put the When and Then section info a for loop and test the execution thread safety multiple times. This is sometimes needed as some multi-threaded issues are not always visible during the first run.

If you would like to add this utility into your project here is the actual implementation of ConcurrentExecutor (please note that it is in java 1.7 so if you're using some older version you might miss some generics definitions):

public class ConcurrentExecutor {

    private List<Callable<?>> actions = new ArrayList<>();
    private List<Object> results = new ArrayList<>();

    private volatile boolean finished = false;

    public <T> Future<T> addAction(Callable<T> callable) {
        actions.add(callable);
        int actionIndex = actions.size() - 1;
        return new FutureResult<>(actionIndex);
    }

    public void executeAtTheSameTime() {
        try {
            finished = false;
            results.clear();

            ExecutorService executor = Executors.newFixedThreadPool(actions.size());

            CyclicBarrier barrier = new CyclicBarrier(actions.size());
            CountDownLatch doneCountDown = new CountDownLatch(actions.size());

            List<Future<?>> futureResults = new ArrayList<>();
            for (Callable<?> action : actions)
            {
                futureResults.add(executor.submit(new ActionCallable<>(action, barrier, doneCountDown)));
            }
            doneCountDown.await();

            for (Future<?> futureResult : futureResults)
            {
                try
                {
                    results.add(futureResult.get(500, TimeUnit.MILLISECONDS));
                }
                catch (Exception e)
                {
                    throw new RuntimeException("not able to retrieve result from thread execution", e);
                }
            }

            executor.shutdownNow();

            finished = true;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    class ActionCallable<T> implements Callable<T> {

        private final Callable<T> action;
        private final CyclicBarrier startBarrier;
        private final CountDownLatch doneCountDown;

        ActionCallable(Callable<T> action, CyclicBarrier startBarrier, CountDownLatch doneCountDown) {
            this.action = action;
            this.startBarrier = startBarrier;
            this.doneCountDown = doneCountDown;
        }

        @Override
        public T call() throws Exception {
            T result;

            startBarrier.await();
            result = action.call();
            doneCountDown.countDown();

            return result;
        }
    }

    class FutureResult<T> implements Future<T> {

        private final int resultIndex;

        public FutureResult(int resultIndex) {
            this.resultIndex = resultIndex;
        }

        @Override
        public boolean isDone() {
            return finished;
        }

        @Override
        @SuppressWarnings("unchecked")
        public T get() throws InterruptedException, ExecutionException {
            if (!isDone()) {
                throw new IllegalStateException("execution has not yet finished");
            }
            return (T) results.get(resultIndex);
        }

        @Override
        public boolean cancel(boolean mayInterruptIfRunning) {
            throw new RuntimeException("not implemented by intention");
        }

        @Override
        public boolean isCancelled() {
            throw new RuntimeException("not implemented by intention");
        }

        @Override
        public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
            throw new RuntimeException("not implemented by intention");
        }
    }
}