/*
 * The contents of this file are subject to the terms of the Common Development and
 * Distribution License (the License). You may not use this file except in compliance with the
 * License.
 *
 * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
 * specific language governing permission and limitations under the License.
 *
 * When distributing Covered Software, include this CDDL Header Notice in each file and include
 * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
 * Header, with the fields enclosed by brackets [] replaced by your own identifying
 * information: "Portions Copyright [year] [name of copyright owner]".
 *
 * Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.backends.jdbc;

import org.forgerock.opendj.ldap.ByteString;
import org.forgerock.opendj.server.config.server.JDBCBackendCfg;
import org.opends.server.DirectoryServerTestCase;
import org.opends.server.backends.pluggable.spi.AccessMode;
import org.opends.server.backends.pluggable.spi.Importer;
import org.opends.server.backends.pluggable.spi.StorageRuntimeException;
import org.opends.server.backends.pluggable.spi.TreeName;
import org.testng.annotations.AfterClass;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;

import java.sql.Connection;
import java.sql.Driver;
import java.sql.DriverManager;
import java.sql.DriverPropertyInfo;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.IdentityHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;
import java.util.logging.Logger;

import static org.forgerock.opendj.config.ConfigurationMock.mockCfg;
import static org.mockito.Mockito.anyInt;
import static org.mockito.Mockito.anyString;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;

/**
 * The {@link Importer} of the JDBC backend is used by several threads at once - phase two of an
 * import runs one thread per tree - and {@code java.sql.Connection} is not thread-safe. Sharing
 * one connection between those threads corrupts the driver rather than merely serializing the
 * work: the sql server driver keeps the reconnect listeners of a connection in a plain
 * {@code ArrayList} that every {@code prepareStatement()} and every statement {@code close()}
 * mutates, and two import threads on one connection walked it past the end of its array
 * (issue #891).
 * <p>
 * Needs no database: the connections are handed out by a driver of this test, which records every
 * thread inside a statement of a connection - and which thread commits, rolls back or returns one
 * while another is in there - so what it records is the defect itself, two threads on one
 * connection at the same time, rather than an exception of one driver. It therefore holds for every
 * dialect this backend takes.
 * <p>
 * Two ways of getting the threads to overlap, because trees that get connections of their own and
 * trees that share one cannot be asked the same question. Where the connections differ, the driver
 * holds the first statement of each thread at a rendezvous until its peers have one in flight of
 * their own. Where they do not, a second thread never reaches such a rendezvous - the monitor of
 * the connection is what this is about - so the threads line up before they write instead, and the
 * first one in holds the connection for a moment to see whether the other turns up beside it.
 */
@SuppressWarnings("javadoc")
@Test(groups = { "precommit", "jdbc" }, sequential = true)
public class ImportConnectionsTestCase extends DirectoryServerTestCase {

	/** Long enough for a peer thread to reach a statement on a loaded machine, and no longer. */
	private static final long RENDEZVOUS_SECONDS = 30;

	/**
	 * How long the first thread inside a statement holds the connection it is on. Paid in full by
	 * every run where the importer serializes its threads as it should - which is the point: a peer
	 * that is let in arrives at once, having been waiting on the monitor since the starting line.
	 */
	private static final long HOLD_SECONDS = 2;

	/** How long a close of an import may take before a thread of the test is taken to be stuck. */
	private static final long CLOSE_SECONDS = 30;

	/**
	 * How long a thread parked inside a statement stays in there before it gives up on the test
	 * that parked it. Paid by no passing run - the test lets it go as soon as it has what it came
	 * for - and it is bounded so that a run which never gets there is a red test rather than a
	 * build that hangs on a monitor nobody is going to release.
	 */
	private static final long PARK_SECONDS = 30;

	private static final TreeName ID2ENTRY = new TreeName("dc=example,dc=com", "id2entry");
	private static final TreeName DN2ID = new TreeName("dc=example,dc=com", "dn2id");
	private static final TreeName STATE = new TreeName("dc=example,dc=com", "state");

	private final StubDriver stub = new StubDriver();
	/**
	 * What the driver of this test answers a connect with, or null for a connection of its own.
	 * A failure per connect rather than one instance thrown over and over: a caller is free to
	 * give what it caught a cause and a suppressed exception of its own.
	 */
	private final AtomicReference<Supplier<SQLException>> refusal = new AtomicReference<>();

	/** The connections an import issued a statement on, in no particular order. */
	private final Set<Connection> used =
		Collections.synchronizedSet(Collections.newSetFromMap(new IdentityHashMap<Connection, Boolean>()));
	/** The same, in the order the import first issued a statement on them. */
	private final List<Connection> usedInOrder = Collections.synchronizedList(new ArrayList<Connection>());
	/** Every commit of the import, in the order the connections took them. */
	private final List<Connection> commits = Collections.synchronizedList(new ArrayList<Connection>());
	/**
	 * The threads inside a statement of a connection right now - every one of them, rather than the
	 * first to arrive: a thread that finds a peer there records the pair and stays out of the map,
	 * and the peer's exit would then leave the map saying the connection is free while it is not.
	 * What comes after - another statement, a commit, the rollback of a return - would be recorded
	 * against nobody, which is the one thing this suite is here to notice.
	 */
	private final Map<Connection, Set<Thread>> inStatement =
		Collections.synchronizedMap(new IdentityHashMap<Connection, Set<Thread>>());
	/** Every pair of threads that was inside a statement of one connection at the same time. */
	private final List<String> shared = Collections.synchronizedList(new ArrayList<String>());

	/** Where the threads of a test meet, so that their statements are in flight at the same time. */
	private volatile CyclicBarrier rendezvous;
	/** Whether this thread has already met its peers: a thread waits there once, whatever it issues after. */
	private final ThreadLocal<Boolean> met = ThreadLocal.withInitial(() -> Boolean.FALSE);

	/**
	 * Where the threads of a test line up before they write, which is in front of every monitor the
	 * importer takes - unlike {@link #rendezvous}, which they reach once they are already inside a
	 * statement, and which two threads sharing a connection therefore cannot both reach.
	 */
	private volatile CyclicBarrier startingLine;
	/**
	 * Counted down by every thread that gets inside a statement, and waited on by the first of them:
	 * it holds the connection long enough for a peer to get in there beside it, if anything lets it.
	 */
	private volatile CountDownLatch insideAStatement;
	/** Whether this thread has already held a connection: it holds the first statement it issues. */
	private final ThreadLocal<Boolean> held = ThreadLocal.withInitial(() -> Boolean.FALSE);

