[#891] Give the trees of an import connections of their own (#940)

Valery Kharseko

7 hours ago a1b8537e1e0cd10e4d614dd9c6bdda541a614572

[#891] Give the trees of an import connections of their own (#940)

 opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java

@@ -359,16 +359,28 @@
    }

    /**
     * The moment a borrow of this length gives up, or {@link Long#MAX_VALUE} where it gives up
     * never - a property of 0, and a value so large that the milliseconds of it would overflow.
     * Whether a wait of this many seconds is one with no bound at all.
     * <p>
     * The sum is guarded and not only the product: a value under the clamp above but large enough
     * Two spellings of it, and one predicate for both so that they cannot drift: zero, which is how
     * {@value #POOL_TIMEOUT_PROPERTY} says "wait for as long as it takes", and a number so large
     * that the milliseconds it stands for do not fit in a {@code long} - a deadline computed from
     * one of those overflows into the past, which is the opposite of what it asked for.
     */
    static boolean isUnboundedWait(long seconds) {
        return seconds == 0 || seconds >= Long.MAX_VALUE / 1000;
    }

    /**
     * The moment a borrow of this length gives up, or {@link Long#MAX_VALUE} where it gives up
     * never - the two readings {@link #isUnboundedWait} names.
     * <p>
     * The sum is guarded and not only the product: a value that predicate lets through but large enough
     * that the moment it names is past the end of the epoch would wrap to a deadline already behind
     * us, and a borrow configured to wait practically forever would give up on its first retryable
     * failure - the opposite of what was asked for.
     */
    static long deadlineOf(long startedAt, long poolTimeoutSeconds) {
        if (poolTimeoutSeconds == 0 || poolTimeoutSeconds >= Long.MAX_VALUE / 1000) {
        if (isUnboundedWait(poolTimeoutSeconds)) {
            return Long.MAX_VALUE;
        }
        final long deadline = startedAt + poolTimeoutSeconds * 1000;
@@ -522,8 +534,10 @@
     * <p>
     * A lower bound than that is what is reported, not every way past it: the replay threads of
     * replication default to the same count again and borrow on top of the workers, and an import or
     * a rebuild borrows besides. So this names one difference the operator can act on rather than
     * standing for the whole demand on the pool.
     * a rebuild borrows besides - one connection per tree it writes at a time, up to the bound of
     * {@code JDBCStorage.IMPORT_CONNECTIONS_PROPERTY}, and it holds them for its whole duration
     * (#891). So this names one difference the operator can act on rather than standing for the
     * whole demand on the pool.
     * <p>
     * Nothing fails for the difference alone: the surplus waits for a connection to be returned,
     * which is what the bound is there for. But every one of those waits is paid on an operation,
@@ -809,8 +823,15 @@
     * Returns the value of a numeric system property, ignoring a value that is not a non-negative
     * number in favor of the default. The unit is the one the property is read in, so that the
     * value the message names is not mistaken for another.
     * <p>
     * Package private rather than private so that the bound on the connections of an import
     * ({@code JDBCStorage.IMPORT_CONNECTIONS_PROPERTY}) can be read through it too, the way the
     * bounds of the pool are: an operator who mistypes one of those is told rather than left with a
     * default. Not every number this backend takes from a property comes through here - the
     * statistics timeout and the fetch sizes of {@code JDBCStorage} are read with
     * {@code Integer.getInteger}, which replaces a value it cannot parse in silence.
     */
    private static long getNonNegativeProperty(String name, long defaultValue, String unit) {
    static long getNonNegativeProperty(String name, long defaultValue, String unit) {
        final String value = System.getProperty(name);
        if (value != null) {
            try {
@@ -1260,11 +1281,45 @@
     * them is one borrow of a cold path, where the round trip the window saves is worth nothing.
     */
    static Connection getConnection(String connectionString, boolean trusted) throws Exception {
        return getConnection(connectionString, trusted, 0);
    }

    /**
     * The wait a borrow that carries a bound of its own actually gets: the shorter of what the
     * deployment asked for and what the caller can afford, and the caller's where the deployment
     * asked for no bound at all.
     *
     * @param maxWaitSeconds 0 for a caller that has no bound of its own, which takes the wait of
     * the deployment whatever it is
     */
    static long boundedWait(long poolTimeoutSeconds, long maxWaitSeconds) {
        if (isUnboundedWait(maxWaitSeconds)) {
            return poolTimeoutSeconds;
        }
        return isUnboundedWait(poolTimeoutSeconds)
            ? maxWaitSeconds : Math.min(poolTimeoutSeconds, maxWaitSeconds);
    }

    /**
     * Borrows a connection, waiting at the bound of the pool no longer than the given number of
     * seconds however long {@value #POOL_TIMEOUT_PROPERTY} says to wait.
     *
     * @param maxWaitSeconds the longest this borrow may wait at the bound of the pool, or 0 to wait
     * as the property says. For a caller whose own connections are what the pool is full of - an
     * import holds one per tree it writes until it ends (#891) - a wait with no bound is a deadlock
     * rather than a queue: nothing is going to return the connection it is waiting for but itself.
     * A caller that names one has somewhere to go when it runs out, so the wait of the deployment
     * is capped rather than merely replaced where it is unbounded: an import that has to ask the
     * pool once per tree would otherwise pay a long {@value #POOL_TIMEOUT_PROPERTY} over again for
     * every tree the pool has nothing to spare for.
     */
    static Connection getConnection(String connectionString, boolean trusted, long maxWaitSeconds)
            throws Exception {
        final Pool pool = poolOf(connectionString);
        final ConnectDialect dialect = ConnectDialect.of(connectionString);
        reportUnknownDialect(connectionString, dialect);
        final long connectTimeoutSeconds = getConnectTimeoutSeconds();
        final long poolTimeoutSeconds = getPoolTimeoutSeconds();
        final long poolTimeoutSeconds = boundedWait(getPoolTimeoutSeconds(), maxWaitSeconds);
        final long ttlMillis = getCacheTtlMillis();
        final long startedAt = System.currentTimeMillis();
        final long deadline = deadlineOf(startedAt, poolTimeoutSeconds);

 opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java

@@ -43,9 +43,12 @@
import java.sql.*;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Predicate;

import static org.opends.server.backends.pluggable.spi.StorageUtils.addErrorMessage;
@@ -600,12 +603,15 @@
     * The socket read timeout of one connection, and the statements running on it. This second
     * layer of the bound is a property of the socket rather than of a statement, so it cannot be
     * armed and put back per statement wherever a connection carries more than one at a time: an
     * {@code ImporterImpl} holds a single connection for the whole of an import and writes to it
     * from every phase-one worker and every phase-two task, and there the first statement to finish
     * would take the backstop away from every statement still in flight - while a statement whose
     * class carries no bound at all would run under whatever value a concurrent one happened to
     * arm, dying at it with nothing to say which property cut it, since such a statement never
     * reaches {@link #timedOut}.
     * {@code ImporterImpl} held a single connection for the whole of an import and wrote to it from
     * every phase-one worker and every phase-two task until the trees of an import were given
     * connections of their own (#891), and there the first statement to finish would take the
     * backstop away from every statement still in flight - while a statement whose class carries no
     * bound at all would run under whatever value a concurrent one happened to arm, dying at it
     * with nothing to say which property cut it, since such a statement never reaches
     * {@link #timedOut}. The arbitration stays now that no path of this class puts two threads on
     * one connection: what it holds is a property of the socket, so a connection that carries two
     * statements again must not have either of them cut by the bound of the other.
     * <p>
     * So the value armed is the loosest of the bounds of the statements in flight, and a statement
     * with no bound of its own takes it off for as long as it runs: this backstop exists to end a
@@ -949,7 +955,19 @@
    // unregistered pool is drained the moment another backend that did register with it closes,
    // with this one still borrowing from it (issue #878).
    Connection getConnection(boolean trusted) throws Exception {
        return CachedConnection.getConnection(poolKey(), trusted);
        return getConnection(trusted, 0);
    }

    /**
     * The borrow every path of this class makes, and the seam a test stands in for the pool at.
     *
     * @param maxWaitSeconds the longest this borrow may wait at the bound of the pool, whatever the
     * deployment asked for - 0 to wait as it says. Only the connections an import takes after its
     * first pass a number here: they are held until the import ends, so a pool full of them has
     * nothing to return to the thread waiting for one (#891).
     */
    Connection getConnection(boolean trusted, long maxWaitSeconds) throws Exception {
        return CachedConnection.getConnection(poolKey(), trusted, maxWaitSeconds);
    }


@@ -4886,10 +4904,168 @@
        return true;
    }

    /**
     * How many connections one import may write through; unset for the default of
     * {@link #importConnections()}, and 0 or 1 for the single connection an import had before #891 -
     * serialized, and the least a database sees of one.
     */
    static final String IMPORT_CONNECTIONS_PROPERTY = "org.openidentityplatform.opendj.jdbc.import.connections";

    /**
     * How many connections one import writes through, which is how many of its threads can write
     * at the same time.
     * <p>
     * More than one because {@code Importer} is thread-safe by contract and a
     * {@code java.sql.Connection} is not: phase two of {@code OnDiskMergeImporter} runs a thread
     * per tree and phase one clears the trees of a container while another thread writes id2entry,
     * so a shared connection has two threads issuing statements on it at once. The drivers differ
     * in what they make of that - pgjdbc and Connector/J serialize the work of a connection behind
     * a lock of their own - but 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 walked it past the end of its
     * array (issue #891).
     * <p>
     * Bounded because an import holds what it takes for its whole duration, out of a pool that is
     * bounded since #878 and shared with every other backend on that database: a default backend
     * has trees enough - one per index, three per attribute index - for a connection per tree to
     * empty a default pool and leave the LDAP traffic of an online rebuild waiting at its bound.
     * <p>
     * Half the bound of the pool by default, and no more than half of what the default bound would
     * be: a deployment that turned the bound off ({@code pool.max=0}) asked for its operations not
     * to queue behind one another, not for an import to open a connection per tree.
     * <p>
     * A number the operator set is taken as given, up to the bound of the pool: an import-ldif of a
     * backend that is offline has no traffic of its own to leave room for, and whoever raises this
     * on a server that is answering is making that trade knowingly - the default is where the
     * caution belongs.
     * <p>
     * What this bounds is how many connections one import holds, which is not the same as staying
     * inside the bound of the pool: a thread that already holds one connection of a pool is exempt
     * from waiting at its bound (see {@code CachedConnection.Pool}), and every thread of an import
     * takes a connection per tree it touches - so the borrows after the first go unmetered where the
     * pool stands at its bound, and are destroyed rather than pooled when they come back. That
     * exemption is what keeps the threads of an import from waiting on each other: they hold their
     * connections until the import ends, so a pool full of them has nothing left to return, and a
     * bound they had to wait at would be a deadlock rather than a queue.
     */
    int importConnections() {
        final int poolMax=poolMax();
        final long byDefault=Math.max(1, Math.min(poolMax, CachedConnection.DEFAULT_POOL_MAX)/2);
        // read the way every other bound of this backend is: a value that is not a non-negative
        // number is reported to the operator rather than quietly replaced. The default is handed to
        // it rather than a zero, so that the number the warning names is the number the import goes
        // on to use.
        final long configured=CachedConnection.getNonNegativeProperty(IMPORT_CONNECTIONS_PROPERTY, byDefault, "connections");
        // clamped to the pool: connections past its bound are not there to be had, so a number above
        // it buys nothing and costs the borrow deadline of every tree over the bound
        // (CachedConnection.POOL_TIMEOUT_PROPERTY) before the fallback of connectionOf() takes over
        return (int) Math.max(1, Math.min(configured, poolMax));
    }

    /**
     * The bound of the pool this storage borrows from. A seam of its own, like
     * {@link #getConnection(boolean)}: a test that stands in for the pool must not have this reach
     * past it into the static registry, which would intern a pool for its connection string.
     */
    int poolMax() {
        return CachedConnection.poolOf(poolKey()).max();
    }

    final class ImporterImpl implements Importer {
        final Connection con;
        final ReadableTransactionImpl txr;
        final WriteableTransactionTransactionImpl txw;
        /**
         * One connection of an import, the two transactions over it and the monitor that keeps one
         * thread at a time on it. Every statement of an import is issued under that monitor: what
         * the threads of an import must not do is share a connection, and the trees of an import
         * outnumber the connections it may take.
         */
        final class ImportConnection {
            final Connection con;
            final ReadableTransactionImpl txr;
            final WriteableTransactionTransactionImpl txw;
            /**
             * When this connection was last written, taken from {@link ImporterImpl#writes}, and
             * zero while it has nothing to commit. Written and read under the monitor of this
             * object, like the statements it counts: read anywhere else it says what the connection
             * held rather than what it holds, because it is set once the write it stands for has
             * come back - and a write still in flight is exactly the one a commit point must not
             * pass over.
             */
            long lastWrite;

            ImportConnection(Connection con) {
                this.con=con;
                this.txr=new ReadableTransactionImpl(con, StatementBound.BULK);
                this.txw=new WriteableTransactionTransactionImpl(con, StatementBound.BULK);
                // the mode this import was started with rather than the one the storage carries now:
                // the transaction reads that mutable field as it is built, and these are built as the
                // trees of an import are first touched - so a storage reopened read-only under a
                // running import would have the connections it opened before that keep writing while
                // every one after it refused, halfway through and with the clears already committed
                this.txw.isReadOnly=false;
            }

            /** Called under this monitor by whoever writes through this connection. */
            void written() {
                lastWrite=writes.incrementAndGet();
            }
        }

        /** The connections this import has taken, by the index the trees are handed out against. */
        private final ConcurrentMap<Integer,ImportConnection> connections = new ConcurrentHashMap<>();
        /** The connection index each tree is written through: a tree keeps the one it was first given. */
        private final ConcurrentMap<TreeName,Integer> connectionOfTree = new ConcurrentHashMap<>();
        /** Handed out round-robin, so that the first trees of an import get connections of their own. */
        private final AtomicInteger nextConnection = new AtomicInteger();
        /** One per index, so that the connection of an index is borrowed once however many trees want it. */
        private final ConcurrentMap<Integer,Object> borrowing = new ConcurrentHashMap<>();
        /**
         * The indexes no connection could be borrowed for, whose trees write through the first
         * connection of this import instead. Remembered rather than asked again per tree: there are
         * at most {@link #maxConnections} of them, and a tree that lands on one would otherwise pay
         * the borrow deadline of the pool over again for the answer the tree before it already got.
         */
        private final Set<Integer> sharedIndexes = ConcurrentHashMap.newKeySet();
        /**
         * Read once rather than per statement: this is a property of the whole import, and reading
         * it per record would be a property lookup per entry of an import-ldif.
         */
        final int maxConnections;

        /**
         * The connection the constructor borrows. Also the one a borrow that cannot be made falls
         * back to, so it is the one connection of an import that is always there.
         */
        private static final int FIRST_CONNECTION = 0;

        /**
         * Set by {@code close()} before it commits, after which this import has no transaction left
         * for a write to belong to. 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 can arrive with the connections
         * of the import already committed and back in the pool.
         */
        private volatile boolean closed;

        /**
         * Numbers the writes of this import, so that {@code close()} can commit its connections in
         * the order they were last written to.
         * <p>
         * The connections of an import are transactions of their own, so {@code close()} cannot make
         * them durable as one - and the last thing an import writes is the flag that says the rest of
         * it is good: {@code afterPhaseTwo} sets the trust flag of every index of a container once
         * phase two has written them all, one write per base DN. Committed in that order, those flags
         * go last, and an earlier commit that fails takes them down with it: a failed import cannot
         * leave an index marked trusted over data that never got there.
         * <p>
         * That the flags are the last writes is the caller's doing rather than this class's:
         * {@code OnDiskMergeImporter} waits for every phase-two task before it runs
         * {@code afterPhaseTwo}, and nothing here refuses a write that arrives after the flags while
         * the importer is still open. An importer whose writes did not end there would order its
         * commits by what its last writes actually were, which is what this counter says and all it
         * says.
         */
        private final AtomicLong writes = new AtomicLong();

        // The trees this import wrote: close() refreshes the statistics of these and only these,
        // so rebuilding a single index does not gather statistics for the whole backend. A full
        // import legitimately covers every tree - AbstractTwoPhaseImportStrategy.beforePhaseOne
@@ -4943,10 +5119,21 @@
                if (!accessMode.isWriteable()) {
                    throw new ReadOnlyStorageException();
                }
                // Inside the try like the refusal above: the pool this asks about is the one the open
                // just registered with, so a failure here has the storage this constructor opened to
                // give back as well.
                maxConnections=importConnections();
                // What an import takes out of the pool is worth reading when a borrow of one fails at
                // the bound: the pool names the property that bounds it, and this names the one that
                // bounds the demand.
                logger.debug(LocalizableMessage.raw("jdbc: import writes through up to %d connections (%s)",
                    maxConnections, IMPORT_CONNECTIONS_PROPERTY));
                borrowed=getValidatedConnection();
                txr =new ReadableTransactionImpl(borrowed, StatementBound.BULK);
                txw =new WriteableTransactionTransactionImpl(borrowed, StatementBound.BULK);
                con = borrowed;
                // The first connection is taken here rather than on the first tree, as it was before
                // the connections of an import became several (#891): an import of a database that
                // takes no connection is refused where it is started, and a storage this constructor
                // opened is given back by the catch below rather than by a put() far from it.
                connections.put(FIRST_CONNECTION, new ImportConnection(borrowed));
                borrowed=null;
            }catch (Throwable e){
                // Throwable rather than Exception, the way close() below catches it and for the same
@@ -4983,73 +5170,518 @@
        }

        /**
         * Hands the connection back to the pool and closes the sessions the transaction opened
         * beside it - the stamp one and the catalog one (#888), both outside the pool and neither
         * outliving the import that opened it - whatever went before. Returns the failure the
         * caller is to report: the return rolls back, and the rollback fails on exactly the
         * connection whose commit just did, so the commit stays the exception the caller sees and
         * this one rides along with it instead of replacing it.
         * The connection the given tree is written through, borrowed from the pool the first time
         * this import touches the tree.
         * <p>
         * Bound to the tree rather than to the thread, although it is the threads of an import that
         * must not share one: 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 - and an import
         * commits at {@code close()}, when every thread of it is long done. That is not a shape to
         * leave lying about: {@code setTrust()} writes the state tree of a container from
         * {@code beforePhaseOne} on an import thread and again from {@code afterPhaseTwo} on the
         * thread that closes the importer, and bound to the thread those two writes would be two
         * transactions waiting for each other with nothing left to break the wait - the bulk class
         * carries no bound, and the default lock wait is forever on three of the four engines.
         * Bound to the tree they are one transaction that waits for nothing, and phase two - which
         * runs a thread per tree - still gets the connection per thread this is all about.
         */
        private SQLException releaseConnection(SQLException failure) {
        ImportConnection connectionOf(TreeName treeName) {
            // The lookup before the assignment, not for want of a computeIfAbsent: this runs once per
            // record of an import-ldif, and the mapping function below allocates a capture of this
            // importer on every call however long the tree has had a connection.
            final Integer assigned=connectionOfTree.get(treeName);
            if (assigned!=null) {
                final ImportConnection open=connections.get(assigned);
                if (open!=null) {
                    return open;
                }
            }
            checkOpen();
            // floorMod rather than %: the counter is shared by every thread of the import and an
            // index of its own is all a tree needs, so it is never reset - and a negative index
            // would be one no connection is ever opened for
            final Integer index=connectionOfTree.computeIfAbsent(treeName,
                tree -> Math.floorMod(nextConnection.getAndIncrement(), maxConnections));
            final ImportConnection open=connections.get(index);
            if (open!=null) {
                return open;
            }
            final ImportConnection taken=sharedIndexes.contains(index) ? sharedConnection() : openConnection(index);
            if (connections.get(index)!=taken) {
                // this tree was given a connection of another index, the pool having none to spare:
                // pointed at it, every record of the tree takes the lookup at the top of this method
                // rather than this path, which allocates a capture of this importer per call
                connectionOfTree.put(treeName, FIRST_CONNECTION);
            }
            return taken;
        }

        /**
         * Takes the connection of an index, or the one this import already has where the pool has
         * none left to give.
         * <p>
         * The borrow is made outside the map rather than in a {@code computeIfAbsent}: it waits for a
         * connection to be returned where the pool stands at its bound - up to
         * {@code CachedConnection.POOL_TIMEOUT_PROPERTY} - and a wait of that length inside a mapping
         * function holds the bin of that key against every other index that hashes to it, which the
         * contract of {@code ConcurrentHashMap} says not to do. One borrow per index all the same:
         * two threads first touching trees of one index would otherwise each take a connection, and
         * the loser's would be a borrow of the pool made and given back for nothing.
         * <p>
         * A connection that does not end up in the map is given back here rather than left behind:
         * only its {@code close()} returns the permit it took, and a pool is never removed from the
         * map, so one lost here would be lost for the life of the server (#878).
         */
        private ImportConnection openConnection(Integer index) {
            synchronized (borrowing.computeIfAbsent(index, i -> new Object())) {
                final ImportConnection opened=connections.get(index);
                if (opened!=null) {
                    return opened; // another thread of this import got here first
                }
                if (sharedIndexes.contains(index)) {
                    // ... and found the pool with nothing to spare: this thread has the same answer
                    // waiting for it, and would pay the borrow deadline of the pool again to get it
                    return sharedConnection();
                }
                return openConnectionOnce(index);
            }
        }

        private ImportConnection openConnectionOnce(Integer index) {
            final Connection borrowed=borrowedOrShared(index);
            if (borrowed==null) { // shared, see borrowedOrShared()
                return sharedConnection();
            }
            final ImportConnection built;
            try {
                con.close();
            } catch (Throwable e) {
                // Throwable rather than SQLException: this close() is the return to the pool, whose
                // rollback a driver is free to fail unchecked. Reported rather than thrown, since a
                // throw out of here would leave with the failure the caller actually came for - the
                // commit above, and in the Throwable branch of close() the Error that branch exists
                // to preserve - dropped on the floor (issue #878).
                final SQLException reported=e instanceof SQLException ? (SQLException) e
                    : new SQLException("the connection of the import could not be returned to the pool", e);
                if (failure==null) {
                    failure=reported;
                }else {
                    failure.addSuppressed(reported);
                }
            } finally {
                // nothing holds the borrow until this returns: new WriteableTransactionTransactionImpl
                // runs a StampSession in a field initializer, and an Error out of a bulk import - an
                // OutOfMemoryError is the one to expect - would otherwise leave it out of the pool
                built=new ImportConnection(borrowed);
            }catch (Throwable e) {
                try {
                    txw.stampSession.close();
                } finally {
                    txw.catalogSession.close();
                    borrowed.close();
                }catch (Throwable e2) {
                    // suppressed rather than logged, the way the constructor of this importer joins the
                    // same pair: the failure being unwound is the one the caller asked about, and a
                    // return that failed on top of it is worth reading
                    e.addSuppressed(e2);
                }
                throw e;
            }
            // Entered under the monitor of the map, which close() takes to mark this import closed and
            // to take its connections away: without it a borrow in flight could be put back after
            // close() had walked the map, leaving a connection nothing would commit or return - or,
            // worse, be released twice, the second time onto a connection the pool had already handed
            // to somebody else.
            boolean tooLate=false;
            synchronized (connections) {
                if (closed) {
                    tooLate=true;
                }else {
                    connections.put(index, built);
                }
            }
            if (tooLate) {
                // outside the monitor: a return is a round trip, and close() must not wait behind it
                releaseUnwatched(built);
                throw importIsClosed();
            }
            return built;
        }

        /**
         * A connection of the pool for the given index, or null for a tree that is to share the
         * connection this import already has.
         * <p>
         * The pool having none left is not a reason to fail an import: what an import must not do is
         * put two threads on one connection, and the monitor of an {@link ImportConnection} sees to
         * that whether one tree writes through it or five. So a tree that cannot be given a
         * connection of its own is given the first one instead - the import runs with less of the
         * parallelism it asked for, rather than stopping halfway through with its clears already
         * committed. Before #891 an import held one connection for all of its trees and this is what
         * that looked like.
         * <p>
         * Every answer of the pool and of the database is taken this way, not only the one that ran
         * out of time. The pool at its bound, and a database that {@code CachedConnection} waited out
         * for the whole deadline, are reported as a {@code SQLTimeoutException} - but a limit of the
         * database that its dialect table does not recognize is raised at once instead: a mysql
         * account with a {@code MAX_USER_CONNECTIONS} of its own answers 1226 on SQLState 42000, and
         * a driver of no known dialect has no vendor code read at all (issue #1011). The type is no
         * rule either: a failure whose chain names the credentials of the backend is rebuilt as a
         * plain {@code SQLException} whatever the driver threw. Sorting them here would be that
         * classification written out a second time, with the failure of a multi-hour import as the
         * cost of getting it wrong (issue #1013).
         * <p>
         * Nothing is hidden by taking them all. The credentials, the driver and the database were
         * proved by the borrow this importer was built on, so a later one fails for a reason of the
         * database or of the network - and where the database really is gone, the connection this
         * import falls back to is gone with it and the next statement fails with what actually
         * happened, which is a better report than the failure of a borrow.
         * <p>
         * An interrupt is not one of those answers: it is how phase two stops an import
         * ({@code OnDiskMergeImporter.invokeParallel} gives its threads five seconds to answer one),
         * and a thread that met it by writing the tree through another connection would be carrying
         * on with the work it was told to drop. It goes back on the thread as well - the wait of a
         * borrow clears it - since what reads it next is the executor of phase two.
         */
        private Connection borrowedOrShared(Integer index) {
            try {
                // bounded, and by the default wait of a borrow however much longer the deployment
                // made that wait: the connections this one waits for are held by this import until
                // it ends, so an unbounded wait here is a thread waiting for itself - and a long one
                // is paid over again for every index the pool has no connection to spare for
                return getConnection(false, CachedConnection.DEFAULT_POOL_TIMEOUT_SECONDS);
            }catch (SQLException e) {
                sharedIndexes.add(index);
                if (e instanceof SQLTimeoutException) {
                    logger.debug(LocalizableMessage.raw("jdbc: the pool has no connection to spare for a tree of this import,"
                        + " which writes it through one it already holds: %s", stackTraceToSingleLineString(e)));
                }else {
                    // louder than the wait above: that one is the deployment's own bound being
                    // reached, while this is a database refusing a connection outright - the import
                    // goes on with less parallelism than it asked for, and an operator reading a
                    // long import has nothing else to tell them why
                    logger.warn(LocalizableMessage.raw("jdbc: no further connection is given to this import, which"
                        + " writes the tree through one it already holds: %s", stackTraceToSingleLineString(e)));
                }
                return null;
            }catch (InterruptedException e) {
                // put back where the borrow found it, before the wrapper leaves this class
                Thread.currentThread().interrupt();
                throw new StorageRuntimeException(e);
            }catch (Exception e) {
                throw e instanceof StorageRuntimeException ? (StorageRuntimeException) e : new StorageRuntimeException(e);
            }
        }

        /**
         * The connection every tree of this import falls back to, which is the one its constructor
         * borrowed - the only connection an import is sure to have.
         */
        private ImportConnection sharedConnection() {
            final ImportConnection shared=connections.get(FIRST_CONNECTION);
            if (shared==null) {
                throw importIsClosed(); // close() took the connections away
            }
            return shared;
        }

        /**
         * Refuses what arrives after {@code close()}: there is no transaction of this import left to
         * join, and the connection a write would go to is committed and back in the pool, serving
         * whoever borrowed it next.
         * <p>
         * Asked again under the monitor of the connection by everything that issues a statement. The
         * flag alone is a moment in time: a thread that read it before {@code close()} raised it, and
         * reached the connection after, would write on a connection of another borrower. Read under
         * the monitor it cannot: {@code close()} takes that same monitor to commit and to return the
         * connection, so either this thread is in front of the commit and part of the import, or it
         * is behind the return and refused.
         * <p>
         * Reasoned rather than pinned by a test: what a test would have to do is park a thread
         * between the read of the flag and the monitor, and the monitor is the only boundary there
         * is to park at. That a write after {@code close()} is refused at all is asserted, on one
         * thread, by {@code ImportConnectionsTestCase}.
         */
        private void checkOpen() {
            if (closed) {
                throw importIsClosed();
            }
        }

        private StorageRuntimeException importIsClosed() {
            return new StorageRuntimeException(new IllegalStateException(
                "this import is closed: its connections are committed and back in the pool"));
        }

        /**
         * Hands one connection back to the pool and closes the sessions its transaction opened
         * beside it - the stamp one and the catalog one (#888), both outside the pool and neither
         * outliving the import that opened it - under the monitor of the connection: the return
         * rolls back and hands the connection to the next borrower, so a statement of a straggling
         * import thread must not still be in flight on it - that is the
         * two-threads-on-one-connection of #891 with another borrower's operation on the other side
         * of it.
         * <p>
         * The failure of a return is returned rather than thrown: a throw out of here would leave
         * with the failure the caller actually came for - the commit of {@code close()}, and in its
         * {@code Throwable} branch the {@code Error} that branch exists to preserve - dropped on the
         * floor, and the connections after this one unreturned (#878).
         * <p>
         * By the time {@code close()} reaches this, no statement of the import can be in flight on
         * the connection anyway: every connection of that list has been through {@code commit()}
         * under this same monitor, and a write arriving after {@code close()} raised its flag is
         * refused under it by {@code checkOpen()}. So this monitor is not what the guarantee rests
         * on today and no test can tell it apart from the commit in front of it - it is here so that
         * the guarantee does not depend on that ordering, and it is all there is on the paths that
         * reach here with no commit in front of them ({@link #releaseUnwatched(ImportConnection)}).
         *
         * @return what went wrong on the way back, or null
         */
        private SQLException release(ImportConnection connection) {
            synchronized (connection) {
                try {
                    connection.con.close();
                    return null;
                } catch (Throwable e) {
                    // Throwable rather than SQLException: this close() is the return to the pool, whose
                    // rollback a driver is free to fail unchecked.
                    return e instanceof SQLException ? (SQLException) e
                        : new SQLException("a connection of the import could not be returned to the pool", e);
                } finally {
                    // under the monitor with the return itself: each of these sessions holds a connection
                    // of its own in a plain field that its close() reads and nulls without one
                    try {
                        connection.txw.stampSession.close();
                    } finally {
                        connection.txw.catalogSession.close();
                    }
                }
            }
        }

        /**
         * The return of a connection no caller is waiting on - the loser of a race to open one, and
         * one borrowed into an import that closed underneath it - which has no failure of an
         * operation for a failure of the return to ride along with.
         */
        private void releaseUnwatched(ImportConnection connection) {
            final SQLException failure=release(connection);
            if (failure!=null) {
                logger.trace(LocalizableMessage.raw("jdbc: unable to return a connection of the import: %s",
                    stackTraceToSingleLineString(failure)));
            }
        }

        /** The return of one connection, joined to the failure the caller is going to report. */
        private SQLException release(ImportConnection connection, SQLException failure) {
            final SQLException reported=release(connection);
            if (reported==null) {
                return failure;
            }
            if (failure==null) {
                return reported;
            }
            failure.addSuppressed(reported);
            return failure;
        }

        /**
         * Commits one connection of this import under its monitor, which is what keeps the commit
         * from being the second statement in flight on it: an import thread that did not answer the
         * interrupt of phase two can still be inside a statement while {@code close()} runs
         * ({@code OnDiskMergeImporter.invokeParallel} waits five seconds for its threads and closes
         * the importer whether they stopped or not), and two threads on one connection is the whole
         * of #891.
         * <p>
         * So a close waits for a statement of a straggler to finish, and a bulk statement carries no
         * bound of its own. That wait is not new: pgjdbc and Connector/J serialize the work of a
         * connection behind a lock of their own, so a commit issued beside a statement in flight
         * already waited there - what is new is that it waits on every dialect, sql server included,
         * rather than corrupting the driver's state on the one that does not lock.
         * <p>
         * A connection with nothing written since its last commit is left alone: the clears of
         * {@code beforePhaseOne} pass through here for every tree of a container, and a commit of an
         * empty transaction is a round trip to the database for nothing. Asked here rather than by
         * the caller, and under this monitor: read in front of it, {@code lastWrite} says what the
         * connection held rather than what it holds - a write in flight has not counted itself yet.
         * <p>
         * On both paths that reach this the wait is already paid in front of it: {@code close()}
         * reads {@code lastWrite} of every connection of the import under this same monitor before
         * it commits any of them, and {@code commitPeersOf()} holds it over this call - so no test
         * can tell this monitor from the ones ahead of it, and it is here so that what keeps two
         * threads off one connection does not rest on the order another method happens to work in.
         */
        private void commit(ImportConnection connection) throws SQLException {
            synchronized (connection) {
                if (connection.lastWrite==0) {
                    return;
                }
                connection.con.commit();
                connection.lastWrite=0;
            }
        }

        /**
         * Commits the other connections of this import, which is what the one connection an import
         * used to hold did of its own accord: {@code clearTree()} ends in a commit, and that commit
         * made durable every write the import had made so far. Split over several connections and
         * left to {@code close()}, those writes would stay uncommitted while the tables they are
         * about were emptied and committed one after another.
         * <p>
         * What that is worth depends on the order the caller writes in, and the two strategies of
         * {@code OnDiskMergeImporter} differ. A {@code rebuild-index} writes the
         * {@code setTrust(false)} of {@code RebuildIndexStrategy.beforePhaseOne} before it empties
         * the trees that flag describes, so this makes the flag durable in front of the clear it
         * belongs to: a server that stops in between comes back to an index that is empty and says
         * so. An {@code import-ldif} takes {@code AbstractTwoPhaseImportStrategy.beforePhaseOne},
         * which empties every tree of the container first and writes the flags after - so there this
         * bounds how much of an import stays uncommitted (the flags of one container are made
         * durable by the clears of the next), rather than closing that window. Not a regression of
         * the connections an import now takes: the one connection it held before #891 committed in
         * exactly the same places, because the caller writes in exactly the same order.
         * <p>
         * Run in front of the clear rather than after it, so that a peer whose commit fails leaves
         * the import with the tree not yet emptied: 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.
         * <p>
         * Which peers hold something to commit is decided under the monitor of each of them - by
         * {@link #commit(ImportConnection)}, which passes over a connection with nothing written
         * since its last commit - rather than by a read of {@code lastWrite} taken in front of that
         * monitor. A peer whose first write is still in flight has not set it yet ({@code put()}
         * counts a write once the statement has come back), and that is precisely the write a
         * cheaper test would pass over: {@code beforePhaseOne} runs on the import threads, one
         * container at a time per thread and several containers at once, so the flags of a container
         * being written are a peer of the clears of the next. The one connection an import held
         * before #891 carried that write into the commit of the clear - it was the same transaction,
         * and a clear issued beside it waited for it in the driver - so passing over it here would
         * be a commit point the single connection did not have.
         * <p>
         * What that costs is the wait: a clear now waits for a statement in flight on each peer, as
         * every write of an import waited for the one connection it all went through.
         * <p>
         * One connection is held at a time, so no thread of an import ever holds two of these
         * monitors and two threads clearing at once cannot wait for each other.
         */
        private void commitPeersOf(ImportConnection cleared) {
            for (final ImportConnection peer : connections.values()) {
                if (peer==cleared) {
                    continue;
                }
                try {
                    synchronized (peer) {
                        // under the monitor, like every other statement of an import: a clear that
                        // arrives once close() has committed and returned this connection would
                        // otherwise commit the transaction of whoever borrowed it next
                        checkOpen();
                        commit(peer);
                    }
                }catch (SQLException e) {
                    // reported the way the commit of the clear itself is: what these make durable is
                    // the work the clear is the commit point for
                    throw new StorageRuntimeException(e);
                }
            }
        }

        /**
         * Commits the given connections in the order they were last written to - see {@link #writes}.
         * <p>
         * The order is taken from a copy of what each connection carries rather than read as the sort
         * goes: a straggling import thread that got in front of {@code close()} can raise the number
         * of the connection it holds while this runs, and a sort whose keys move under it is reported
         * by {@code TimSort} as a comparator that violates its contract rather than as the race it is.
         */
        private void commitAll(List<ImportConnection> taken) throws SQLException {
            final Map<ImportConnection,Long> lastWrites=new IdentityHashMap<>();
            for (final ImportConnection connection : taken) {
                synchronized (connection) {
                    lastWrites.put(connection, connection.lastWrite);
                }
            }
            final List<ImportConnection> byLastWrite=new ArrayList<>(taken);
            byLastWrite.sort(Comparator.comparingLong(lastWrites::get));
            for (final ImportConnection connection : byLastWrite) {
                commit(connection);
            }
        }

        /**
         * Hands every connection this import took back to the pool, whatever went before. Returns
         * the failure the caller is to report: the return rolls back, and the rollback fails on
         * exactly the connection whose commit just did, so the commit stays the exception the caller
         * sees and this one rides along with it instead of replacing it.
         * <p>
         * Every connection is released even when one of them fails on the way: what a connection
         * left behind holds is a permit of the pool, and a pool is never removed from the map.
         */
        private SQLException releaseConnections(List<ImportConnection> taken, SQLException failure) {
            for (final ImportConnection connection : taken) {
                failure=release(connection, failure);
            }
            return failure;
        }

        // The connection goes back whatever the commit does, and the storage this importer opened
        // is closed whatever the connection does: an importer is closed on the way out of a failed
        // import as readily as a finished one - a clearTree() that reaches the bulk bound is one
        // way there - and a commit that throws on the way would otherwise leave the connection
        // out of the pool for good, holding the transaction and the locks of that import.
        // The connections go back whatever the commit does, and the storage this importer opened
        // is closed whatever they do: an importer is closed on the way out of a failed import as
        // readily as a finished one - a clearTree() that reaches the bulk bound is one way there -
        // and a commit that throws on the way would otherwise leave the connections out of the
        // pool for good, holding the transactions and the locks of that import.
        //
        // Every connection is committed, not only the one the constructor borrowed: each is a
        // transaction of its own, and what one of them holds uncommitted is the work of every tree
        // it was given (#891).
        @Override
        public void close() {
            try {
                // Taken out of the map rather than walked in it, under the monitor that guards it:
                // from here this import has no transaction left for a write to belong to, a borrow
                // still in flight is refused rather than left behind, and a second close() finds
                // nothing to commit or return - one that walked the map again would roll back and
                // re-pool connections the pool had already handed to somebody else.
                final List<ImportConnection> taken;
                final ImportConnection describing;
                synchronized (connections) {
                    closed=true;
                    describing=connections.get(FIRST_CONNECTION);
                    taken=new ArrayList<>(connections.values());
                    connections.clear();
                }
                SQLException failure=null;
                try {
                    con.commit();
                    if (aborted) {
                        logger.debug(LocalizableMessage.raw("jdbc: import aborted: statistics of the trees it wrote are left alone"));
                    }else {
                        updateTableStatistics(con, writtenTrees);
                    }
                    commitAll(taken);
                } catch (SQLException e) {
                    failure=e;
                } catch (Throwable t) {
                    // Back to the pool whatever came out of the commit, not only on the SQLException
                    // a driver is supposed to throw: nothing else holds this connection, and only
                    // its close() gives back the permit it took. A pool is never removed from the
                    // map, so a permit lost to an Error out of a bulk import - or to a driver
                    // a driver is supposed to throw: nothing else holds these connections, and only
                    // their close() gives back the permits they took. A pool is never removed from
                    // the map, so a permit lost to an Error out of a bulk import - or to a driver
                    // failing unchecked - is lost for the life of the server, and enough of them
                    // walk the bound down to nothing (issue #878).
                    final SQLException onTheWayOut=releaseConnection(null);
                    final SQLException onTheWayOut=releaseConnections(taken, null);
                    if (onTheWayOut!=null) {
                        t.addSuppressed(onTheWayOut);
                    }
                    throw t;
                }
                // Back to the pool even when the commit failed: nothing else holds this connection,
                // Everything but the connection that describes goes back before the statistics are
                // gathered: that is one statement per tree the import wrote, a full scan of the table
                // on oracle and bounded by a property of its own, and the connections of an import are
                // the pool's to hand to the operations of the server as soon as they are committed.
                for (final ImportConnection connection : taken) {
                    if (connection!=describing) {
                        failure=release(connection, failure);
                    }
                }
                try {
                    if (aborted) {
                        logger.debug(LocalizableMessage.raw("jdbc: import aborted: statistics of the trees it wrote are left alone"));
                    }else if (describing!=null && failure==null) {
                        // On the connection the constructor borrowed, under its monitor like every
                        // other statement of an import: the statements below describe a table to the
                        // optimizer rather than read one, and they run once every connection of this
                        // import is committed - so there is no work of another one left for them to
                        // miss.
                        synchronized (describing) {
                            updateTableStatistics(describing.con, writtenTrees);
                        }
                    }
                } catch (Throwable t) {
                    if (describing!=null) {
                        final SQLException onTheWayOut=release(describing, null);
                        if (onTheWayOut!=null) {
                            t.addSuppressed(onTheWayOut);
                        }
                    }
                    throw t;
                }
                // Back to the pool even when a commit failed: nothing else holds this connection,
                // so leaving it behind would leak it along with the failure.
                failure=releaseConnection(failure);
                if (describing!=null) {
                    failure=release(describing, failure);
                }
                if (failure!=null) {
                    throw new StorageRuntimeException(failure);
                }
@@ -5062,21 +5694,36 @@

        @Override
        public void clearTree(TreeName name) {
            txw.clearTree(name);
            final ImportConnection connection=connectionOf(name);
            commitPeersOf(connection); // in front of the clear, see there
            synchronized (connection) {
                checkOpen();
                connection.txw.clearTree(name);
                connection.lastWrite=0; // the clear ends in a commit of this connection
            }
            writtenTrees.add(name);
        }

        @Override
        public void put(TreeName treeName, ByteSequence key, ByteSequence value) {
            txw.put(treeName, key, value);
            final ImportConnection connection=connectionOf(treeName);
            synchronized (connection) {
                checkOpen();
                connection.txw.put(treeName, key, value);
                connection.written();
            }
            writtenTrees.add(treeName);
        }
		

        @Override
        public ByteString read(TreeName treeName, ByteSequence key) {
            return txr.read(treeName, key);
            final ImportConnection connection=connectionOf(treeName);
            synchronized (connection) {
                checkOpen();
                return connection.txr.read(treeName, key);
            }
        }
		

        // Bulk like every other statement of an import, by the class of the transaction it comes
        // from: this walks a whole tree with no client waiting on it - phase one of a rebuild-index
        // reads every record of id2entry through this cursor (OnDiskMergeImporter.ID2EntrySource) -
@@ -5084,7 +5731,88 @@
        // rather than a step along an index.
        @Override
        public SequentialCursor<ByteString, ByteString> openCursor(TreeName treeName) {
            return txr.openCursor(treeName);
            final ImportConnection connection=connectionOf(treeName);
            synchronized (connection) {
                checkOpen();
                return new ImportCursor(connection, connection.txr.openCursor(treeName));
            }
        }

        /**
         * A cursor of an import, every method of which runs under the monitor of the connection it
         * walks. An import has more trees than connections, so the tree this cursor walks shares
         * its connection with the trees written through it, and a batch of this cursor must not be
         * in flight there beside a statement of one of them.
         * <p>
         * The methods that issue no statement take the monitor as well, rather than being excused
         * on the ground that the thread which opened the cursor is the only one to call them:
         * nothing here enforces that, and what they read - the batch the last {@code next()} left
         * in {@link CursorImpl}, an {@code ArrayDeque} and four plain fields - is written under
         * that monitor and carries no memory barrier of its own. An uncontended monitor is what
         * that costs.
         */
        private final class ImportCursor implements SequentialCursor<ByteString, ByteString> {
            private final ImportConnection connection;
            private final SequentialCursor<ByteString, ByteString> cursor;

            ImportCursor(ImportConnection connection, SequentialCursor<ByteString, ByteString> cursor) {
                this.connection=connection;
                this.cursor=cursor;
            }

            @Override
            public boolean next() {
                synchronized (connection) {
                    checkOpen();
                    return cursor.next();
                }
            }

            // A cursor of an import is opened on the read transaction of its connection, whose
            // isReadOnly CursorImpl carries, so this forwards the refusal that transaction answers
            // with rather than a delete. Nothing is recorded against the connection for that reason:
            // a mark here would be a write number this cursor never made, and the highest one at
            // that - it would move its connection to the end of the order close() commits in, which
            // is where the trust flags of afterPhaseTwo belong. An importer that ever opens a
            // writeable cursor has to record the write there, next to the delete that made it.
            @Override
            public void delete() {
                synchronized (connection) {
                    checkOpen();
                    cursor.delete();
                }
            }

            @Override
            public boolean isDefined() {
                synchronized (connection) {
                    return cursor.isDefined();
                }
            }

            @Override
            public ByteString getKey() {
                synchronized (connection) {
                    return cursor.getKey();
                }
            }

            @Override
            public ByteString getValue() {
                synchronized (connection) {
                    return cursor.getValue();
                }
            }

            // Not refused after close() the way a read or a write of the importer is: this frees what
            // the last batch left in the cursor and reaches no connection, and the try-with-resources
            // of a cancelled phase-two task closes its cursor after the importer it walked.
            @Override
            public void close() {
                synchronized (connection) {
                    cursor.close();
                }
            }
        }
    }
    
@@ -5092,9 +5820,9 @@
    @Override
    public Importer startImport() throws ConfigException, StorageRuntimeException {
        // Everything this used to do before building the importer - opening a closed storage, and
        // borrowing the connection an import keeps for its whole duration - is the importer's own now
        // (#878). Split between the two, a failure in between had to be given back by whichever of them
        // had taken what, and the constructor's own throw was covered by neither.
        // borrowing the first of the connections an import keeps for its whole duration - is the
        // importer's own now (#878). Split between the two, a failure in between had to be given back
        // by whichever of them had taken what, and the constructor's own throw was covered by neither.
        return new ImporterImpl();
    }
    

 opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/CachedConnectionTestCase.java

@@ -15,8 +15,10 @@
 */
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.TreeName;
import org.opends.server.backends.pluggable.spi.AccessMode;
import org.opends.server.backends.pluggable.spi.Importer;
import org.testng.annotations.AfterClass;
@@ -36,6 +38,7 @@
import java.sql.Driver;
import java.sql.DriverManager;
import java.sql.DriverPropertyInfo;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.SQLTimeoutException;
import java.util.ArrayDeque;
@@ -61,6 +64,7 @@

import static org.mockito.Mockito.any;
import static org.mockito.Mockito.anyInt;
import static org.mockito.Mockito.anyString;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doThrow;
@@ -685,6 +689,10 @@
        final String url = StubDriver.PREFIX + "import-unchecked-commit";
        final Connection parent = mock(Connection.class);
        when(parent.isValid(anyInt())).thenReturn(true);
        // the statement of the write below, which is what gives the commit of close() something to do:
        // a connection an import wrote nothing through is not committed at all (#891), so an import
        // that writes nothing would reach neither the Error this test injects nor the return it is about
        when(parent.prepareStatement(anyString())).thenReturn(mock(PreparedStatement.class));
        doThrow(new Error("out of memory while importing")).when(parent).commit();
        stub.answerWith(parent);
        final JDBCBackendCfg cfg = mock(JDBCBackendCfg.class);
@@ -692,12 +700,17 @@
        final JDBCStorage storage = new JDBCStorage(cfg, null);
        storage.open(AccessMode.READ_WRITE);
        final Importer importer = storage.startImport();
        importer.put(new TreeName("dc=example,dc=com", "id2entry"),
            ByteString.valueOfUtf8("key"), ByteString.valueOfUtf8("value"));

        try {
            importer.close();
            fail("the failure of the commit was not reported");
        } catch (Error expected) {
            // reported to the caller, which is what an Error out of an import has to be
            // reported to the caller, which is what an Error out of an import has to be. Asserted
            // rather than accepted whole: an AssertionError of the fail() above is an Error too, and
            // would otherwise be caught here and read as the injected one
            assertEquals(expected.getMessage(), "out of memory while importing");
        }

        assertEquals(CachedConnection.poolOf(url).idleCount(), 1, "the import kept the connection of the pool");
@@ -1695,6 +1708,32 @@
        assertEquals(CachedConnection.deadlineOf(startedAt, 60), startedAt + 60_000);
    }

    /**
     * A borrow that carries a bound of its own waits for the shorter of the two, whichever way the
     * deployment spelled a wait with no bound at all.
     * <p>
     * Only an import carries one: it holds a connection per tree it writes until it ends (#891), so
     * the pool it waits at may be full of nothing but its own connections and the wait would be a
     * deadlock rather than a queue. It has somewhere to go when the wait runs out - the tree is
     * written through a connection the import already holds - so a long wait of the deployment is
     * capped rather than merely replaced: paid over again for every tree the pool has nothing to
     * spare for, it would be the duration of an import rather than a bound on it.
     */
    @Test
    public void testABorrowWithABoundOfItsOwnWaitsForTheShorterOfTheTwo() {
        assertEquals(CachedConnection.boundedWait(30, 60), 30, "the wait of the deployment was inside the bound");
        assertEquals(CachedConnection.boundedWait(600, 60), 60, "a long finite wait was not capped");
        assertEquals(CachedConnection.boundedWait(0, 60), 60, "0 stands for a wait with no bound");
        // the other spelling of a wait with no bound: seconds enough that the milliseconds they
        // stand for do not fit in a long, which the deadline of the borrow reads as forever
        assertEquals(CachedConnection.boundedWait(Long.MAX_VALUE / 1000, 60), 60,
            "a wait whose milliseconds overflow a long is unbounded too");
        assertEquals(CachedConnection.boundedWait(Long.MAX_VALUE, 60), 60);
        // ... and a borrow that carries no bound of its own takes the wait of the deployment whole
        assertEquals(CachedConnection.boundedWait(600, 0), 600);
        assertEquals(CachedConnection.boundedWait(0, 0), 0);
    }

    /** The connection string holds the credentials of the backend: a stall report must not carry them. */
    @Test
    public void testLoggedConnectionStringCarriesNoCredentials() throws Exception {

 opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/ImportConnectionsTestCase.java

New file
@@ -0,0 +1,1348 @@
/*
 * 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());
        }
    }
}

 opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/JDBCStatementBoundTestCase.java

@@ -1081,6 +1081,14 @@
            }

            @Override
            int poolMax() {
                // stood in for like the borrow above: read for real, this would intern a pool of the
                // static registry for the url of the mock configuration, and start its sweeper, from
                // a test that has no pool at all (#891)
                return 1;
            }

            @Override
            public StorageStatus getStorageStatus() {
                return StorageStatus.working(); // open already, so the importer borrows and no more
            }