	/**
	 * Counted down by the first thread of a test to get inside a statement, and awaited by the test
	 * itself: what a write in flight looks like to the rest of the import - a connection whose
	 * statement has not come back, and which therefore has nothing recorded against it yet.
	 */
	private volatile CountDownLatch insideAStatementNow;
	/** Counted down by the test to let the parked write finish, once it has seen what it came to see. */
	private volatile CountDownLatch letTheStatementFinish;
	/** Whether this thread has already been parked: a thread is parked in the first statement it issues. */
	private final ThreadLocal<Boolean> parked = ThreadLocal.withInitial(() -> Boolean.FALSE);

	private ExecutorService threads;
	/** The pool bound and its borrow deadline as this JVM had them, put back after every test. */
	private String poolMaxOfTheJvm;
	private String poolTimeoutOfTheJvm;

	@BeforeClass
	public void registerStubDriver() throws Exception {
		DriverManager.registerDriver(stub);
	}

	/**
	 * The bound of the pools of this suite, pinned: how many connections an import takes is clamped
	 * to it, so a value another suite of this package left set - CachedConnectionTestCase varies it
	 * - would decide the counts asserted here. A pool reads it once, when the first borrow of a
	 * connection string creates it, so it has to stand before the storage of a test is opened.
	 */
	@BeforeMethod
	public void pinThePoolBound() {
		poolMaxOfTheJvm = System.getProperty(CachedConnection.POOL_MAX_PROPERTY);
		poolTimeoutOfTheJvm = System.getProperty(CachedConnection.POOL_TIMEOUT_PROPERTY);
		System.setProperty(CachedConnection.POOL_MAX_PROPERTY, "16");
	}

	@AfterClass
	public void deregisterStubDriver() throws Exception {
		DriverManager.deregisterDriver(stub);
	}

	@AfterMethod
	public void forgetWhatTheTestRecorded() {
		System.clearProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY);
		// put back rather than cleared: a sibling suite of this package varies this property, and a
		// pool is created with the value that stands when its connection string is first borrowed on
		putBack(CachedConnection.POOL_MAX_PROPERTY, poolMaxOfTheJvm);
		putBack(CachedConnection.POOL_TIMEOUT_PROPERTY, poolTimeoutOfTheJvm);
		// the rendezvous of the next test is not the one this thread already went to
		met.remove();
		held.remove();
		parked.remove();
		// let go of anything this test left parked before its threads are shut down: a write held
		// inside a statement holds the monitor of a connection, and the next test of this suite
		// starts by opening a storage on that pool
		if (letTheStatementFinish != null) {
			letTheStatementFinish.countDown();
		}
		if (threads != null) {
			threads.shutdownNow();
			threads = null;
		}
		refusal.set(null);
		rendezvous = null;
		startingLine = null;
		insideAStatement = null;
		insideAStatementNow = null;
		letTheStatementFinish = null;
		forgetTheConnections();
		inStatement.clear();
		shared.clear();
	}

	private static void putBack(String property, String value) {
		if (value == null) {
			System.clearProperty(property);
		} else {
			System.setProperty(property, value);
		}
	}

	private void forgetTheConnections() {
		used.clear();
		usedInOrder.clear();
		commits.clear();
	}

	/**
	 * Two trees written at the same time are written through connections of their own. This is the
	 * contract {@code Importer} states and the defect of #891: every thread of phase two wrote
	 * through the one connection the importer borrowed in its constructor.
	 */
	@Test(timeOut = 120000)
	public void testTwoTreesAreWrittenThroughConnectionsOfTheirOwn() throws Exception {
		System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "2");
		final String url = StubDriver.PREFIX + "importer-parallel";
		final JDBCStorage storage = importingStorage(url);
		final Importer importer = storage.startImport();
		try {
			importing(importer, () -> {
				// both threads have to be inside a statement at once for the record below to say
				// anything: a run where one finished before the other started shares no connection
				// however the importer borrows them
				rendezvous = new CyclicBarrier(2);
				threads = daemonThreads(2, "import-writer");
				final Future<?> id2entry = put(importer, ID2ENTRY);
				final Future<?> dn2id = put(importer, DN2ID);
				id2entry.get(RENDEZVOUS_SECONDS * 2, TimeUnit.SECONDS);
				dn2id.get(RENDEZVOUS_SECONDS * 2, TimeUnit.SECONDS);

				// belt and braces here rather than the cover of this suite: with a connection each
				// the record below is empty however the import behaves, and a run where the two
				// trees did share one fails at the rendezvous instead - the second thread would be
				// on the monitor of the connection and never reach it. What pins the monitor is
				// testTwoTreesOnOneConnectionAreNotWrittenAtTheSameTime
				assertEquals(shared, Collections.emptyList(),
					"two import threads issued a statement on one connection at the same time");
				assertEquals(used.size(), 2, "the two trees of the import did not get a connection each");
			});
		} finally {
			storage.close();
		}
	}

	/**
	 * Two trees that share one connection are not written at the same time. The bound of an import,
	 * and a pool with nothing to spare, both put several trees on one connection - and what #891 is
	 * about is two threads issuing statements on one connection, not how many connections an import
	 * holds: the monitor of the connection is what has to keep the second thread out.
	 * <p>
	 * The two threads cannot be brought together at a barrier inside their statements the way the
	 * test above does it - the second one never gets in to reach it - so the first one in holds the
	 * connection for {@link #HOLD_SECONDS} instead, having lined up with its peer beforehand. A
	 * second thread let in beside it arrives at once, since it has been waiting on that monitor
	 * since the starting line, and both are then recorded in {@link #shared}.
	 */
	@Test(timeOut = 120000)
	public void testTwoTreesOnOneConnectionAreNotWrittenAtTheSameTime() throws Exception {
		System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "1"); // both trees on one connection
		final String url = StubDriver.PREFIX + "importer-serialized";
		final JDBCStorage storage = importingStorage(url);
		final Importer importer = storage.startImport();
		try {
			importing(importer, () -> {
				startingLine = new CyclicBarrier(2);
				insideAStatement = new CountDownLatch(2);
				threads = daemonThreads(2, "import-writer");
				final Future<?> id2entry = put(importer, ID2ENTRY);
				final Future<?> dn2id = put(importer, DN2ID);
				id2entry.get(RENDEZVOUS_SECONDS * 2, TimeUnit.SECONDS);
				dn2id.get(RENDEZVOUS_SECONDS * 2, TimeUnit.SECONDS);

				assertEquals(used.size(), 1, "the two trees of this import did not share one connection");
				assertEquals(shared, Collections.emptyList(),
					"two import threads issued a statement on one connection at the same time");
			});
		} finally {
			storage.close();
		}
	}

	/**
	 * ... and no more connections than that: an import takes them for its whole duration, and the
	 * pool they come from is bounded and shared with the operations of every other backend on that
	 * database (#878). One thread per tree is what phase two runs, and a default backend has trees
	 * enough to empty a default pool.
	 */
	@Test(timeOut = 120000)
	public void testTreesShareTheConnectionsOfABoundAboveOne() throws Exception {
		System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "2");
		final String url = StubDriver.PREFIX + "importer-bounded-two";
		final JDBCStorage storage = importingStorage(url);
		final Importer importer = storage.startImport();
		try {
			// five trees over two connections: the bound is what the round robin wraps at
			importer.put(ID2ENTRY, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
			importer.put(DN2ID, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
			importer.put(STATE, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
			importer.put(new TreeName("dc=example,dc=com", "id2childrencount"),
				ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
			importer.put(new TreeName("dc=example,dc=com", "referral"),
				ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));

			assertEquals(used.size(), 2, "the import did not spread its trees over the connections of its bound");
		} finally {
			importer.close();
			storage.close();
		}
	}

	/**
	 * The bound an import takes where the operator set none, which is the branch every real install
	 * takes: half the bound of the pool, and no more than half of what the default bound would be.
	 * An import holds what it takes for its whole duration out of a pool that is shared with the
	 * LDAP traffic of every backend on that database (#878), so the deployment that raised the
	 * bound of its pool did not thereby ask for an import to take half of it.
	 */
	@Test(timeOut = 120000)
	public void testTheBoundOfAnImportDefaultsToHalfThePool() throws Exception {
		// set by no test of this suite, and cleared after every one of them: this is the default
		System.clearProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY);
		// a bound of the pool that is not the default one: DEFAULT_POOL_MAX is
		// max(16, processors * 2), so the 16 pinThePoolBound() puts on the pools of this suite is
		// the default itself on a machine of eight cores or fewer - and half the pool and half the
		// default are then the same number, which is no test of which of the two an import halves
		System.setProperty(CachedConnection.POOL_MAX_PROPERTY, "6");
		final JDBCStorage half = importingStorage(StubDriver.PREFIX + "importer-default-bound");
		try {
			assertEquals(half.importConnections(), 3, "an import did not default to half the bound of the pool");
		} finally {
			half.close();
		}
		System.setProperty(CachedConnection.POOL_MAX_PROPERTY, String.valueOf(CachedConnection.DEFAULT_POOL_MAX * 4));
		final JDBCStorage capped = importingStorage(StubDriver.PREFIX + "importer-default-bound-large-pool");
		try {
			assertEquals(capped.importConnections(), CachedConnection.DEFAULT_POOL_MAX / 2,
				"an import of a pool larger than the default took more than half of what the default bound would be");
		} finally {
			capped.close();
		}
	}

	/** ... and a bound the pool cannot honour is the pool's, not the property's. */
	@Test(timeOut = 120000)
	public void testTheBoundOfAnImportIsClampedToTheBoundOfThePool() throws Exception {
		System.setProperty(CachedConnection.POOL_MAX_PROPERTY, "2");
		System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "8");
		final String url = StubDriver.PREFIX + "importer-clamped";
		final JDBCStorage storage = importingStorage(url);
		try {
			assertEquals(storage.importConnections(), 2, "the bound of an import was not clamped to the pool");
		} finally {
			storage.close();
		}
	}

	@Test(timeOut = 120000)
	public void testTheConnectionsOfAnImportAreBounded() throws Exception {
		System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "1");
		final String url = StubDriver.PREFIX + "importer-bounded";
		final JDBCStorage storage = importingStorage(url);
		final Importer importer = storage.startImport();
		try {
			importer.put(ID2ENTRY, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
			importer.put(DN2ID, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
			importer.put(STATE, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));

			assertEquals(used.size(), 1, "the import took more connections than the bound allows");
		} finally {
			importer.close();
			storage.close();
		}
	}

	/**
	 * The trees decide which connection a write goes through, not the threads that write them. The
	 * connections of an import are transactions of their own, so two of them writing one row would
	 * have the second wait for the first to commit - which an import does in {@code close()}, when
	 * the thread that would have to release the row is long done, and the bulk class carries no
	 * bound to break the wait. {@code setTrust()} writes the state tree of a container from an
	 * import thread in {@code beforePhaseOne} and from the closing thread in {@code afterPhaseTwo},
	 * which is exactly that shape.
	 */
	@Test(timeOut = 120000)
	public void testOneTreeIsWrittenThroughOneConnectionWhoeverWritesIt() throws Exception {
		System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "4");
		final String url = StubDriver.PREFIX + "importer-one-tree";
		final JDBCStorage storage = importingStorage(url);
		final Importer importer = storage.startImport();
		try {
			importing(importer, () -> {
				// three threads, one tree: an import thread of phase one, another of phase two, and
				// the thread that closes the importer, which is the one afterPhaseTwo runs on
				putOnAThreadOfItsOwn(importer, STATE);
				putOnAThreadOfItsOwn(importer, STATE);
				importer.put(STATE, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));

				assertEquals(used.size(), 1, "one tree was written through more than one connection");
			});
		} finally {
			storage.close();
		}
	}

	/**
	 * Every connection an import took is committed and given back, not only the one its
	 * constructor borrowed: what a connection left behind holds is a transaction of the import and
	 * a permit of the pool, and a pool is never removed from the map - so a permit lost to an
	 * import is lost for the life of the server (#878).
	 */
	@Test(timeOut = 120000)
	public void testEveryConnectionOfAnImportIsCommittedAndReturned() throws Exception {
		System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "3");
		final String url = StubDriver.PREFIX + "importer-returned";
		final JDBCStorage storage = importingStorage(url);
		final Importer importer = storage.startImport();
		try {
			importer.put(ID2ENTRY, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
			importer.put(DN2ID, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
			importer.put(STATE, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));

			importer.close();

			assertEquals(used.size(), 3, "the three trees of the import did not get a connection each");
			for (final Connection con : connectionsUsed()) {
				verify(con, times(1)).commit();
			}
			assertEquals(CachedConnection.poolOf(url).idleCount(), 3, "an import kept a connection of the pool");
		} finally {
			storage.close();
		}
		assertEquals(CachedConnection.poolOf(url).meteredCount(), 0, "an import kept a permit of the pool");
	}

	/**
	 * A clear commits every connection of the import, not only the one whose tree it clears, and it
	 * does so before it empties anything. The one connection an import used to hold made that so of
	 * its own accord - {@code clearTree()} ends in a commit, and that commit made durable every
	 * write the import had made so far. What that is worth to a {@code rebuild-index} is the
	 * {@code setTrust(false)} of {@code RebuildIndexStrategy.beforePhaseOne}, which is written just
	 * in front of the clear of the tree it describes: left to {@code close()} that flag would still
	 * be uncommitted while the table was emptied and committed, and a server that stopped in
	 * between would come back to an index that is empty and marked trusted.
	 */
	@Test(timeOut = 120000)
	public void testAClearCommitsTheOtherConnectionsOfTheImport() throws Exception {
		System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "2");
		final String url = StubDriver.PREFIX + "importer-clear-commits";
		final JDBCStorage storage = importingStorage(url);
		final Importer importer = storage.startImport();
		try {
			// the state tree stands for the trust flag: written, and left uncommitted until
			// something commits the connection it went to
			importer.put(STATE, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
			importer.clearTree(ID2ENTRY);

			// asserted in order rather than counted: the destructive half of a clear is the one
			// thing that must not be durable while a write it is meant to invalidate is not, so the
			// commit of the peer belongs in front of the one the clear itself ends in - a count of
			// two is the same whichever way round they happened
			assertEquals(commits, Arrays.asList(usedInOrder.get(0), usedInOrder.get(1)),
				"the clear did not commit the other connection of the import before it emptied its own tree");
		} finally {
			importer.close();
			storage.close();
		}
	}

	/**
	 * ... including a peer whose first write is still in flight, which is not a connection with
	 * nothing to commit but one whose write has not come back yet.
	 * <p>
	 * An import counts a write once its statement returns, so a connection written for the first
	 * time has nothing recorded against it for exactly as long as that statement takes. That window
	 * is reached by more than one thread at once: {@code beforePhaseOne} runs on the import threads
	 * - {@code OnDiskMergeImporter.processEntry} guards it with a latch per container, not one for
	 * all of them - so on an {@code import-ldif} of two base DNs the {@code setTrust(false)} of one
	 * container is written while the clears of the next are running. The one connection an import
	 * held before #891 carried such a write into the commit of the clear, being the same
	 * transaction and issued in front of it in the driver, so a clear that passed over it here
	 * would be a commit point the single connection did not have.
	 */
	@Test(timeOut = 120000)
	public void testAClearCommitsAPeerWhoseWriteIsStillInFlight() throws Exception {
		System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "2");
		final String url = StubDriver.PREFIX + "importer-clear-commits-in-flight";
		final JDBCStorage storage = importingStorage(url);
		final Importer importer = storage.startImport();
		try {
			importing(importer, () -> {
				insideAStatementNow = new CountDownLatch(1);
				letTheStatementFinish = new CountDownLatch(1);
				// the state tree stands for the trust flag of beforePhaseOne: written on a thread of
				// the import, and held inside the statement it issues
				final AtomicReference<Throwable> flagFailure = new AtomicReference<>();
				final Thread writing = daemon(() -> {
					try {
						importer.put(STATE, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
					} catch (Throwable t) {
						flagFailure.set(t);
					}
				}, "import-flag");
				// the clear of the next container, on a connection of its own
				final AtomicReference<Throwable> clearFailure = new AtomicReference<>();
				final Thread clearing = daemon(() -> {
					try {
						importer.clearTree(ID2ENTRY);
					} catch (Throwable t) {
						clearFailure.set(t);
					}
				}, "import-clear");
				try {
					writing.start();
					assertTrue(insideAStatementNow.await(RENDEZVOUS_SECONDS, TimeUnit.SECONDS),
						"the write of the trust flag never got inside a statement");
					// where the clear gets to on its own: waiting on the monitor of the peer, which
					// is the commit point being paid for - a clear that decided the peer had nothing
					// to commit runs through to the end instead
					clearing.start();
					untilBlockedOrDone(clearing);
				} finally {
					letTheStatementFinish.countDown();
				}
				writing.join(TimeUnit.SECONDS.toMillis(RENDEZVOUS_SECONDS));
				clearing.join(TimeUnit.SECONDS.toMillis(RENDEZVOUS_SECONDS));
				reportFailureOf("the write of the trust flag", flagFailure);
				reportFailureOf("the clear", clearFailure);
				assertFalse(writing.isAlive(), "the write of the trust flag did not finish");
				assertFalse(clearing.isAlive(), "the clear did not finish");

				assertEquals(usedInOrder.size(), 2, "the two trees of this import did not get a connection each");
				assertEquals(commits, Arrays.asList(usedInOrder.get(0), usedInOrder.get(1)),
					"the clear passed over a peer whose write was still in flight");
			});
		} finally {
			storage.close();
		}
	}

	/**
	 * The connection of the last write is committed last. The connections of an import are
	 * transactions of their own, so {@code close()} cannot commit them as one - and the last thing
	 * an import writes is the flag that says the rest of it is good ({@code setTrust(true)} of
	 * {@code afterPhaseTwo}). Committed last, that flag is rolled back with the connection it is on
	 * whenever an earlier commit fails, so a failed import cannot leave an index marked trusted
	 * over data that never got there.
	 */
	@Test(timeOut = 120000)
	public void testTheConnectionsAreCommittedInTheOrderTheyWereLastWrittenTo() throws Exception {
		System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "3");
		final String url = StubDriver.PREFIX + "importer-last-write";
		final JDBCStorage storage = importingStorage(url);
		final Importer importer = storage.startImport();
		try {
			importer.put(ID2ENTRY, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
			importer.put(DN2ID, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
			importer.put(STATE, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
			// the two writes that stand for the trust flags afterPhaseTwo writes, one per base DN:
			// both have to be committed after the connection that holds nothing but data
			importer.put(ID2ENTRY, ByteString.valueOfUtf8("k2"), ByteString.valueOfUtf8("v"));
			importer.put(DN2ID, ByteString.valueOfUtf8("k2"), ByteString.valueOfUtf8("v"));

			importer.close();

			assertEquals(usedInOrder.size(), 3, "the three trees of the import did not get a connection each");
			assertEquals(commits, Arrays.asList(usedInOrder.get(2), usedInOrder.get(0), usedInOrder.get(1)),
				"the connections were not committed in the order they were last written to");
		} finally {
			storage.close();
		}
	}

	/**
	 * {@code close()} waits for a statement of a straggler to finish rather than committing the
	 * connection beside it.
	 * <p>
	 * Phase two gives its threads five seconds to answer an interrupt and closes the importer
	 * whether they answered or not ({@code OnDiskMergeImporter.invokeParallel}), so a write of an
	 * import can still be in flight when its connections are committed and handed back. A commit
	 * issued beside that statement is two threads on one connection - the whole of #891, with the
	 * closing thread on one side of it. Every other test of this suite joins its writers before it
	 * closes, so this is the only one where anything of a close happens beside a statement, and it
	 * is what the other half of the detector is for: the one that watches a commit, a rollback or a
	 * return land while a thread is inside a statement.
	 * <p>
	 * The straggler is let go once the close has got as far as it can on its own, and its write is
	 * then committed rather than lost: a thread that was inside the import when {@code close()}
	 * began is part of it. That is the assertion which bites - a close that reads the state of a
	 * connection without waiting for the statement in flight on it finds nothing written and
	 * commits nothing, and the write of the straggler is rolled back by the return instead.
	 */
	@Test(timeOut = 120000)
	public void testACloseWaitsForAStatementOfAStragglerRatherThanCommittingBesideIt() throws Exception {
		System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "2");
		final String url = StubDriver.PREFIX + "importer-close-waits";
		final JDBCStorage storage = importingStorage(url);
		final Importer importer = storage.startImport();
		try {
			importing(importer, () -> {
				insideAStatementNow = new CountDownLatch(1);
				letTheStatementFinish = new CountDownLatch(1);
				final AtomicReference<Throwable> stragglerFailure = new AtomicReference<>();
				final Thread straggling = daemon(() -> {
					try {
						importer.put(ID2ENTRY, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
					} catch (Throwable t) {
						stragglerFailure.set(t);
					}
				}, "import-straggler");
				final AtomicReference<Throwable> closeFailure = new AtomicReference<>();
				final Thread closing = daemon(() -> {
					try {
						importer.close();
					} catch (Throwable t) {
						closeFailure.set(t);
					}
				}, "import-close-beside-a-statement");
				try {
					straggling.start();
					assertTrue(insideAStatementNow.await(RENDEZVOUS_SECONDS, TimeUnit.SECONDS),
						"the write of the straggler never got inside a statement");
					// where the close gets to on its own: waiting on the monitor of the connection
					// the straggler is writing through
					closing.start();
					untilBlockedOrDone(closing);
				} finally {
					letTheStatementFinish.countDown();
				}
				straggling.join(TimeUnit.SECONDS.toMillis(RENDEZVOUS_SECONDS));
				closing.join(TimeUnit.SECONDS.toMillis(CLOSE_SECONDS));
				reportFailureOf("the write of the straggler", stragglerFailure);
				reportFailureOf("the close of the import", closeFailure);
				assertFalse(straggling.isAlive(), "the write of the straggler did not finish");
				assertFalse(closing.isAlive(), "the import did not close");

				assertEquals(shared, Collections.emptyList(),
					"the close touched a connection while a thread of the import was inside a statement of it");
				assertEquals(usedInOrder.size(), 1, "the write of the straggler did not reach a connection");
				verify(usedInOrder.get(0), times(1)).commit();
			});
		} finally {
			storage.close();
		}
		assertEquals(CachedConnection.poolOf(url).meteredCount(), 0, "the import kept a permit of the pool");
	}

	/**
	 * A write that arrives after {@code close()} takes no connection. Phase two gives its threads
	 * five seconds to answer an interrupt and then closes the importer whether they did or not, so
	 * a write can reach it once its connections are back in the pool - and one that borrowed there
	 * would take a permit nothing is left to give back, out of a pool that is never removed from
	 * the map (#878).
	 */
	@Test(timeOut = 120000)
	public void testAWriteAfterCloseTakesNoConnection() throws Exception {
		System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "4");
		final String url = StubDriver.PREFIX + "importer-after-close";
		final JDBCStorage storage = importingStorage(url);
		final Importer importer = storage.startImport();
		try {
			importer.put(ID2ENTRY, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
			importer.close();

			try {
				importer.put(DN2ID, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
				fail("a write that arrives after the import is closed must be refused");
			} catch (StorageRuntimeException expected) {
				// the import is over: there is no transaction left for this write to belong to
			}
			try {
				// the same for a clear, which commits the other connections of the import before it
				// empties a tree: those are back in the pool, serving whoever borrowed them next
				importer.clearTree(ID2ENTRY);
				fail("a clear that arrives after the import is closed must be refused");
			} catch (StorageRuntimeException expected) {
				// as above
			}

			assertEquals(used.size(), 1, "a write after close() took a connection of the pool");
			assertEquals(CachedConnection.poolOf(url).idleCount(), 1, "a write after close() kept a connection");
		} finally {
			storage.close();
		}
		assertEquals(CachedConnection.poolOf(url).meteredCount(), 0, "a write after close() kept a permit");
	}

	/**
	 * A commit that fails still gives every connection of the import back, and takes the ones after
	 * it down with it. What a connection left behind holds is a permit of a pool that is never
	 * removed from the map, so a permit lost to a failed import is lost for the life of the server
	 * (#878) - and what a connection committed after the failure of an earlier one would hold is
	 * the trust flags of {@code afterPhaseTwo}, which {@code close()} commits last for exactly this
	 * reason: a failed import must not leave an index marked trusted over data that never got there.
	 */
	@Test(timeOut = 120000)
	public void testAFailedCommitStillReturnsEveryConnection() throws Exception {
		System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "3");
		final String url = StubDriver.PREFIX + "importer-failed-commit";
		final JDBCStorage storage = importingStorage(url);
		final Importer importer = storage.startImport();
		try {
			importer.put(ID2ENTRY, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
			importer.put(DN2ID, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
			importer.put(STATE, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
			doThrow(new SQLException("the socket went away")).when(usedInOrder.get(1)).commit();

			try {
				importer.close();
				fail("the failure of the commit was not reported");
			} catch (StorageRuntimeException expected) {
				assertEquals(expected.getCause().getMessage(), "the socket went away");
			}

			// the connections are committed in the order they were last written to, so this one
			// stands where the trust flags of afterPhaseTwo stand: behind the connection that just
			// failed, and rolled back by the return rather than committed on top of a failed import
			verify(usedInOrder.get(2), never()).commit();
			assertEquals(CachedConnection.poolOf(url).idleCount(), 3, "a failed import kept a connection of the pool");
		} finally {
			storage.close();
		}
		assertEquals(CachedConnection.poolOf(url).meteredCount(), 0, "a failed import kept a permit of the pool");
	}

	/**
	 * A tree that cannot be given a connection of its own is written through one the import already
	 * holds, rather than failing the import. The connections are taken as the trees are first
	 * touched, so a pool at its bound would otherwise stop an import halfway through - with the
	 * clears of {@code beforePhaseOne} already committed - where the one connection an import held
	 * before #891 would have carried it to the end.
	 */
	@Test(timeOut = 120000)
	public void testATreeSharesAConnectionWhenThePoolHasNoneToSpare() throws Exception {
		System.setProperty(CachedConnection.POOL_MAX_PROPERTY, "2");
		System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "1"); // nothing is going to be returned
		System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "2");
		final String url = StubDriver.PREFIX + "importer-pool-full";
		final JDBCStorage storage = importingStorage(url);
		try {
			final Importer importer = storage.startImport(); // one of the two connections of the pool
			// and the other one to an operation of the server, so that the pool has none to spare
			final Connection heldByAnOperation = CachedConnection.getConnection(url);
			try {
				importing(importer, () -> {
					importer.put(ID2ENTRY, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
					// on a thread of its own: a thread that already holds a connection of the pool is
					// exempt from waiting at its bound, and every thread of phase two starts holding
					// none
					putOnAThreadOfItsOwn(importer, DN2ID);

					assertEquals(used.size(), 1, "the tree the pool had no connection for did not share one");
				});
			} finally {
				heldByAnOperation.close();
			}
		} finally {
			storage.close();
		}
	}

	/**
	 * A second {@code close()} touches nothing: the connections of the import are back in the pool
	 * by then, and the pool hands them out again - so a close that walked its map a second time
	 * would roll back and re-pool a connection another borrower is holding.
	 */
	@Test(timeOut = 120000)
	public void testASecondCloseTouchesNothingTheImportGaveBack() throws Exception {
		System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "2");
		final String url = StubDriver.PREFIX + "importer-second-close";
		final JDBCStorage storage = importingStorage(url);
		try {
			final Importer importer = storage.startImport();
			importer.put(ID2ENTRY, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
			importer.put(DN2ID, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
			importer.close();

			// the pool hands one of them straight back out, the way an LDAP operation would
			final Connection borrowedAgain = CachedConnection.getConnection(url);
			try {
				importer.close();
				assertEquals(CachedConnection.poolOf(url).idleCount(), 1,
					"a second close() returned a connection the pool had already handed to somebody else");
			} finally {
				borrowedAgain.close();
			}
		} finally {
			storage.close();
		}
		assertEquals(CachedConnection.poolOf(url).meteredCount(), 0, "a connection of the import was lost");
	}

	/**
	 * A tree the database refuses a connection for is written through one the import already holds,
	 * the way a tree the pool has nothing to spare for is (#1013).
	 * <p>
	 * The two ends of a borrow that can run out are not reported alike: the bound of the pool is
	 * answered with a {@code SQLTimeoutException}, while a database at a limit of its own is
	 * answered with whatever its driver says - and a code {@code CachedConnection.isWorthRetrying}
	 * does not recognize is raised at once rather than waited out. A mysql account with a
	 * {@code MAX_USER_CONNECTIONS} of its own answers 1226 on SQLState 42000, which is such a code
	 * (#1011). An import that failed on it would stop halfway through, with the clears of
	 * {@code beforePhaseOne} already committed, while every connection it holds still works.
	 */
	@Test(timeOut = 120000)
	public void testATreeSharesAConnectionWhenTheDatabaseRefusesOne() throws Exception {
		System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "2");
		final String url = StubDriver.PREFIX + "importer-refused";
		final JDBCStorage storage = importingStorage(url);
		final Importer importer = storage.startImport();
		try {
			importing(importer, () -> {
				// the first tree takes the connection the constructor borrowed, so that the second is
				// one the import has to go to the database for
				importer.put(ID2ENTRY, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
				// armed only now: what this is about is the borrows an import makes as it goes, not
				// the one it is started with - an import of a database that takes no connection at
				// all is refused where it is started
				refuseFurtherConnects("User 'opendj' has exceeded the 'max_user_connections' resource"
					+ " (current value: 1)", "42000", 1226);
				// on a thread of its own, the way every thread of phase two starts: one that already
				// holds a connection of the pool takes another path through the borrow
				putOnAThreadOfItsOwn(importer, DN2ID);

				assertEquals(used.size(), 1, "the tree the database refused a connection for did not share one");
			});
		} finally {
			storage.close();
		}
		assertEquals(CachedConnection.poolOf(url).meteredCount(), 0, "a refused borrow kept a permit of the pool");
	}

	/**
	 * ... and an import whose borrow is interrupted fails instead, keeping the interrupt.
	 * <p>
	 * The connection this import cannot have is the loss of a parallelism it asked for, and that is
	 * what sharing is the answer to. An interrupt is not that: phase two interrupts the threads of
	 * an import to stop it ({@code OnDiskMergeImporter.invokeParallel} gives them five seconds to
	 * answer), and a thread that answered by writing the tree through another connection would be
	 * carrying on with the work it was told to drop. The interrupt goes back on the thread as well:
	 * a borrow that throws {@code InterruptedException} has cleared it, and what reads it next is
	 * the executor of phase two.
	 */
	@Test(timeOut = 120000)
	public void testAnInterruptedBorrowFailsTheImportRatherThanSharingAConnection() throws Exception {
		System.setProperty(JDBCStorage.IMPORT_CONNECTIONS_PROPERTY, "2");
		final String url = StubDriver.PREFIX + "importer-interrupted";
		final JDBCStorage storage = importingStorage(url);
		final Importer importer = storage.startImport();
		try {
			importing(importer, () -> {
				importer.put(ID2ENTRY, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));

				final AtomicReference<Throwable> failure = new AtomicReference<>();
				final AtomicBoolean interruptKept = new AtomicBoolean();
				final Thread writer = daemon(() -> {
					// set before the write rather than raced with it: the pool waits for an idle
					// connection interruptibly, so a thread that carries an interrupt into the borrow
					// gets the same InterruptedException as one interrupted while parked there
					Thread.currentThread().interrupt();
					try {
						importer.put(DN2ID, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
					} catch (Throwable t) {
						failure.set(t);
					} finally {
						interruptKept.set(Thread.currentThread().isInterrupted());
					}
				}, "import-interrupted");
				writer.start();
				writer.join(TimeUnit.SECONDS.toMillis(RENDEZVOUS_SECONDS));
				assertFalse(writer.isAlive(), "the interrupted write did not finish");

				final Throwable reported = failure.get();
				assertTrue(reported instanceof StorageRuntimeException,
					"an interrupted borrow was not reported as a failure of the import: " + reported);
				assertTrue(reported.getCause() instanceof InterruptedException,
					"an interrupted borrow was reported as something else: " + reported.getCause());
				assertTrue(interruptKept.get(), "the borrow swallowed the interrupt of an import thread");
			});
		} finally {
			storage.close();
		}
	}

	/** A storage open for writing over the driver of this test, which is what an import needs. */
	private JDBCStorage importingStorage(String url) throws Exception {
		// mockCfg rather than a bare mock: it answers every getter with the value declared in
		// JDBCBackendConfiguration.xml, so a storage that comes to read a setting this test never
		// thought of gets the default instead of a null - the way the sibling suites of this package
		// build their configurations
		final JDBCBackendCfg cfg = mockCfg(JDBCBackendCfg.class);
		when(cfg.getDBDirectory()).thenReturn(url);
		final JDBCStorage storage = new JDBCStorage(cfg, null);
		storage.open(AccessMode.READ_WRITE);
		// the connection the open borrowed and gave back is not one of the import's: only what a
		// statement was issued on is recorded, and the open issues none
		forgetTheConnections();
		return storage;
	}

	/**
	 * Makes the driver of this test answer every further connect with a failure of the given shape,
	 * the way a database at a limit of its own does.
	 */
	private void refuseFurtherConnects(final String message, final String sqlState, final int vendorCode) {
		refusal.set(() -> new SQLException(message, sqlState, vendorCode));
	}

	private Future<?> put(final Importer importer, final TreeName treeName) {
		return threads.submit(() -> {
			toTheStartingLine();
			importer.put(treeName, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
		});
	}

	/** A write of a thread that has written nothing before, the way every thread of an import starts. */
	private void putOnAThreadOfItsOwn(final Importer importer, final TreeName treeName) throws Exception {
		final AtomicReference<Throwable> failure = new AtomicReference<>();
		final Thread thread = daemon(() -> {
			try {
				importer.put(treeName, ByteString.valueOfUtf8("k"), ByteString.valueOfUtf8("v"));
			} catch (Throwable t) {
				failure.set(t);
			}
		}, "import-" + treeName.getIndexId());
		thread.start();
		thread.join(TimeUnit.SECONDS.toMillis(RENDEZVOUS_SECONDS));
		if (failure.get() != null) {
			throw new IllegalStateException("the write of " + treeName + " failed", failure.get());
		}
		// asserted rather than left to the join: a write still blocked - on the monitor of a
		// connection, or on a borrow - would otherwise leave the assertions of the test standing on
		// a run that never made the state they are about
		assertFalse(thread.isAlive(), "the write of " + treeName + " did not finish");
	}

	/** Raises what a thread of this test caught, under a name that says which thread it was. */
	private static void reportFailureOf(String what, AtomicReference<Throwable> failure) {
		final Throwable caught = failure.get();
		if (caught != null) {
			throw new IllegalStateException(what + " failed", caught);
		}
	}

	/**
	 * A thread of this suite. Daemon, every one of them: a write left standing on the monitor of a
	 * connection by a failing run must not keep the JVM of the build alive once the test that
	 * started it has been reported.
	 */
	private static Thread daemon(Runnable body, String name) {
		final Thread thread = new Thread(body, name);
		thread.setDaemon(true);
		return thread;
	}

	/** The threads a test writes through, so that a stuck one is not a build that never ends. */
	private static ExecutorService daemonThreads(int count, final String name) {
		final AtomicInteger numbered = new AtomicInteger();
		return Executors.newFixedThreadPool(count,
			runnable -> daemon(runnable, name + "-" + numbered.incrementAndGet()));
	}

	/** The body of a test that writes through an import - see {@link #importing(Importer, ImportBody)}. */
	private interface ImportBody {
		void run() throws Exception;
	}

	/**
	 * Runs the body of a test against an import and closes that import afterwards, whatever the
	 * body did - without the close having the last word on what went wrong.
	 * <p>
	 * The close used to stand in a {@code finally} of the test, and it is here instead because the
	 * two things that can go wrong there are not disjoint: {@code close()} commits and returns every
	 * connection under the monitor of that connection, so a write of this test left standing on one
	 * is exactly what makes the close wait - and a close that gave up waiting and threw out of that
	 * {@code finally} replaced the failure the test came for with one naming the monitor. Here the
	 * body has the first say and the trouble of the close rides along with it as suppressed.
	 */
	private void importing(Importer importer, ImportBody body) throws Exception {
		Throwable found = null;
		try {
			body.run();
		} catch (Throwable t) {
			found = t;
		}
		final Throwable reported = closeWithoutHanging(importer, found);
		if (reported instanceof Error) {
			throw (Error) reported;
		}
		if (reported instanceof Exception) {
			throw (Exception) reported;
		}
		if (reported != null) {
			throw new IllegalStateException("the import failed to close", reported);
		}
	}

	/**
	 * Closes an import without waiting for a straggling thread of the test forever, and returns
	 * what is to be reported: what the body of the test found where it found something, with the
	 * trouble of the close - a close that never finished, or one that failed - suppressed into it.
	 */
	private Throwable closeWithoutHanging(Importer importer, Throwable found) {
		final AtomicReference<Throwable> failure = new AtomicReference<>();
		final Thread closing = daemon(() -> {
			try {
				importer.close();
			} catch (Throwable t) {
				failure.set(t);
			}
		}, "import-close");
		closing.start();
		Throwable trouble;
		try {
			closing.join(TimeUnit.SECONDS.toMillis(CLOSE_SECONDS));
			trouble = closing.isAlive()
				? new IllegalStateException("the import did not close within " + CLOSE_SECONDS
					+ "s: a thread of this test is holding the monitor of one of its connections")
				: failure.get();
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
			trouble = new IllegalStateException("interrupted while closing the import", e);
		}
		if (found == null) {
			return trouble;
		}
		if (trouble != null) {
			found.addSuppressed(trouble);
		}
		return found;
	}

	private List<Connection> connectionsUsed() {
		synchronized (used) {
			return new ArrayList<>(used);
		}
	}

	/**
	 * A connection of this test: it answers every statement with one mock of its own and records
	 * the thread that asked for it while that thread is inside the call.
	 */
	private Connection newConnection() throws SQLException {
		final Connection con = org.mockito.Mockito.mock(Connection.class);
		final PreparedStatement statement = org.mockito.Mockito.mock(PreparedStatement.class);
		when(con.isValid(anyInt())).thenReturn(true);
		// the statement names the connection it runs on, as CachedConnection.prepareStatement() has
		// it: the bound of a statement is arbitrated on the connection, which is read off the statement
		when(statement.getConnection()).thenReturn(con);
		// a thread is inside a statement of a connection for longer than the prepareStatement that
		// hands it one: what the sql server driver walks past the end of its array is the list of
		// listeners that the execution and the close of a statement mutate as readily (#891), so
		// those are recorded here too - the window this suite watches is the whole of a statement
		doAnswer(invocation -> {
			enterAStatement(con);
			try {
				return 0; // what a mock of this returns anyway: no row of this test is counted
			} finally {
				leaveAStatement(con);
			}
		}).when(statement).executeUpdate();
		doAnswer(invocation -> {
			enterAStatement(con);
			try {
				return Boolean.FALSE; // as above: nothing of this test reads a result set
			} finally {
				leaveAStatement(con);
			}
		}).when(statement).execute();
		doAnswer(invocation -> {
			enterAStatement(con);
			try {
				return null;
			} finally {
				leaveAStatement(con);
			}
		}).when(statement).close();
		// the commits of the import in the order they happen: which connection is committed when is
		// what keeps a failed import from leaving an index marked trusted over data that is not there
		doAnswer(invocation -> {
			recordAThreadInsideAStatement(con, "commit");
			commits.add(con);
			return null;
		}).when(con).commit();
		// the return of a connection rolls it back and hands it to the next borrower, so neither may
		// happen while a statement of an import thread is in flight on it
		doAnswer(invocation -> {
			recordAThreadInsideAStatement(con, "rollback");
			return null;
		}).when(con).rollback();
		doAnswer(invocation -> {
			recordAThreadInsideAStatement(con, "close");
			return null;
		}).when(con).close();
		when(con.prepareStatement(anyString())).thenAnswer(invocation -> {
			if (used.add(con)) {
				usedInOrder.add(con);
			}
			enterAStatement(con);
			try {
				// the hold outside the monitor of the map above, and after this thread has entered
				// it: what it is waiting for is a peer thread getting in here and recording itself
				parkInsideTheStatement();
				holdTheConnection();
				meetPeers();
				return statement;
			} finally {
				leaveAStatement(con);
			}
		});
		return con;
	}

	/** Records this thread as inside a statement of the given connection, with whoever is already there. */
	private void enterAStatement(Connection con) {
		synchronized (inStatement) {
			final Set<Thread> inside = inStatement.get(con);
			final Set<Thread> threadsInside = inside != null ? inside : new LinkedHashSet<Thread>();
			for (final Thread peer : threadsInside) {
				if (peer != Thread.currentThread()) {
					shared.add(peer.getName() + " and " + Thread.currentThread().getName());
				}
			}
			threadsInside.add(Thread.currentThread());
			inStatement.put(con, threadsInside);
		}
	}

	private void leaveAStatement(Connection con) {
		synchronized (inStatement) {
			final Set<Thread> threadsInside = inStatement.get(con);
			if (threadsInside != null && threadsInside.remove(Thread.currentThread()) && threadsInside.isEmpty()) {
				inStatement.remove(con);
			}
		}
	}

	/** Records a connection touched from one thread while another is inside a statement on it. */
	private void recordAThreadInsideAStatement(Connection con, String what) {
		synchronized (inStatement) {
			final Set<Thread> threadsInside = inStatement.get(con);
			if (threadsInside == null) {
				return;
			}
			for (final Thread inside : threadsInside) {
				if (inside != Thread.currentThread()) {
					shared.add(what + " of " + Thread.currentThread().getName() + " while " + inside.getName()
						+ " was inside a statement");
				}
			}
		}
	}

	/**
	 * Parks a thread inside the first statement it issues and tells the test it is in there, until
	 * the test lets it go.
	 * <p>
	 * A write in flight, as the rest of the import sees one: the connection holds a statement that
	 * has not come back, so nothing is recorded against it yet - {@code put()} counts a write once
	 * the statement returns - and its monitor is held for as long as this lasts. What a test does
	 * from there is ask another thread of the import a question about that connection.
	 */
	private void parkInsideTheStatement() {
		final CountDownLatch letGo = letTheStatementFinish;
		if (letGo == null || parked.get()) {
			return;
		}
		parked.set(Boolean.TRUE);
		insideAStatementNow.countDown();
		try {
			letGo.await(PARK_SECONDS, TimeUnit.SECONDS);
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
		}
	}

	/**
	 * Waits until a thread of this test has got as far as it is going to get on its own: either it
	 * is blocked on a monitor - the thing this suite is about - or it is done.
	 * <p>
	 * Waited for rather than slept past: what the tests below ask is whether a thread stops at a
	 * monitor or walks through it, and a fixed pause would answer that with the load of the machine
	 * on a slow run.
	 */
	private static void untilBlockedOrDone(Thread thread) throws InterruptedException {
		final long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(RENDEZVOUS_SECONDS);
		while (System.currentTimeMillis() < deadline) {
			final Thread.State state = thread.getState();
			if (state == Thread.State.BLOCKED || state == Thread.State.TERMINATED) {
				return;
			}
			Thread.sleep(10);
		}
	}

	/**
	 * Holds the first statement of a thread on the connection it is on, until a peer thread has one
	 * of its own in flight there or the hold runs out.
	 * <p>
	 * How a test asks whether two threads can be inside a statement of one connection at the same
	 * time: they cannot be brought together at a barrier the way the threads of two connections are
	 * - a second thread held on the monitor of the connection never reaches one - so the first
	 * thread in waits there instead, and the peer either turns up beside it, which is #891, or does
	 * not, which is the hold being paid in full.
	 */
	private void holdTheConnection() {
		final CountDownLatch inside = insideAStatement;
		if (inside == null || held.get()) {
			return;
		}
		held.set(Boolean.TRUE);
		inside.countDown();
		try {
			inside.await(HOLD_SECONDS, TimeUnit.SECONDS);
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
		}
	}

	/** Where the threads of a test start together, in front of the monitors of the importer. */
	private void toTheStartingLine() {
		final CyclicBarrier line = startingLine;
		if (line == null) {
			return;
		}
		try {
			line.await(RENDEZVOUS_SECONDS, TimeUnit.SECONDS);
		} catch (InterruptedException | TimeoutException | BrokenBarrierException e) {
			throw new IllegalStateException("the threads of this test did not reach the starting line", e);
		}
	}

	/**
	 * Holds the first statement of a thread until every peer of the rendezvous has one in flight
	 * too. Reported as a failure of the statement rather than waited out: a rendezvous nobody else
	 * reaches is an import that serialized its threads, which is what the parallel test is about.
	 */
	private void meetPeers() throws SQLException {
		final CyclicBarrier barrier = rendezvous;
		if (barrier == null || met.get()) {
			return;
		}
		met.set(Boolean.TRUE);
		try {
			barrier.await(RENDEZVOUS_SECONDS, TimeUnit.SECONDS);
		} catch (TimeoutException | BrokenBarrierException e) {
			throw new SQLException("no peer thread had a statement of its own in flight within "
				+ RENDEZVOUS_SECONDS + "s: the import did not let its threads write at the same time", e);
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
			throw new SQLException("interrupted while waiting for the peer threads of the import", e);
		}
	}

	/** A driver of this test, so that the connections of an import need no database behind them. */
	private final class StubDriver implements Driver {
		static final String PREFIX = "jdbc:opendj-import-stub:";

		@Override
		public Connection connect(String url, Properties info) throws SQLException {
			if (!acceptsURL(url)) {
				return null;
			}
			final Supplier<SQLException> refused = refusal.get();
			if (refused != null) {
				throw refused.get();
			}
			return newConnection();
		}

		@Override
		public boolean acceptsURL(String url) {
			return url != null && url.startsWith(PREFIX);
		}

		@Override
		public DriverPropertyInfo[] getPropertyInfo(String url, Properties info) {
			return new DriverPropertyInfo[0];
		}

		@Override
		public int getMajorVersion() {
			return 1;
		}

		@Override
		public int getMinorVersion() {
			return 0;
		}

		@Override
		public boolean jdbcCompliant() {
			return false;
		}

		@Override
		public Logger getParentLogger() {
			return Logger.getLogger(StubDriver.class.getName());
		}
	}
}