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

@@ -244,6 +244,16 @@
    static final long STALL_WARNING_AFTER_MS = 1000;
    static final long STALL_WARNING_INTERVAL_MS = 10000;

    /**
     * The next wait of a connect that is being retried: a millisecond, doubling to {@link
     * #MAX_BACKOFF_MS} and staying there. One schedule for both loops that retry a connect of this
     * backend - the borrow of {@link #getConnection} and the catalog connect of {@code
     * JDBCStorage.newCatalogConnection} - so that the next change to it is a change to both (#929).
     */
    static long nextBackoffMs(long backoffMs) {
        return Math.min(backoffMs == 0 ? 1 : backoffMs * 2, MAX_BACKOFF_MS);
    }

    /** How many links of the cause and getNextException() chains of a failure are looked at. */
    private static final int MAX_CHAIN_LENGTH = 32;

@@ -275,7 +285,24 @@
    // safe form of it, the way warnedOnce below is: a static field of this class outlives every
    // borrow, and the password of the backend has no business in one.
    private static final Map<String, AtomicLong> lastStallWarning = new ConcurrentHashMap<>();
    private static final AtomicLong lastReadBoundWarning = new AtomicLong();

    /**
     * When a connection this backend could not put a read bound on was last reported, per
     * consequence of that failure rather than once for the JVM.
     * <p>
     * Keyed for the reason the stall warning above is keyed, and the reason is sharper here: the
     * connections this happens to do not share a fate. A connection of the pool is closed rather
     * than pooled, while the one catalog connection of a backend is kept and carries the bound of
     * its login for the rest of its life ({@code JDBCStorage.connectCatalog}, #929) - so a single
     * timestamp has the pool, which meets the failure first and on every connect, report for the
     * catalog connection beside it, and an operator reads that the connection was closed when the
     * one they will meet again was kept.
     * <p>
     * Package private so that a test can read what was reported and clear it, the way
     * {@link #warnedOnce} is read: the warning itself goes to a logger with a throttle no test can
     * wind back.
     */
    static final Map<String, AtomicLong> lastReadBoundWarning = new ConcurrentHashMap<>();

    /**
     * When an operation last reported that the database had dropped a connection of a pool, as a
@@ -1500,7 +1527,15 @@
                    timeout.initCause(reported(e, connectionString));
                    throw timeout;
                }
                backoffMs = Math.min(backoffMs == 0 ? 1 : backoffMs * 2, MAX_BACKOFF_MS);
                // The schedule the catalog connect of JDBCStorage waits on as well, from the one
                // place that holds it (#929). What the two do with the wait is not the same and is
                // not meant to be: this one hands it to the poll of the deque at the head of the
                // loop, where a peer returning a connection ends the wait early, while a catalog
                // connect has no peer to wait for and sleeps it out. The failure at the end of the
                // deadline differs as deliberately - 08001 here, the driver's own state there, a
                // manufactured class 08 being read by JDBCStorage.write() as a connection the
                // database dropped.
                backoffMs = nextBackoffMs(backoffMs);
                waitMs = Math.min(backoffMs, remaining);
                warnStall(connectionString, attempts, startedAt, e);
            } catch (RuntimeException e) {
@@ -1733,8 +1768,48 @@
        return previous < 0 ? 0 : previous;
    }

    /** How a connection of this pool is named where the read bound of its login would not come off. */
    static final String POOLED_CONNECTION = "a connection of this pool";
    /**
     * And what becomes of it: whatever the driver will not take there it has warned about already,
     * so such a connection serves the borrower that is waiting for it and is closed rather than
     * pooled - the bound of its login does not outlive it in the pool.
     */
    static final String POOLED_CONNECTION_FATE = "it is closed rather than pooled";

    static CachedConnection connect(String connectionString, ConnectDialect dialect, long connectTimeoutSeconds,
            Pool pool, boolean metered) throws SQLException {
        final Established established = establish(connectionString, dialect, connectTimeoutSeconds,
            POOLED_CONNECTION, POOLED_CONNECTION_FATE);
        final CachedConnection con = new CachedConnection(connectionString, established.con, pool, metered,
            established.loginBoundLifted);
        con.lastKnownAliveNanos = established.provenAt;
        return con;
    }

    /**
     * The login and the set-up behind it: one attempt of a connect, wherever this backend makes one.
     * <p>
     * The pool makes them through {@link #connect}, and the tree catalog of a backend opens a
     * connection of its own beside the pool ({@code JDBCStorage.newCatalogConnection}), the caller of
     * {@code openTree()} being inside a transaction and holding a pooled connection already. What an
     * attempt is, is the same either way and is here: the properties the driver is handed, the login,
     * the transaction the rows of this backend need, and the read bound the connection carries from
     * here on. Written out twice it drifted within one round - the standing read bound of #885 was
     * given to the pooled half alone, leaving the catalog connection with the lift and no bound at
     * all between its statements - which is why the two share this rather than being kept in step by
     * hand (#929).
     * <p>
     * What is not shared is what a caller makes of the connection, which is what {@link Established}
     * hands back, and what becomes of one whose read bound would not come off: the pool closes such a
     * connection rather than pooling it, while the catalog keeps it, having no second connection to
     * fall back to. That difference is the {@code fate} of the warning, and it is the whole of it.
     *
     * @param what how the connection is named in the warning about a read bound that would not take
     * @param fate what becomes of such a connection, named in the same warning
     */
    static Established establish(String connectionString, ConnectDialect dialect, long connectTimeoutSeconds,
            String what, String fate) throws SQLException {
        // A driver is free to write into the map it is handed, so it gets one of its own.
        final Properties properties = new Properties();
        final boolean readBoundSet = dialect != null && connectTimeoutSeconds > 0
@@ -1744,22 +1819,39 @@
        // it returned could outlive a drop reported while it was still going on.
        final long provenAt = System.nanoTime();
        final Connection conNew = DriverManager.getConnection(connectionString, properties);
        boolean poolable = true;
        final boolean loginBoundLifted;
        try {
            // still under the read bound: both of these are round trips of their own
            conNew.setAutoCommit(false);
            conNew.setTransactionIsolation(TRANSACTION_READ_COMMITTED);
            // whatever the driver will not take here it has warned about already: a connection left
            // carrying the read bound of its login serves the borrower that is waiting for it and is
            // closed rather than pooled, so that bound does not outlive it in the pool
            poolable = applyStandingReadBound(conNew, connectionString, dialect, connectTimeoutSeconds, readBoundSet);
            loginBoundLifted = applyStandingReadBound(conNew, connectionString, dialect, connectTimeoutSeconds,
                readBoundSet, what, fate);
        } catch (SQLException | RuntimeException e) { // nothing holds this connection yet: it would leak
            closeQuietly(conNew);
            closeQuietly(conNew, e);
            throw e;
        }
        final CachedConnection established = new CachedConnection(connectionString, conNew, pool, metered, poolable);
        established.lastKnownAliveNanos = provenAt;
        return established;
        return new Established(conNew, loginBoundLifted, provenAt);
    }

    /** A connection just established, and what the caller that asked for it has to know about it. */
    static final class Established {
        /** The connection, set up and carrying the read bound it keeps from here on. */
        final Connection con;
        /**
         * Whether the read bound of the login is off it. False for a driver that would not take the
         * bound back, and the one thing a caller has to act on: such a connection fails every
         * statement slower than a connect for the rest of its life, so the pool hands it to the
         * borrower waiting for it and does not pool it afterwards.
         */
        final boolean loginBoundLifted;
        /** When the login was known to be through, as {@link System#nanoTime()} reads it. */
        final long provenAt;

        Established(Connection con, boolean loginBoundLifted, long provenAt) {
            this.con = con;
            this.loginBoundLifted = loginBoundLifted;
            this.provenAt = provenAt;
        }
    }

    /**
@@ -1780,50 +1872,91 @@
     * the deadline of the borrow instead, and a deployment running that way would otherwise set the
     * property here and get nothing for it.
     * <p>
     * Returns whether the connection may be pooled. A connection still carrying the bound of its
     * login must not be: it would fail the statements of every borrower after this one. A
     * connection that merely never took the standing bound may be - that is the connection this
     * pool handed out before the property existed.
     * Returns whether the read bound of the login is off the connection. One still carrying it must
     * not be pooled: it would fail the statements of every borrower after this one. A connection
     * that merely never took the standing bound may be - that is the connection this pool handed out
     * before the property existed.
     *
     * @param what how the connection is named in the warning; the caller's, since the same failure
     *        ends a connection of the pool and the one connection of a backend's tree catalog
     * @param fate what becomes of a connection whose read bound would not take, named in the same
     *        warning: the pool closes it rather than pooling it, the catalog keeps it
     */
    private static boolean applyStandingReadBound(Connection con, String connectionString, ConnectDialect dialect,
                                                  long loginBoundSeconds, boolean readBoundSet) {
                                                  long loginBoundSeconds, boolean readBoundSet,
                                                  String what, String fate) {
        final int millis = standingReadBoundMillis(connectionString, dialect);
        if (millis == 0 && !readBoundSet) {
            return true; // nothing of ours on this connection: nothing to set here, and nothing to lift
        }
        final String consequence = readBoundSet
            ? "statements taking longer than the " + loginBoundSeconds
                + "s the login of this connection was bounded by fail on it, and it is closed rather than pooled"
            : "this connection carries no read bound of its own, so a read of it the database stops answering waits"
        return setNetworkTimeout(con, millis, readBoundConsequence(readBoundSet, loginBoundSeconds, what, fate))
            || !readBoundSet;
    }

    /**
     * What a driver refusing the read bound costs the connection it refused it on, as the warning
     * about it says so.
     * <p>
     * Built apart from the logging of it for the reason {@link #stallMessage} is, and it is what
     * tells the two callers of {@link #establish} apart in the one place they differ: a connection
     * of the pool is closed rather than pooled where this fails, while the one catalog connection
     * of a backend is kept and carries the bound of its login for the rest of its life (#929).
     * Reached through the shipped path only by a driver that really refuses the call, which is why
     * it is a method a test can hold to the difference rather than a string inline.
     */
    static String readBoundConsequence(boolean readBoundSet, long loginBoundSeconds, String what, String fate) {
        return readBoundSet
            ? "statements on " + what + " taking longer than the " + loginBoundSeconds
                + "s the login of this connection was bounded by fail on it, and " + fate
            : what + " carries no read bound of its own, so a read of it the database stops answering waits"
                + " with no deadline able to reach it";
        return setNetworkTimeout(con, millis, consequence) || !readBoundSet;
    }

    /**
     * Puts a network timeout on a connection, reporting a driver that will not take one. The
     * consequence is the caller's to name: the same failure ends a freshly established connection
     * carrying the read bound of its login and a pooled one whose bound could not be put back.
     * <p>
     * The failure itself rather than its message alone: the drivers that refuse this call refuse it
     * as {@code SQLFeatureNotSupportedException}, which a driver is free to raise with no message
     * at all - and a line reading "(null)" says neither what refused nor why.
     */
    private static boolean setNetworkTimeout(Connection con, int millis, String consequence) {
        try {
            con.setNetworkTimeout(DIRECT_EXECUTOR, millis);
            return true;
        } catch (SQLException | RuntimeException e) {
            // Throttled rather than reported once for the life of the JVM: every connection this
            // happens to carries a read bound it was never meant to keep, and a statement dying of
            // it hours later needs a warning of its own to be traced back to here.
            final long now = System.currentTimeMillis();
            final long last = lastReadBoundWarning.get();
            if (now - last >= STALL_WARNING_INTERVAL_MS && lastReadBoundWarning.compareAndSet(last, now)) {
            if (readBoundWarningDue(consequence, System.currentTimeMillis())) {
                logger.warn(LocalizableMessage.raw(
                    "The read bound of a JDBC connection could not be set to %d ms (%s): %s",
                    millis, e.getMessage(), consequence));
                    millis, e, consequence));
            }
            return false;
        }
    }

    /**
     * Whether the warning about a read bound that would not take is due, and the filing of it.
     * <p>
     * Throttled rather than reported once for the life of the JVM: every connection this happens to
     * carries a read bound it was never meant to keep, and a statement dying of it hours later needs
     * a warning of its own to be traced back to here. Throttled per consequence and not once for the
     * JVM, for the reason {@link #stallWarningDue} is keyed per url: a connection of the pool meets
     * this failure on every connect and would otherwise silence the line about the one catalog
     * connection of a backend, which is kept rather than closed and has no other line about it at
     * all (#929).
     * <p>
     * Filing the moment is part of deciding it, so that two threads meeting the failure at once
     * report once.
     */
    static boolean readBoundWarningDue(String consequence, long now) {
        final AtomicLong lastOfThisConsequence =
            lastReadBoundWarning.computeIfAbsent(consequence, key -> new AtomicLong());
        final long last = lastOfThisConsequence.get();
        return now - last >= STALL_WARNING_INTERVAL_MS && lastOfThisConsequence.compareAndSet(last, now);
    }

    /**
     * Whether the database took no connection for the moment, rather than refusing one for good:
     * it is at its connection limit - one of our own connections is on its way back to the pool -
     * or it is not accepting connections yet, the state a database on its way up reports while it
@@ -1965,6 +2098,11 @@
     * RootContainer.open() makes the message of the cause the message of what it throws - so a
     * link left as it stands would carry the password past the wrapper. The SQLState and the
     * vendor code of every link survive it: they are what tells a caller what happened.
     * <p>
     * The whole chain is the causes, the further exceptions of {@code getNextException()} and what
     * was suppressed on each of them - the last being where this class puts the failure of a close
     * it could not make (#929), and a link nothing else points at is a link a walk that skipped it
     * would report unredacted.
     */
    static SQLException reported(SQLException e, String connectionString) {
        return holdsCredentials(e, connectionString)
@@ -1972,7 +2110,15 @@
            : e;
    }

    /** The same of an unchecked failure: a driver is free to report a connect it will not make as one. */
    /**
     * The same of an unchecked failure: a driver is free to report a connect it will not make as one.
     * <p>
     * Flat where {@link #reported} rebuilds a chain - the cause of a failure of this shape is the
     * driver's own business - with the one exception of what was suppressed on it, which is this
     * class's own: the close of a connection whose set-up failed is reported there (#929), and a
     * rebuild dropping it would say nothing of a driver that will not close on the only path where
     * the message is redacted at all.
     */
    static Exception reportedUnchecked(RuntimeException e, String connectionString) {
        if (!holdsCredentials(e, connectionString)) {
            return e;
@@ -1981,6 +2127,7 @@
            + (e.getMessage() == null ? "" : ": " + redact(e.getMessage(), connectionString)),
            CONNECT_FAILED_SQL_STATE);
        redacted.setStackTrace(e.getStackTrace());
        copySuppressed(e, redacted, connectionString, new int[] { MAX_CHAIN_LENGTH });
        return redacted;
    }

@@ -2008,6 +2155,14 @@
                enqueue(pending, visited, ((SQLException) t).getNextException());
            }
            enqueue(pending, visited, t.getCause());
            // and the links nothing else of a failure points at: a close that could not be made is
            // carried here (establish(), #929) and an interrupt that ended a wait for a connect is
            // (JDBCStorage.newCatalogConnection), and a driver names the url it could not close as
            // readily as the one it could not open. Left out of this walk, such a link is the one
            // way a password of this backend reaches the log unredacted.
            for (final Throwable suppressed : t.getSuppressed()) {
                enqueue(pending, visited, suppressed);
            }
        }
        return false;
    }
@@ -2038,9 +2193,25 @@
        if (e.getCause() != null) {
            copy.initCause(budget[0] > 0 ? redactedLink(e.getCause(), connectionString, budget) : droppedTail());
        }
        copySuppressed(e, copy, connectionString, budget);
        return copy;
    }

    // The suppressed links of a failure are rebuilt with the rest of it and counted against the
    // same budget. They are not decoration here: the failure of a close that could not be made
    // rides on the failure being unwound (establish(), #929), and a rebuild that dropped them
    // would lose it at the one deployment whose log is redacted - which is the deployment whose
    // url has a password in it, the log that is worth reading.
    private static void copySuppressed(Throwable from, Throwable to, String connectionString, int[] budget) {
        for (final Throwable suppressed : from.getSuppressed()) {
            if (budget[0] <= 0) {
                to.addSuppressed(droppedTail());
                return;
            }
            to.addSuppressed(redactedLink(suppressed, connectionString, budget));
        }
    }

    // What stands where the budget ran out. Without it the same failure logs its root cause when
    // the url of the backend has no password in it and loses it without a word when it has, which
    // is a report of a connect nobody can read against a report of one they can.
@@ -2062,6 +2233,7 @@
        if (t.getCause() != null) {
            copy.initCause(budget[0] > 0 ? redactedLink(t.getCause(), connectionString, budget) : droppedTail());
        }
        copySuppressed(t, copy, connectionString, budget);
        return copy;
    }

@@ -2320,6 +2492,25 @@
        }
    }

    /**
     * Closes a connection nothing holds yet, reporting the failure of the close on the one being
     * unwound: the connection is gone either way, and a driver that will not close is worth knowing
     * about where the failure that leads here is reported.
     * <p>
     * Package private because every connect of this backend closes a connection it could not set up
     * the same way: the two roads {@link #establish} holds, and the stamp connection of {@code
     * JDBCStorage.newStampConnection}, which has bounds of its own but the same rule about a close.
     */
    static void closeQuietly(Connection con, Throwable unwinding) {
        try {
            con.close();
        } catch (SQLException | RuntimeException e) {
            // the unchecked one as well: this runs from the catch of a failure it must not replace
            // (JLS 14.20.2)
            unwinding.addSuppressed(e);
        }
    }

    @Override
    public Statement createStatement() throws SQLException {
        return parent.createStatement();

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

@@ -1629,9 +1629,10 @@
            // transaction.
            con.commit();
        }catch (SQLException e) { // nothing else holds this connection yet: it would leak
            try {
                con.close();
            }catch (SQLException e2) {}
            // and a driver that will not close is said so on the failure being unwound, as it is on
            // the two roads CachedConnection.establish() holds (#929): this connection is gone either
            // way, and the failure of the set-up is the one report this attempt makes
            CachedConnection.closeQuietly(con, e);
            throw e;
        }
        return con;
@@ -1678,8 +1679,9 @@
     * thread that is waiting. It is the pool's retry that is wanted here and not its queue, which is
     * why the loop below is its own rather than a borrow of {@link CachedConnection#getConnection}.
     * What one attempt is, is the login and the set-up behind it, exactly as an attempt of a borrow is
     * ({@code CachedConnection.connect}): a session the server takes and then kills off answers the
     * first statement of the set-up rather than the login, and it is the same refusal either way.
     * - the same code, {@link CachedConnection#establish} (#929): a session the server takes and then
     * kills off answers the first statement of the set-up rather than the login, and it is the same
     * refusal either way.
     * <p>
     * That is also what this wait is weaker than a borrow at, and it is worth writing down rather than
     * leaving to be discovered: a borrow can be answered by a peer handing a connection back, while
@@ -1812,7 +1814,11 @@
                        deadline==budgetDeadline, now-startedAt, attempts, e);
                }
                CachedConnection.warnStallOutsidePool(connectionString, "tree catalog", attempts, startedAt, e);
                backoffMs=Math.min(backoffMs==0 ? 1 : backoffMs*2, CachedConnection.MAX_BACKOFF_MS);
                // the schedule of the pool, from the one place that holds it (#929). The wait is slept
                // out rather than handed to the deque of the pool, which is the difference this loop
                // exists for: nothing here can be answered by a peer returning a connection, so there
                // is no queue to wait on - see the head of this method
                backoffMs=CachedConnection.nextBackoffMs(backoffMs);
                try {
                    Thread.sleep(Math.min(backoffMs, remaining));
                }catch (InterruptedException interrupted) {
@@ -1881,57 +1887,47 @@
        return timeout;
    }

    /** How this connection is named where the read bound of its login would not come off. */
    static String catalogConnectionNamed(String backendId) {
        return "the catalog connection of backend "+backendId;
    }

    /**
     * And what becomes of it, which is the whole of what this road does not share with the pool's:
     * the pool closes such a connection rather than pooling it, having a borrower to hand another
     * to, while this backend has one catalog connection and nothing behind it.
     */
    static final String CATALOG_CONNECTION_FATE=
        "it is kept as it is: this backend has one catalog connection and no second to fall back to";

    /**
     * One attempt of {@link #newCatalogConnection}, established and set up or left holding nothing.
     * Failures leave here as the driver reported them, checked and unchecked alike: what a retry is
     * decided on is the chain of the original, and the redaction is the caller's - a redacted copy is
     * rebuilt link by link, so redacting an attempt that is about to be retried would pay for a
     * failure nobody ever sees.
     * <p>
     * The login, the transaction it is set up for and the read bound it carries afterwards are the
     * pool's own, from the one place that holds them ({@link CachedConnection#establish}, #929):
     * written out here as well they drifted inside a single round, the standing read bound of #885
     * reaching the pooled half alone and leaving this connection with nothing bounding the reads
     * between its statements - its {@code commit()}, the {@code rollback()} of a session given up and
     * the {@code close()} of one that lost the race to another thread.
     * <p>
     * What is this connection's own is what becomes of it where the read bound of the login will not
     * come off. A driver that refuses to take it back leaves it in force for the life of the
     * connection, and that bound is the one this attempt was given - near the end of the deadline of
     * the retry, a second. The connection is kept all the same, where the pool closes such a
     * connection rather than pooling it: this backend has one catalog connection and no borrower
     * behind it to hand another to, and failing here instead would stop the backend opening on a
     * driver whose setNetworkTimeout is not implemented at all, where the pooled connection beside it
     * works. There is no state to fail with that {@code write()} does not read as a connection the
     * database dropped, either. So it is reported and the connection is used, at the bound in force.
     */
    private Connection connectCatalog(String connectionString, CachedConnection.ConnectDialect dialect,
            long timeoutSeconds) throws SQLException {
        // A driver is free to write into the map it is handed, so every attempt gets one of its own.
        final Properties properties=new Properties();
        final boolean readBoundSet=dialect!=null && timeoutSeconds>0
            && dialect.bound(connectionString, properties, timeoutSeconds);
        final Connection con=DriverManager.getConnection(connectionString, properties);
        try {
            con.setAutoCommit(false);
            con.setTransactionIsolation(Connection.TRANSACTION_READ_COMMITTED);
        }catch (SQLException | RuntimeException e) { // nothing else holds this connection yet: it would leak
            closeQuietly(con, e);
            throw e;
        }
        if (readBoundSet) {
            try {
                // only where this code set one: a read bound of the connection string is the
                // administrator's and is not lifted along with it, exactly as the pool leaves it
                con.setNetworkTimeout(Runnable::run, 0);
            }catch (SQLException | RuntimeException e) {
                // A driver that will not take the bound back leaves it in force for the life of the
                // connection, and that bound is the one this attempt was given - near the end of the
                // deadline of the retry, a second. The connection is kept all the same, which is the
                // pool's own answer to this failure: it stops pooling such a connection and still hands
                // it to the borrower that is waiting. Failing here instead would stop the backend opening
                // on a driver whose setNetworkTimeout is not implemented at all, where the pooled
                // connection beside it works - and there is no state to fail with that write() does not
                // read as a connection the database dropped. So it is reported, at the bound in force.
                logger.warn(LocalizableMessage.raw("jdbc: the catalog connection of backend %s keeps the %ds read bound its login was given, so a statement of the catalog slower than that fails on it: %s",
                    config.getBackendId(), timeoutSeconds, stackTraceToSingleLineString(e)));
            }
        }
        return con;
    }

    /** Closes a connection nothing holds yet, reporting the failure of the close on the one being unwound. */
    private static void closeQuietly(Connection con, Throwable unwinding) {
        try {
            con.close();
        }catch (SQLException | RuntimeException e) {
            // the unchecked one as well: this runs from the catch of a failure it must not replace
            // (JLS 14.20.2), which is the rule every close of this class keeps
            unwinding.addSuppressed(e);
        }
        return CachedConnection.establish(connectionString, dialect, timeoutSeconds,
            catalogConnectionNamed(config.getBackendId()), CATALOG_CONNECTION_FATE).con;
    }

    /**

 opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ExternalChangelogDomain.java

@@ -153,12 +153,26 @@
  {
    return true;
  }
  /** {@inheritDoc} */
  /**
   * {@inheritDoc}
   * <p>
   * Refused while a total update runs, as a change of the domain entry is: the attributes
   * this entry carries are published as the session of the domain comes up, so applying
   * them restarts that session, and an import into this replica reads its entries over
   * it. Through the server configuration such a restart would wait for the listener
   * thread the import runs on, which waits in turn for the lock of the configuration the
   * change holds, to enable the backend back once the stream it was reading ends.
   */
  @Override
  public boolean isConfigurationChangeAcceptable(
      ExternalChangelogDomainCfg configuration,
      List<LocalizableMessage> unacceptableReasons)
  {
    if (domain.ieRunning())
    {
      unacceptableReasons.add(NOTE_ERR_CANNOT_CHANGE_CONFIG_DURING_TOTAL_UPDATE.get());
      return false;
    }
    return true;
  }


 opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java

@@ -6158,8 +6158,9 @@
        changeConfig(configuration);

        // Read assured + fractional configuration and each time reconnect if needed. A
        // domain which owns its session gets none of those reconnections.
        final boolean allowReconnection = !ownsItsSession();
        // session which has an owner - the domain itself, or a total update into this
        // replica - gets none of those reconnections.
        final boolean allowReconnection = !sessionHasAnOwner();
        readAssuredConfig(configuration, allowReconnection);
        readFractionalConfig(configuration, allowReconnection);
        solveConflictFlag = isSolveConflict(configuration);
@@ -6207,9 +6208,10 @@
  }

  /**
   * Whether the session of this domain has an owner other than the replay thread which
   * would restart it after a failed replay: the domain itself, when it is shutting down or
   * disabled ({@link #ownsItsSession()}), or a total update into this replica.
   * Whether the session of this domain has an owner other than the thread which would
   * restart it - a replay thread after a failed replay, or a configuration change for what
   * it carries: the domain itself, when it is shutting down or disabled
   * ({@link #ownsItsSession()}), or a total update into this replica.
   * <p>
   * The total update owns the session from the moment it is asked for, not from the
   * moment its entries stream: the {@code InitializeTargetMsg} which answers the request
@@ -6223,6 +6225,15 @@
   * Listed, it holds the ServerState back as well: a commit moves the state no further than
   * the oldest uncommitted change, so the state in memory, and the one persisted from it,
   * stop at the change until that restart.
   * <p>
   * A configuration change is refused while a total update runs - by the listener of the
   * domain entry and by the one of its external changelog entry - so what reaches the
   * domain all the same is a change accepted before the total update was asked for. The
   * restart it asks for is refused on this predicate too, and the session the import
   * starts when it ends reads the configuration stored meanwhile. Made through the server
   * configuration, that restart would not end: it waits for the listener thread, which is
   * the import, and the import waits for the lock of the configuration the change holds,
   * to enable the backend back once the stream it was reading ends.
   */
  private boolean sessionHasAnOwner()
  {
@@ -6234,21 +6245,30 @@
  {
    synchronized (serviceStateLock)
    {
      if (ownsItsSession())
      if (sessionHasAnOwner())
      {
        /*
         * The domain is going away or is being imported into: a restart here would bring
         * a session, and the listener thread which goes with it, back up on a domain
         * whose ServerState is gone from memory. The session started when the domain is
         * enabled again reads the configuration this restart was asked for.
         * The domain is going away or is disabled: a restart here would bring a session,
         * and the listener thread which goes with it, back up on a domain whose
         * ServerState is gone from memory. The session started when the domain is enabled
         * again reads the configuration this restart was asked for.
         *
         * Or a total update into this replica is reading the session: the import streams
         * over it, on the listener thread a restart would stop and wait for. Stopped, the
         * broker ends the stream on the entries which had arrived; waited for, the listener
         * thread ends the import and enables the backend back through the server
         * configuration - whose lock a change made through it holds while it waits. The
         * import starts the next session itself when it ends, from the state it loaded,
         * and that session reads the configuration this restart was asked for.
         *
         * Recorded rather than passed over in silence: the configuration a restart was
         * asked for is stored, and it is the session which is not brought up on it, so a
         * change which reports plain success would have the administrator believe the
         * domain is running on it already. A domain disabled for a total update comes up
         * on it when the total update ends; one which stays disabled - enable() gives up
         * when the data state it reads cannot be loaded, and nothing calls it again -
         * never does, and that is what the administrator is told to act on.
         * on it when the total update ends, and so does one being imported into; one which
         * stays disabled - enable() gives up when the data state it reads cannot be
         * loaded, and nothing calls it again - never does, and that is what the
         * administrator is told to act on.
         */
        onSessionRestartSuppressed();
        return;
@@ -6299,7 +6319,13 @@
  public boolean isConfigurationChangeAcceptable(
         ReplicationDomainCfg configuration, List<LocalizableMessage> unacceptableReasons)
  {
    // Check that a import/export is not in progress
    /*
     * Check that a import/export is not in progress. The listener of the external
     * changelog entry of this domain refuses its change for the same reason: what either
     * change restarts the session for, an import into this replica is reading over that
     * session. One which starts between this check and the change being applied meets
     * the restart guard instead (see sessionHasAnOwner()).
     */
    if (ieRunning())
    {
      unacceptableReasons.add(

 opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java

@@ -2127,19 +2127,21 @@

        if (msg == null)
        {
          if (broker.shuttingDown())
          {
            // The server is in the shutdown process
            return null;
          }
          else
          {
            // Handle connection issues
            ieCtx.setExceptionIfNoneSet(new DirectoryException(
                ResultCode.OTHER, ERR_INIT_RS_DISCONNECTION_DURING_IMPORT
                    .get(broker.getReplicationServer())));
            return null;
          }
          /*
           * The stream ended before the DoneMsg of the exporter: the broker lost its
           * connection, or it was stopped under the import - by the shutdown of the server,
           * or by a restart of the session which every road takes through disableService().
           * Either way the import is a failure and is recorded as one (issue #1039): the
           * import which ends on the entries which had arrived would otherwise be reported
           * as finished, with the generationId of the exporter loaded from the base entry
           * among them, and the replica would come up as a peer of the exporter over part of
           * its data. A failed import has its generationId computed over the data instead.
           */
          final LocalizableMessage cause = broker.shuttingDown()
              ? ERR_INIT_SESSION_STOPPED_DURING_IMPORT.get(getBaseDN(), getServerId(), ieCtx.importSource)
              : ERR_INIT_RS_DISCONNECTION_DURING_IMPORT.get(broker.getReplicationServer());
          ieCtx.setExceptionIfNoneSet(new DirectoryException(ResultCode.OTHER, cause));
          return null;
        }

        // Check good ordering of msg received
@@ -3487,8 +3489,9 @@
   * back between the stop and the start, and both halves are counted by the session
   * generation. A subclass may leave it alone: a domain which is shutting
   * down, or which was disabled for a total update, owns its session and is not given one
   * back by a configuration change. One which does reports it through
   * {@link #onSessionRestartSuppressed()}.
   * back by a configuration change, and a total update into this replica reads its
   * entries over the session and starts the next one itself. One which does reports it
   * through {@link #onSessionRestartSuppressed()}.
   */
  protected void restartService()
  {
@@ -3505,9 +3508,10 @@
   * <p>
   * The configuration is stored either way, and the session started next reads it - so
   * this says that the change is not live yet rather than that it was lost. A domain
   * which restarts its session for every change never reaches this; one which owns its
   * session while it is shutting down or disabled for a total update overrides it to tell
   * the administrator what is waiting for that session.
   * which restarts its session for every change never reaches this; one whose session has
   * an owner - itself while it is shutting down or disabled for a total update, or a total
   * update into this replica reading it - overrides it to tell the administrator what is
   * waiting for that session.
   */
  protected void onSessionRestartSuppressed()
  {
@@ -4002,9 +4006,10 @@
       * configuration is: the assured timeout is the one property a session does not have
       * to be restarted for, so a change carrying it alone - reported as applied and then
       * dropped, before - is applied here. A caller which does not allow the reconnection
       * has no session running assured replication either: the domain is being built, is
       * shutting down, or is disabled for the length of a total update, and the session
       * its enable() starts reads what is stored here.
       * has no session to negotiate it over: the domain is being built, is shutting down,
       * or is disabled for the length of a total update - or a total update into this
       * replica is reading the session, which it must not stop - and the session started
       * next, by enable() or by the import when it ends, reads what is stored here.
       */
      assuredConfig = config;
      if (needRestart)

 opendj-server-legacy/src/messages/org/opends/messages/replication.properties

@@ -692,9 +692,10 @@
 replication domain on "%s": %s
NOTE_REPLICATION_DOMAIN_SESSION_NOT_RESTARTED_327=The configuration change was applied to the \
 replication domain on "%s", but the session to the replication server was not restarted for it: \
 the domain is shutting down, or it is disabled for the length of a total update. The change is \
 stored and takes effect when the session is started again, which a domain left disabled by a \
 failed import or restore never does
 the domain is shutting down, it is disabled for the length of a total update, or a total update \
 into this replica is reading that session. The change is stored and takes effect when the \
 session is started again - a total update starts it again when it ends, a domain left disabled \
 by a failed import or restore never does
ERR_REPLAY_ENTRYUUID_SEARCH_FAILED_322=Could not read the data to check change %s for a conflict \
 in domain "%s": the search of the entry with entryUUID %s did not run (%s). The change is not \
 applied on what a search which read nothing seemed to say about the data, and is not recorded \
@@ -704,3 +705,7 @@
 comes back with the last state it did write and replays the changes since
ERR_STATE_CHECKPOINTER_NOT_STOPPED_324=The state checkpointer of domain "%s" has not stopped within \
 %d ms : the shutdown of the domain goes on without it
ERR_INIT_SESSION_STOPPED_DURING_IMPORT_329=Domain %s (server id: %s) : the session to the \
 replication server was stopped before the initialization from server %s completed. The \
 entries which had arrived are imported, and the generation id of the data is computed over \
 them rather than taken from the exporter

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

@@ -843,6 +843,9 @@
     * Walked by identity rather than link by link: a driver is free to make the cause and the next
     * exception of a link the same failure, which is the very shape the sibling test builds, and a
     * helper looping on it would hang the run it is checking for exactly that.
     * <p>
     * The suppressed links along with the rest: the close of a connection whose set-up failed is
     * carried there (#929), and everything that prints a failure prints those too.
     */
    private static void assertNoCredentials(Throwable failure) {
        final Set<Throwable> seen = Collections.newSetFromMap(new IdentityHashMap<Throwable, Boolean>());
@@ -856,6 +859,9 @@
                enqueue(pending, seen, ((SQLException) t).getNextException());
            }
            enqueue(pending, seen, t.getCause());
            for (final Throwable suppressed : t.getSuppressed()) {
                enqueue(pending, seen, suppressed);
            }
        }
    }

@@ -936,6 +942,44 @@
    }

    /**
     * A credential named by a suppressed link alone is redacted like any other. The close of a
     * connection whose set-up failed rides there (establish(), #929) and an interrupt that ended a
     * wait for a catalog connect does, and a driver names the url it could not close as readily as
     * the one it could not open. Left out of the walk, such a link is the one way the password of
     * this backend reaches the server error log as it stands: the failure it hangs on says nothing
     * of the url, so the failure is handed on unredacted, suppressed link and all.
     */
    @Test(timeOut = 60000)
    public void testACredentialNamedOnlyBySuppressedIsStillRedacted() throws Exception {
        final String url = "jdbc:postgresql://opendj:S3cretOfTheBackend@127.0.0.1:5432/opendj";
        final SQLException setupFailure = new SQLException("Connection to 127.0.0.1:5432 refused", "08006", 1);
        setupFailure.addSuppressed(new SQLException("could not close " + url, "08003", 2));

        assertNoCredentials(CachedConnection.reported(setupFailure, url));
    }

    /**
     * ... and the link itself survives the rebuild rather than being dropped along with the
     * password. A redacted failure is the only failure the deployment with a password in its url
     * ever sees, so a rebuild that left the suppressed links behind would answer "the set-up failed"
     * where the log of that deployment alone has to say "and the connection would not close either".
     */
    @Test(timeOut = 60000)
    public void testTheSuppressedLinksOfAFailureSurviveItsRedaction() throws Exception {
        final String url = "jdbc:postgresql://opendj:S3cretOfTheBackend@127.0.0.1:5432/opendj";
        final SQLException setupFailure = new SQLException("no transaction on " + url, "08006", 1);
        setupFailure.addSuppressed(new SQLException("the connection would not close", "08003", 2));

        final SQLException reported = CachedConnection.reported(setupFailure, url);

        assertNoCredentials(reported);
        assertEquals(reported.getSuppressed().length, 1,
            "the rebuild of a redacted failure dropped what was suppressed on it");
        assertTrue(String.valueOf(reported.getSuppressed()[0].getMessage()).contains("would not close"),
            "the suppressed link of a redacted failure: " + reported.getSuppressed()[0]);
    }

    /**
     * A database that is not listening at all: every dialect reports it instead of retrying the
     * refused connect until the caller gives up on the operation.
     */
@@ -1288,12 +1332,20 @@
            "the session limit of an instance is cleared by a session ending");
    }

    /** A connection the setup of which failed belongs to nobody: it has to be closed, not leaked. */
    /**
     * A connection the setup of which failed belongs to nobody: it has to be closed, not leaked.
     * <p>
     * And a driver that will not close is said so on the failure being unwound rather than swallowed
     * (#929): the connection is gone either way, but a driver refusing to close is the shape of a
     * leak nobody would otherwise hear about, and the failure of the set-up is the one report this
     * attempt makes.
     */
    @Test(timeOut = 120000)
    public void testConnectionIsClosedWhenItsSetupFails() throws Exception {
        final String url = StubDriver.PREFIX + "setup-failure";
        final Connection broken = mock(Connection.class);
        doThrow(new SQLException("read only")).when(broken).setAutoCommit(false);
        doThrow(new SQLException("will not close")).when(broken).close();
        stub.answerWith(broken);

        try {
@@ -1301,6 +1353,9 @@
            fail("a connection that cannot be set up must be reported");
        } catch (SQLException expected) {
            assertEquals(expected.getMessage(), "read only");
            assertEquals(expected.getSuppressed().length, 1,
                "the failure of the close is not carried on the failure being unwound");
            assertEquals(expected.getSuppressed()[0].getMessage(), "will not close");
        }
        verify(broken).close();
    }
@@ -1311,6 +1366,7 @@
        final String url = StubDriver.PREFIX + "setup-unchecked";
        final Connection broken = mock(Connection.class);
        doThrow(new IllegalStateException("driver internal")).when(broken).setTransactionIsolation(anyInt());
        doThrow(new IllegalStateException("will not close")).when(broken).close();
        stub.answerWith(broken);

        try {
@@ -1318,6 +1374,11 @@
            fail("a connection that cannot be set up must be reported");
        } catch (IllegalStateException expected) {
            assertEquals(expected.getMessage(), "driver internal");
            // the unchecked failure of a close as well: it runs from the catch of a failure it must
            // not replace (JLS 14.20.2), which is the rule every close of this class keeps
            assertEquals(expected.getSuppressed().length, 1,
                "the failure of the close is not carried on the failure being unwound");
            assertEquals(expected.getSuppressed()[0].getMessage(), "will not close");
        }
        verify(broken).close();
    }
@@ -2293,6 +2354,58 @@
    }

    /**
     * The wait between two attempts of a connect: a millisecond, doubling to the ceiling and staying
     * there, so that a database refusing connections for a moment is asked again at once and one
     * refusing them for a minute is asked once a second rather than in a spin. Pinned here because
     * both loops that retry a connect of this backend wait on this schedule - the borrow of this
     * class and the catalog connect of {@code JDBCStorage.newCatalogConnection} - and a change to it
     * is a change to the pair (#929).
     */
    @Test(timeOut = 120000)
    public void testTheBackoffOfARetriedConnectDoublesToItsCeiling() {
        // the schedule itself and not a property of it: "grows until it reaches the ceiling" is
        // answered by every factor there is - a schedule tripling from 1 reaches 1000 in eight steps
        // and grows at every one of them - and the factor is what the two loops share
        long backoffMs = 0;
        for (final long expected : new long[] { 1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1000, 1000 }) {
            final long previous = backoffMs;
            backoffMs = CachedConnection.nextBackoffMs(backoffMs);
            assertEquals(backoffMs, expected,
                "the wait of a retry left the doubling schedule after " + previous + "ms");
        }
        assertEquals(backoffMs, CachedConnection.MAX_BACKOFF_MS, "the wait of a retry is not at its ceiling");
    }

    /**
     * The warning about a read bound a driver would not take is throttled per consequence and not
     * once for the JVM, so that one connection does not report for another.
     * <p>
     * The two connections this happens to do not share a fate: a connection of the pool is closed
     * rather than pooled, while the one catalog connection of a backend is kept and carries the
     * bound of its login for the rest of its life (#929). The pool meets the failure on every
     * connect and the catalog once per open of the backend, so a single timestamp has the pool
     * silence the line about the catalog connection - and the line that did come out says the
     * connection was closed, of a connection that is still being read from.
     */
    @Test(timeOut = 120000)
    public void testTheReadBoundWarningOfOneConnectionDoesNotSilenceAnother() {
        final long now = System.currentTimeMillis();
        final String pooled = "a connection of this pool of testTheReadBoundWarning";
        final String catalog = "the catalog connection of backend testTheReadBoundWarning";
        CachedConnection.lastReadBoundWarning.remove(pooled);
        CachedConnection.lastReadBoundWarning.remove(catalog);

        assertTrue(CachedConnection.readBoundWarningDue(pooled, now), "the first line about a connection was not due");
        assertTrue(CachedConnection.readBoundWarningDue(catalog, now),
            "a line about one connection silenced the line about another, which has no other line about it");

        assertFalse(CachedConnection.readBoundWarningDue(pooled, now + 1),
            "the same consequence was reported twice inside one interval");
        assertTrue(CachedConnection.readBoundWarningDue(pooled, now + CachedConnection.STALL_WARNING_INTERVAL_MS),
            "a connection carrying a bound it was never meant to keep was reported once and never again");
    }

    /**
     * The bound is configured in seconds and reaches the driver in milliseconds; 0, a negative
     * value and a value that is no number all leave a connection unbounded, which is what this
     * backend did before the property existed. A value past the ceiling of a socket read timeout is

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

@@ -93,6 +93,9 @@

    private ProbeDriver probeDriver;

    /** The standing read bound as this JVM was started with it, put back before every case. */
    private static final int CONFIGURED_READ_TIMEOUT_MILLIS = CachedConnection.readTimeoutMillis;

    @BeforeClass
    public void registerProbeDriver() throws SQLException {
        probeDriver = new ProbeDriver();
@@ -120,6 +123,13 @@
        probeDriver.attempts.set(0);
        probeDriver.refusalDelayMs = 0;
        probeDriver.interruptOnAttempt = false;
        // the same for the bound the connect reads off the class: a case that varies it and fails
        // before its finally would otherwise hand its value to whatever runs after it
        CachedConnection.readTimeoutMillis = CONFIGURED_READ_TIMEOUT_MILLIS;
        // and for the throttle the warning about a read bound that would not take is filed in: it is
        // a static of that class, so a case reading it would otherwise be reading whatever the case
        // before it left there
        CachedConnection.lastReadBoundWarning.clear();
    }

    private static JDBCStorage storageFor(String url) {
@@ -408,6 +418,10 @@
     * Both properties, which is what the pool itself says leaves a connect unbounded: the deadline of
     * the retry bounds the attempt where there is one, so turning the per-attempt bound off alone
     * leaves the attempt bounded by what is left of that deadline - the case above.
     * <p>
     * The standing read bound is pinned along with them, the way this class pins the two properties:
     * this connect reads {@link CachedConnection#READ_TIMEOUT_PROPERTY} as a pooled connect does, so
     * a JVM started with a value of it would have "unbounded" mean the bound of that value here.
     */
    @Test
    public void testTheCatalogConnectIsUnboundedWhereTheOperatorTurnedTheBoundOff() throws Exception {
@@ -417,6 +431,7 @@
        final String previousPool = System.getProperty(CachedConnection.POOL_TIMEOUT_PROPERTY);
        System.setProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY, "0");
        System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "0");
        CachedConnection.readTimeoutMillis = 0;
        try {
            probeDriver.lastProperties = null;
            final Connection con = storageFor(ProbeDriver.URL).newCatalogConnection(NO_REPLAY_WINDOW);
@@ -476,14 +491,25 @@

    /**
     * What the connection is handed back for: rows of its own, committed where they are written. The
     * read bound of the login is lifted as soon as the login is through (#872) - it is a bound of the
     * read bound of the login is gone as soon as the login is through (#872) - it is a bound of the
     * connect and not of the statements of the catalog - and the isolation is the pool's, a repeatable
     * read gap-locking a catalog two transactions enrol into.
     * <p>
     * The one {@code setNetworkTimeout} of this case is the standing read bound taking the place of
     * the bound of the login, at the 0 of a deployment that configured none: what the connection
     * carries afterwards is that value and not a lift followed by a second call. Pinned here for the
     * same reason the two bounds of the connect are handed in (#932) - the value is read off {@link
     * CachedConnection} on every connect, and a JVM started with {@link
     * CachedConnection#READ_TIMEOUT_PROPERTY} set would put its own number in this call.
     */
    @Test
    public void testTheCatalogConnectionIsSetUpForItsRows() throws Exception {
        // the read bound is lifted only where the attempt was given one, and the attempt takes the
        // shorter of the two bounds: a deadline of 0 leaves the per-attempt bound as the only one
        // ... and the bound the connection carries afterwards is the standing one of #885, handed in
        // here the way the other two are: a jvm started with the property set would otherwise put its
        // own number in the call this case reads
        CachedConnection.readTimeoutMillis = 0;
        final Connection con = storageFor(ProbeDriver.URL).newCatalogConnection(NO_REPLAY_WINDOW, DEFAULT_BOUND, 0);
        con.close();
        verify(con).setAutoCommit(false);
@@ -515,6 +541,129 @@
    }

    /**
     * ... and the line an operator meets says which connection that was and what became of it. The
     * establish of a connect is shared with the pool now (#929), and this is the whole of what is
     * not: a connection of the pool that would not take the bound is closed rather than pooled,
     * while this one is kept and carries the bound of its login for the rest of its life - an
     * operator reading "closed rather than pooled" about the connection the catalog of this backend
     * goes on using is being told the opposite of what happened.
     * <p>
     * Read off the throttle the warning is filed in rather than off the logger: the line itself goes
     * to a logger with a JVM-wide interval no test can wind back, while the key of the throttle is
     * the consequence that was reported - which is the thing this case is about.
     */
    @Test
    public void testTheReadBoundWarningOfTheCatalogNamesTheConnectionAndWhatBecomesOfIt() throws Exception {
        final String previous = System.getProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        final String previousPool = System.getProperty(CachedConnection.POOL_TIMEOUT_PROPERTY);
        System.setProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY, "30");
        System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "0");
        CachedConnection.readTimeoutMillis = 90000;
        final Connection keeping = mock(Connection.class);
        doThrow(new SQLFeatureNotSupportedException("no network timeout here"))
            .when(keeping).setNetworkTimeout(any(), anyInt());
        probeDriver.answer = keeping;
        try {
            storageFor(ProbeDriver.URL).newCatalogConnection(NO_REPLAY_WINDOW);

            final String reported = CachedConnection.readBoundConsequence(true, 30,
                JDBCStorage.catalogConnectionNamed("catalogProbe"), JDBCStorage.CATALOG_CONNECTION_FATE);
            assertTrue(CachedConnection.lastReadBoundWarning.containsKey(reported),
                "the catalog connect reported the read bound it could not lift as something else: "
                    + CachedConnection.lastReadBoundWarning.keySet());
            assertTrue(reported.contains("catalogProbe") && reported.contains("kept"),
                "the line names neither the backend whose catalog connection this is nor its fate: " + reported);
            assertFalse(CachedConnection.lastReadBoundWarning.containsKey(
                    CachedConnection.readBoundConsequence(true, 30, CachedConnection.POOLED_CONNECTION,
                        CachedConnection.POOLED_CONNECTION_FATE)),
                "the catalog connect reported the fate of a pooled connection, which is closed rather than kept");
        } finally {
            restore(previous);
            restorePool(previousPool);
        }
    }

    /**
     * The catalog connection carries the read bound a deployment asked for
     * ({@link CachedConnection#READ_TIMEOUT_PROPERTY}), exactly as a connection of the pool does.
     * <p>
     * Its statements are bounded by the class of the work they belong to, and nothing else on it is:
     * the {@code commit()} that writes a catalog row, the {@code rollback()} of a session given up
     * and the {@code close()} of one that lost the race have no bound of their own, so against a
     * database which stops answering after the login they wait for as long as the socket does. That
     * is the gap #885 closed for every connection of the pool - this one was written beside them,
     * one round before the property existed, and was left with the lift alone.
     */
    @Test
    public void testTheCatalogConnectionCarriesTheReadBoundAskedFor() throws Exception {
        final String previous = System.getProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        final String previousPool = System.getProperty(CachedConnection.POOL_TIMEOUT_PROPERTY);
        System.setProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY, "30");
        System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "0");
        CachedConnection.readTimeoutMillis = 90000;
        try {
            final Connection con = storageFor(ProbeDriver.URL).newCatalogConnection(NO_REPLAY_WINDOW);
            con.close();
            verify(con).setNetworkTimeout(any(), eq(90000));
            // and the bound of the login is gone with it: the value that replaces it is the whole of
            // what this connection carries, not a lift followed by a second call putting one back
            verify(con, never()).setNetworkTimeout(any(), eq(0));
        } finally {
            restore(previous);
            restorePool(previousPool);
        }
    }

    /**
     * And it carries it whether or not the login had a bound of its own to lift: the read bound of a
     * login is only ever set where the connect is bounded, so a deployment running with
     * {@code connect.timeout=0} - the setting that leaves a connect to the deadline of the retry
     * alone - would otherwise set this property and get nothing for it on this connection.
     */
    @Test
    public void testTheCatalogConnectionCarriesTheReadBoundWhereItsLoginHadNoneToLift() throws Exception {
        final String previous = System.getProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        final String previousPool = System.getProperty(CachedConnection.POOL_TIMEOUT_PROPERTY);
        System.setProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY, "0");
        System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "0");
        CachedConnection.readTimeoutMillis = 90000;
        try {
            final Connection con = storageFor(ProbeDriver.URL).newCatalogConnection(NO_REPLAY_WINDOW);
            con.close();
            verify(con).setNetworkTimeout(any(), eq(90000));
        } finally {
            restore(previous);
            restorePool(previousPool);
        }
    }

    /**
     * A read bound standing in the connection string is the deployment's own: it is not replaced by
     * the configured one here, exactly as it is not on a connection of the pool, and exactly as the
     * read bound of a login is not set on top of it. A guard rather than a regression test - the
     * bound is asked of {@link CachedConnection#standingReadBoundMillis}, which answers 0 for such a
     * url - and it is here because a bound put on from the value of the property alone would pass
     * every other case of this class.
     */
    @Test
    public void testAReadBoundOfTheUrlIsNotReplacedOnTheCatalogConnection() throws Exception {
        final String previous = System.getProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY);
        final String previousPool = System.getProperty(CachedConnection.POOL_TIMEOUT_PROPERTY);
        System.setProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY, "30");
        System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "0");
        CachedConnection.readTimeoutMillis = 90000;
        try {
            final Connection con = storageFor(ProbeDriver.URL + "?socketTimeout=30")
                .newCatalogConnection(NO_REPLAY_WINDOW);
            con.close();
            verify(con, never()).setNetworkTimeout(any(), anyInt());
        } finally {
            restore(previous);
            restorePool(previousPool);
        }
    }

    /**
     * A connection whose set-up failed is held by nobody - the caller is answered with the failure -
     * so it is closed here or it leaks for the life of the process, one per open of a storage.
     */

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

@@ -45,7 +45,9 @@
import java.util.concurrent.TimeUnit;

import static org.forgerock.opendj.config.ConfigurationMock.mockCfg;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotSame;
@@ -136,6 +138,32 @@
    }

    /**
     * A stamp connection whose set-up failed belongs to nobody and is closed, and a driver that will
     * not close is said so on the failure being unwound rather than swallowed - the rule the two
     * roads of {@code CachedConnection.establish()} keep (#929), and this connection is established
     * apart from them, with bounds of its own.
     */
    @Test
    public void testAStampConnectionWhoseSetUpFailsIsClosedAndSaysWhenItWillNotClose() throws Exception {
        final Connection broken = mock(Connection.class);
        doThrow(new SQLException("read only")).when(broken).setAutoCommit(false);
        doThrow(new SQLException("will not close")).when(broken).close();
        probeDriver.answer = broken;
        try {
            storageFor(ProbeDriver.URL).newStampConnection(JDBCStorage.Dialect.POSTGRES);
            fail("a stamp connection that cannot be set up must be reported");
        } catch (SQLException expected) {
            assertEquals(expected.getMessage(), "read only");
            assertEquals(expected.getSuppressed().length, 1,
                "the failure of the close is not carried on the failure being unwound");
            assertEquals(expected.getSuppressed()[0].getMessage(), "will not close");
        } finally {
            probeDriver.answer = null;
        }
        verify(broken).close();
    }

    /**
     * What the bounds are for: a database that keeps its established connections alive but accepts
     * no new ones - a moved vip, a proxy at its connection limit - usually completes the tcp
     * connect and then goes quiet, which leaves the driver in a read. Unbounded, that hangs the
@@ -316,12 +344,18 @@

        volatile Properties lastProperties;

        /** The connection to answer with, for the one case about what is done with a connection; a fresh mock otherwise. */
        volatile Connection answer;

        @Override
        public Connection connect(String url, Properties info) throws SQLException {
            if (!acceptsURL(url)) {
                return null; // not ours: DriverManager goes on to the next driver
            }
            lastProperties = info;
            if (answer != null) {
                return answer;
            }
            final Connection con = mock(Connection.class);
            when(con.createStatement()).thenReturn(mock(Statement.class));
            return con;

 opendj-server-legacy/src/test/java/org/opends/server/replication/InitOnLineTest.java

@@ -587,6 +587,92 @@
  }

  /**
   * A total update into this replica whose session stops before the DoneMsg arrives is a
   * failed import, and the generationId of the exporter does not stay over the part of its
   * data which arrived (issue #1039). Every road which stops the session - the restart a
   * failed replay or a configuration change asks for, the shutdown of the server - goes
   * through disableService(), and the import read the stopped broker as the end of the
   * stream: it kept the entries which had arrived, loaded the exporter's generationId from
   * the base entry among them, and completed its task as a success.
   */
  @Test(enabled=true)
  public void initializeImportSessionStoppedBeforeDone() throws Exception
  {
    String testCase = "initializeImportSessionStoppedBeforeDone";
    log("Starting " + testCase);
    try
    {
      replServer1 = createReplicationServer(replServer1ID, testCase);
      connectServer1ToReplServer(replServer1ID);
      server2 = openReplicationSession(baseDN,
          server2ID, 100, getReplServerPort(replServer1ID), 10000);

      // In S1 launch the total update, and S2 receives the request
      addTask(taskInitFromS2, ResultCode.SUCCESS, null);
      ReplicationMsg msg = server2.receive();
      Assertions.assertThat(msg).isInstanceOf(InitializeRequestMsg.class);

      // S2 announces every entry and sends two of them: the base entry, carrying the
      // generationId of S2 the way the base entry of a real export does, and one more.
      // The DoneMsg never comes.
      final long exporterGenerationId = 7777777L;
      final String baseEntry = updatedEntries[0].substring(0, updatedEntries[0].length() - 1)
          + "ds-sync-generation-id: " + exporterGenerationId + "\n\n";
      server2.publish(new InitializeTargetMsg(baseDN, server2ID, server1ID, server1ID,
          updatedEntries.length, initWindow));
      server2.publish(new EntryMsg(server2ID, server1ID, baseEntry.getBytes(), 1));
      server2.publish(new EntryMsg(server2ID, server1ID, updatedEntries[1].getBytes(), 2));

      // The import has read both and is waiting for the next one
      waitTaskLeft(taskInitFromS2, updatedEntries.length - 2);

      // The session stops under the import, the way every road to a stopped session does
      replDomain.disableService();

      waitTaskCompleted(taskInitFromS2, STOPPED_BY_ERROR, updatedEntries.length - 2, 2);
      // ...for the reason it failed, not for the lost connection every other early end reports
      final String reason =
          ERR_INIT_SESSION_STOPPED_DURING_IMPORT.get(baseDN, server1ID, server2ID).toString();
      Assertions.assertThat(getEntry(taskInitFromS2.getName(), 1000, true)
              .parseAttribute(ATTR_TASK_LOG_MESSAGES).asSetOfString())
          .as("the task does not report the stopped session as the reason its import failed")
          .anyMatch(record -> record.contains(reason));
      assertNotEquals(replDomain.getGenerationID(), exporterGenerationId,
          "the generationId of the exporter stayed over the part of its data which arrived");
      Entry base = getEntry(baseDN, 1000, true);
      assertNotEquals(base.parseAttribute("ds-sync-generation-id").asString(),
          String.valueOf(exporterGenerationId),
          "the generationId of the exporter stayed stored on the base entry of a cut import");

      log("Successfully ending " + testCase);
    }
    finally
    {
      afterTest(testCase);
    }
  }

  /** Waits until the task reports the given number of entries still to be imported. */
  private void waitTaskLeft(Entry taskEntry, long expectedLeft) throws Exception
  {
    final long deadline = System.currentTimeMillis() + 20000;
    String left;
    do
    {
      final SearchRequest request = newSearchRequest(taskEntry.getName(), SearchScope.BASE_OBJECT);
      Entry resultEntry = connection.processSearch(request).getSearchEntries().getFirst();
      left = resultEntry.parseAttribute(ATTR_TASK_INITIALIZE_LEFT).asString();
      if (String.valueOf(expectedLeft).equals(left))
      {
        return;
      }
      Thread.sleep(100);
    }
    while (System.currentTimeMillis() < deadline);
    fail("the import did not reach " + expectedLeft + " entries left within 20s, last read " + left);
  }

  /**
   * Tests the export side of the Initialize task
   * Test steps :
   * - add entries in S1, make S2 publish InitRequest

 opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ConfigChangeDuringImportTest.java

New file
@@ -0,0 +1,436 @@
/*
 * 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.replication.plugin;

import static java.nio.charset.StandardCharsets.*;
import static org.assertj.core.api.Assertions.*;
import static org.forgerock.opendj.ldap.ModificationType.*;
import static org.opends.messages.ReplicationMessages.*;
import static org.opends.server.TestCaseUtils.*;
import static org.opends.server.core.DirectoryServer.*;
import static org.opends.server.protocols.internal.InternalClientConnection.*;
import static org.testng.Assert.*;

import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.atomic.AtomicReference;

import org.forgerock.opendj.config.server.ConfigChangeResult;
import org.forgerock.opendj.ldap.DN;
import org.forgerock.opendj.ldap.ResultCode;
import org.forgerock.opendj.server.config.meta.ReplicationDomainCfgDefn.AssuredType;
import org.opends.server.TestCaseUtils;
import org.opends.server.core.ModifyOperation;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.common.AssuredMode;
import org.opends.server.replication.protocol.DoneMsg;
import org.opends.server.replication.protocol.EntryMsg;
import org.opends.server.replication.protocol.InitializeTargetMsg;
import org.opends.server.replication.server.ReplServerFakeConfiguration;
import org.opends.server.replication.server.ReplicationServer;
import org.opends.server.replication.service.ReplicationBroker;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;

/**
 * Tests a configuration change while this replica is the target of a total update.
 * <p>
 * The import of a total update streams over the session of the domain, on its listener
 * thread, and a configuration change which restarts that session for what it carries stops
 * the broker the import is reading (issue #1040). Through the server configuration the
 * change holds the lock of the configuration while it runs, and the restart waits for the
 * listener thread to end - which needs that lock to enable the backend back once the
 * stream ends: the change never returns. Reached below the configuration listeners, as the
 * change of an entry which was accepted before the import started reaches it, the restart
 * ends the import on the entries which had arrived.
 * <p>
 * The exporter is a broker of this test, so that the test says when the entries arrive:
 * the change is made while the import is waiting for them.
 */
@SuppressWarnings("javadoc")
public class ConfigChangeDuringImportTest extends ReplicationTestCase
{
  /**
   * The memory backend of {@code o=test} loses its data when it is disabled and enabled
   * back, which is what an import does to the backend it replaces: a total update needs a
   * backend which keeps what was imported into it.
   */
  private static final String EXAMPLE_DN = "dc=example,dc=com";
  private static final int RS_ID = 612;
  private static final int DS_ID = 1;
  private static final int EXPORTER_ID = 2;
  private static final int INIT_WINDOW = 100;
  private static final String DOMAIN_CONFIG_NAME = "config change during import test";
  private static final String IMPORTED_ENTRY_DN = "cn=imported,ou=People," + EXAMPLE_DN;
  /** How long a configuration change is given to return before it is read as hung. */
  private static final long CHANGE_TIMEOUT_IN_MS = 30_000;

  private DN baseDN;
  private int rsPort;
  private ReplicationServer replicationServer;
  private LDAPReplicationDomain domain;
  /** The entry the domain is configured in, when it is configured through the server. */
  private DN domainConfigDN;
  private ReplicationBroker exporter;

  @BeforeMethod
  public void setUpLocal() throws Exception
  {
    baseDN = DN.valueOf(EXAMPLE_DN);
    TestCaseUtils.clearBackend("userRoot", EXAMPLE_DN);

    rsPort = TestCaseUtils.findFreePort();
    replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
        rsPort, "configChangeDuringImportTestDb", 0, RS_ID, 0, 100, new TreeSet<String>()));
  }

  @AfterMethod(timeOut = 120_000)
  public void tearDown() throws Exception
  {
    try
    {
      stop(exporter);
      if (domainConfigDN != null)
      {
        // Deletes the "cn=external changelog" entry below it, and the domain, as well.
        deleteEntry(domainConfigDN);
        configEntriesToCleanup.remove(domainConfigDN);
        synchroServerEntry = null;
      }
      else if (domain != null)
      {
        MultimasterReplication.deleteDomain(baseDN);
      }
    }
    finally
    {
      exporter = null;
      domainConfigDN = null;
      domain = null;
      remove(replicationServer);
    }
  }

  /**
   * A change of the external changelog entry while the import streams must be refused, as
   * a change of the domain entry is.
   * <p>
   * The change is made the way {@code dsconfig} makes it, through the server configuration,
   * which holds the lock of the configuration for the length of it. Without the refusal the
   * attributes are applied and the session restarted for them: the restart stops the broker
   * the import is reading and waits for the listener thread, which ends the import on what
   * had arrived and then waits for the lock of the configuration to enable the backend back
   * - the change never returns, the backend stays deregistered, and every configuration
   * change of the server after it waits on the same lock.
   */
  @Test(timeOut = 180_000)
  public void aChangeOfTheExternalChangelogEntryIsRefusedWhileATotalUpdateRuns() throws Exception
  {
    configureDomainThroughTheServer();
    final String[] exported = exportedEntries();
    startImportInto(exported.length);

    final ModifyOperation change = changeConfigurationEntry(
        DN.valueOf("cn=external changelog," + domainConfigDN), "ds-cfg-ecl-include", "cn");

    assertEquals(change.getResultCode(), ResultCode.UNWILLING_TO_PERFORM,
        "a change of the external changelog entry was accepted while a total update ran: "
            + change.getErrorMessage());
    assertThat(change.getErrorMessage().toString())
        .as("the refusal does not say a total update is the reason")
        .contains(NOTE_ERR_CANNOT_CHANGE_CONFIG_DURING_TOTAL_UPDATE.get().toString());

    finishImport(exported);
    assertImported(exported);
    assertFalse(domain.getEclIncludes().contains("cn"),
        "the refused change of the attributes published to the external changelog was applied");
  }

  /**
   * A change of the domain entry while the import streams is refused, as it was before
   * this fix: the twin of the case above, pinned so that the two entries keep answering the
   * same thing.
   */
  @Test(timeOut = 180_000)
  public void aChangeOfTheDomainEntryIsRefusedWhileATotalUpdateRuns() throws Exception
  {
    configureDomainThroughTheServer();
    final String[] exported = exportedEntries();
    startImportInto(exported.length);

    final ModifyOperation change =
        changeConfigurationEntry(domainConfigDN, "ds-cfg-assured-type", "safe-read");

    assertEquals(change.getResultCode(), ResultCode.UNWILLING_TO_PERFORM,
        "a change of the domain entry was accepted while a total update ran: "
            + change.getErrorMessage());
    assertThat(change.getErrorMessage().toString())
        .as("the refusal does not say a total update is the reason")
        .contains(NOTE_ERR_CANNOT_CHANGE_CONFIG_DURING_TOTAL_UPDATE.get().toString());

    finishImport(exported);
    assertImported(exported);
    assertEquals(domain.getAssuredMode(), AssuredMode.SAFE_DATA_MODE,
        "the refused change of the assured configuration was applied");
  }

  /**
   * A change of the domain configuration which reaches the domain while the import streams
   * - one accepted before the import started - must leave the session to the import.
   * <p>
   * The assured configuration is negotiated as the session comes up, so the change asks for
   * a restart. The restart is refused and reported, the way it is for a domain disabled for
   * a total update: the configuration is stored, the import ends on the session it started
   * on, and the session the import brings up next negotiates what was stored.
   */
  @Test(timeOut = 180_000)
  public void aChangeOfTheDomainConfigurationLeavesTheSessionToTheImport() throws Exception
  {
    startDomain(domainCfg(AssuredType.NOT_ASSURED));
    final String[] exported = exportedEntries();
    startImportInto(exported.length);
    final Thread listener = listenerThread();
    assertNotNull(listener, "the import is running on no listener thread");

    final ConfigChangeResult ccr = domain.applyConfigurationChange(domainCfg(AssuredType.SAFE_READ));

    assertEquals(ccr.getResultCode(), ResultCode.SUCCESS, ccr.getMessages().toString());
    assertTrue(ccr.adminActionRequired(),
        "the change was reported as live although the session was not restarted for it");
    assertThat(ccr.getMessages().toString())
        .contains(NOTE_REPLICATION_DOMAIN_SESSION_NOT_RESTARTED.get(baseDN).toString());
    assertSame(listenerThread(), listener,
        "the session the import streams over was restarted for the change");

    finishImport(exported);
    assertImported(exported);
    assertEquals(domain.getAssuredMode(), AssuredMode.SAFE_READ_MODE,
        "the assured configuration was dropped although the change reported success");
  }

  /**
   * A change of the attributes published to the external changelog which reaches the domain
   * while the import streams must leave the session to the import: the attributes are
   * stored, and the session the import brings up next publishes them.
   */
  @Test(timeOut = 180_000)
  public void aChangeOfTheExternalChangelogAttributesLeavesTheSessionToTheImport() throws Exception
  {
    startDomain(domainCfg(AssuredType.NOT_ASSURED));
    final String[] exported = exportedEntries();
    startImportInto(exported.length);
    final Thread listener = listenerThread();
    assertNotNull(listener, "the import is running on no listener thread");

    final SortedSet<String> eclIncludes = new TreeSet<>();
    eclIncludes.add("cn");
    domain.changeConfig(eclIncludes, new TreeSet<String>());

    assertSame(listenerThread(), listener,
        "the session the import streams over was restarted for the change");

    finishImport(exported);
    assertImported(exported);
    assertTrue(domain.getEclIncludes().contains("cn"),
        "the attributes published to the external changelog were dropped");
  }

  /**
   * Configures the domain the way the server does, through its configuration entry: the
   * change listeners of that entry and of the "cn=external changelog" entry below it are
   * registered, and a change of either goes through the lock of the configuration.
   */
  private void configureDomainThroughTheServer() throws Exception
  {
    addSynchroServerEntry(
        "dn: cn=" + DOMAIN_CONFIG_NAME + ",cn=domains," + SYNCHRO_PLUGIN_DN + "\n"
        + "objectClass: top\n"
        + "objectClass: ds-cfg-replication-domain\n"
        + "cn: " + DOMAIN_CONFIG_NAME + "\n"
        + "ds-cfg-base-dn: " + EXAMPLE_DN + "\n"
        + "ds-cfg-replication-server: localhost:" + rsPort + "\n"
        + "ds-cfg-server-id: " + DS_ID + "\n"
        + "ds-cfg-assured-type: safe-data\n"
        + "ds-cfg-assured-sd-level: 1\n");
    domainConfigDN = synchroServerEntry.getName();
    assertTrue(getServerContext().getConfigurationHandler()
            .hasEntry(DN.valueOf("cn=external changelog," + domainConfigDN)),
        "the domain was configured without the external changelog entry this test changes");
    domain = MultimasterReplication.findDomain(baseDN, null);
    assertNotNull(domain, "the configuration entry created no domain");
    assertTrue(domain.isConnected(), "the domain did not connect to the replication server");
    exporter = openReplicationSession(baseDN, EXPORTER_ID, 100, rsPort, 10000);
  }

  private void startDomain(DomainFakeCfg cfg) throws Exception
  {
    domain = MultimasterReplication.createNewDomain(cfg);
    domain.start();
    assertTrue(domain.isConnected(), "the domain did not connect to the replication server");
    exporter = openReplicationSession(baseDN, EXPORTER_ID, 100, rsPort, 10000);
  }

  /**
   * A configuration which differs from the one the domain was started on by its assured
   * type alone: the broker properties are the same, so the change asks for no restart of
   * its own and what is left is the one the assured configuration asks for.
   */
  private DomainFakeCfg domainCfg(AssuredType assuredType)
  {
    final SortedSet<String> replServers = new TreeSet<>();
    replServers.add("localhost:" + rsPort);
    return new DomainFakeCfg(baseDN, DS_ID, replServers, assuredType, 1, -1, 1000, null);
  }

  /**
   * Changes a configuration entry the way {@code dsconfig} does, through the server
   * configuration, and fails when the change does not return: the restart it asks for
   * waits for the listener thread to end, which waits for the lock of the configuration the
   * change holds. The thread of the change is interrupted then, which gives up that wait
   * and lets the domain be taken down.
   */
  private ModifyOperation changeConfigurationEntry(DN entryDN, String attribute, String value)
      throws Exception
  {
    final AtomicReference<ModifyOperation> result = new AtomicReference<>();
    final Thread change = new Thread(new Runnable()
    {
      @Override
      public void run()
      {
        result.set(getRootConnection().processModify(
            modifyRequest(entryDN, REPLACE, attribute, value)));
      }
    }, "configuration change during the import");
    change.start();
    change.join(CHANGE_TIMEOUT_IN_MS);
    if (change.isAlive())
    {
      final String stacks = "\n" + stackOf(change) + "\n" + stackOf(listenerThread());
      change.interrupt();
      change.join(CHANGE_TIMEOUT_IN_MS);
      org.testng.Assert.fail("the change of " + entryDN + " did not return while the import ran:" + stacks);
    }
    return result.get();
  }

  private static String stackOf(Thread thread)
  {
    if (thread == null)
    {
      return "<no thread>";
    }
    final StringBuilder sb = new StringBuilder(thread.getName()).append(" [").append(thread.getState()).append("]\n");
    for (StackTraceElement frame : thread.getStackTrace())
    {
      sb.append("    at ").append(frame).append('\n');
    }
    return sb.toString();
  }

  /**
   * The listener thread of the domain is the one which says which session is running: the
   * import runs on it, and a restart of the session replaces it.
   */
  private Thread listenerThread()
  {
    final String name = "Replica DS(" + DS_ID + ") listener for domain \"" + baseDN + "\"";
    for (Thread thread : Thread.getAllStackTraces().keySet())
    {
      if (thread.getName().contains(name) && thread.isAlive())
      {
        return thread;
      }
    }
    return null;
  }

  /**
   * Has the exporter start a total update into this replica, and returns once the backend
   * of the domain is deregistered for it: from then on the import is reading the session.
   */
  private void startImportInto(int entryCount) throws Exception
  {
    exporter.publish(new InitializeTargetMsg(
        baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, entryCount, INIT_WINDOW));
    final long deadline = System.currentTimeMillis() + 30_000;
    while (getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN) != null)
    {
      assertTrue(System.currentTimeMillis() < deadline,
          "the import did not deregister the backend of the domain");
      Thread.sleep(20);
    }
  }

  /** Has the exporter send the entries of the total update, and waits for the import to end. */
  private void finishImport(String... ldifEntries) throws Exception
  {
    int msgId = 0;
    for (String ldif : ldifEntries)
    {
      exporter.publish(new EntryMsg(EXPORTER_ID, DS_ID, ldif.getBytes(UTF_8), ++msgId));
    }
    exporter.publish(new DoneMsg(EXPORTER_ID, DS_ID));
    final long deadline = System.currentTimeMillis() + 60_000;
    while (domain.ieRunning())
    {
      assertTrue(System.currentTimeMillis() < deadline, "the import did not end");
      Thread.sleep(50);
    }
  }

  private static void assertImported(String... ldifEntries) throws Exception
  {
    for (String ldif : ldifEntries)
    {
      final DN dn = dnOf(ldif);
      assertTrue(entryExists(dn), "the import ended before " + dn
          + " arrived: the session it streams over was stopped from under it");
    }
  }

  /** The data of the exporter: the base entry and two entries below it. */
  private static String[] exportedEntries()
  {
    return new String[] {
      "dn: " + EXAMPLE_DN + "\n"
          + "objectClass: top\n"
          + "objectClass: domain\n"
          + "dc: example\n"
          + "entryUUID: 31111111-1111-1111-1111-111111111111\n"
          + "\n",
      "dn: ou=People," + EXAMPLE_DN + "\n"
          + "objectClass: top\n"
          + "objectClass: organizationalUnit\n"
          + "ou: People\n"
          + "entryUUID: 31111111-1111-1111-1111-111111111112\n"
          + "\n",
      "dn: " + IMPORTED_ENTRY_DN + "\n"
          + "objectClass: top\n"
          + "objectClass: person\n"
          + "cn: imported\n"
          + "sn: imported\n"
          + "entryUUID: 31111111-1111-1111-1111-111111111113\n"
          + "\n",
    };
  }

  private static DN dnOf(String ldif)
  {
    return DN.valueOf(ldif.substring("dn: ".length(), ldif.indexOf('\n')));
  }
}

 opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/LDAPReplicationDomainConfigChangeTest.java

@@ -241,6 +241,58 @@
    }
  }

  /**
   * The fractional twin of the test above: the fractional configuration is what the session
   * filters the changes it receives on, and a change of it restarts the session. A domain
   * which owns its session is given no restart, and the configuration is applied all the
   * same - the session its enable() starts filters on it - and the change says a session
   * is waiting for it.
   */
  @Test
  public void fractionalConfigurationIsAppliedToADomainWhichOwnsItsSession() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    try
    {
      final SortedSet<String> replServers = unstartedReplicationServer();
      final LDAPReplicationDomain domain = startDomain(new DomainFakeCfg(baseDN, SERVER_ID, replServers));
      assertFalse(domain.getFractionalConfig().isFractional());

      // Disabled is what an online import leaves the domain: it owns its session, so this
      // change is applied to it without a session being restarted for it.
      domain.disable();
      waitForListenerThread(baseDN, false);

      final DomainFakeCfg fractional = new DomainFakeCfg(baseDN, SERVER_ID, replServers);
      // The same as the configuration in place, or the broker would restart the session for
      // the heartbeat interval and the change would no longer be the fractional one alone.
      fractional.setHeartbeatInterval(HEARTBEAT_INTERVAL_IN_MS);
      fractional.getFractionalExclude().add("*:description");
      final ConfigChangeResult ccr = domain.applyConfigurationChange(fractional);

      assertEquals(ccr.getResultCode(), ResultCode.SUCCESS, ccr.getMessages().toString());
      assertFalse(hasListenerThread(baseDN),
          "the change started a session on a domain which was disabled for a total update");
      final LDAPReplicationDomain.FractionalConfig applied = domain.getFractionalConfig();
      assertTrue(applied.isFractional(),
          "the fractional configuration was dropped although the change reported success");
      assertTrue(applied.isFractionalExclusive(),
          "the fractional configuration was applied in the wrong mode");
      assertTrue(applied.getFractionalAllClassesAttributes().contains("description"),
          "the attribute the change excludes was dropped: " + applied.getFractionalAllClassesAttributes());
      assertTrue(ccr.adminActionRequired(),
          "the fractional configuration is what a session filters on, and this domain was"
              + " given no session to filter over");

      // Left as a total update leaves it: enabled back, on the configuration it was given.
      domain.enable();
    }
    finally
    {
      MultimasterReplication.deleteDomain(baseDN);
    }
  }

  @Test
  public void changeIsRefusedWhenTheExternalChangelogDomainRejectsIt() throws Exception
  {

 opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartTest.java

@@ -15,32 +15,52 @@
 */
package org.opends.server.replication.plugin;

import static java.util.concurrent.TimeUnit.*;
import static org.assertj.core.api.Assertions.*;
import static org.opends.server.TestCaseUtils.*;
import static org.opends.server.replication.plugin.LDAPReplicationDomain.*;
import static org.opends.server.util.CollectionUtils.*;
import static org.testng.Assert.*;

import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.atomic.AtomicBoolean;

import org.forgerock.opendj.config.server.ConfigChangeResult;
import org.forgerock.opendj.ldap.DN;
import org.forgerock.opendj.ldap.ResultCode;
import org.opends.server.TestCaseUtils;
import org.opends.server.plugins.ShortCircuitPlugin;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.common.CSNGenerator;
import org.opends.server.replication.protocol.DeleteMsg;
import org.opends.server.replication.server.DataServerHandler;
import org.opends.server.replication.server.ReplServerFakeConfiguration;
import org.opends.server.replication.server.ReplicationServer;
import org.opends.server.replication.service.ReplicationBroker;
import org.opends.server.replication.service.ReplicationDomain;
import org.opends.server.types.Entry;
import org.opends.server.types.OperationType;
import org.opends.server.util.TestTimer;
import org.opends.server.util.TestTimer.CallableVoid;
import org.testng.annotations.Test;

/**
 * Tests that a configuration change does not start the session of a domain which stopped
 * its own session: a domain which is shutting down, or whose data is being replaced, owns
 * its session and is the one which brings it back.
 * Tests which sessions a configuration change restarts. A domain which stopped its own
 * session - one which is shutting down, or whose data is being replaced - owns it and is the
 * one which brings it back; one which is running its session is given a new one, and the
 * replication server hears the change over it.
 */
@SuppressWarnings("javadoc")
public class SessionRestartTest extends ReplicationTestCase
{
  private static final int RS_ID = 601;
  private static final int DS_ID = 1;
  /** The replica whose changes the domain under test replays. */
  private static final int PUBLISHER_ID = 2;
  private static final int GROUP_ID = 1;

  @Test
@@ -63,10 +83,12 @@
      domain.disable();
      assertFalse(domain.isConnected());

      changeEclIncludes(domain, domainCfg);
      final ConfigChangeResult ccr = changeEclIncludes(domain, domainCfg);

      assertFalse(domain.isConnected(),
          "a configuration change started the session of a disabled domain");
      // the restart the change asked for was refused, and said so rather than reported as applied
      assertTrue(ccr.adminActionRequired(), "the refused restart was reported as fully applied");
    }
    finally
    {
@@ -97,10 +119,12 @@
      domain.shutdown();
      assertFalse(domain.isConnected());

      changeEclIncludes(domain, domainCfg);
      final ConfigChangeResult ccr = changeEclIncludes(domain, domainCfg);

      assertFalse(domain.isConnected(),
          "a configuration change started the session of a domain which has shut down");
      // the restart the change asked for was refused, and said so rather than reported as applied
      assertTrue(ccr.adminActionRequired(), "the refused restart was reported as fully applied");
    }
    finally
    {
@@ -113,11 +137,191 @@
  }

  /**
   * The twin of the two above: a domain which is running its session is given a new one by
   * the change, and the replication server is what tells. The attributes the external
   * changelog includes are stored in the domain before the session is restarted for them,
   * so the domain says the change is applied whether or not the restart ran; the
   * replication server hears the list once, in the {@code StartSessionMsg} of a session, and
   * a {@code DataServerHandler} which carries the new list is a session started after the
   * change.
   */
  @Test
  public void aConfigurationChangeRestartsTheSessionOfALiveDomain() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    ReplicationServer replicationServer = null;
    LDAPReplicationDomain domain = null;
    try
    {
      final int rsPort = TestCaseUtils.findFreePort();
      replicationServer = createReplicationServer(rsPort, "sessionRestartTestLiveDb");

      final DomainFakeCfg domainCfg = newDomainCfg(baseDN, rsPort);
      domain = MultimasterReplication.createNewDomain(domainCfg);
      domain.start();
      assertTrue(domain.isConnected());
      // What the session which is running told the replication server: no attribute at all.
      assertThat(eclIncludesHeardBy(replicationServer, baseDN)).doesNotContain("cn");

      final ConfigChangeResult ccr = changeEclIncludes(domain, domainCfg);

      // the restart was run, so there is nothing to tell the administrator to wait for
      assertFalse(ccr.adminActionRequired(),
          "a restart which was run was reported as refused: " + ccr.getMessages());
      final ReplicationServer rs = replicationServer;
      new TestTimer.Builder().maxSleep(5, SECONDS).sleepTimes(100, MILLISECONDS).toTimer()
          .repeatUntilSuccess(new CallableVoid()
      {
        @Override
        public void call() throws Exception
        {
          assertThat(eclIncludesHeardBy(rs, baseDN))
              .as("the replication server was never told the new list, so no session was"
                  + " started after the change")
              .contains("cn");
        }
      });
    }
    finally
    {
      if (domain != null)
      {
        MultimasterReplication.deleteDomain(baseDN);
      }
      remove(replicationServer);
    }
  }

  /**
   * A replay thread which could not apply a change stops the session so that the change
   * is delivered again, and waits out a backoff before it starts the session back. The
   * session it stopped is its claim; a configuration change which stops and starts the
   * session while it waits leaves that claim stale, and the thread which comes back from
   * its wait leaves the session which replaced the one it stopped alone.
   * <p>
   * What a stale claim which is not declined does to the session is nothing today: the
   * broker and the listener thread are up already, and starting them again is a no-op. The
   * one thing which tells a declined claim from one which was acted on is the generation of
   * the session, which counts every start: the thread which acted on its stale claim leaves
   * it one past where the restart which replaced its session left it.
   */
  @Test
  public void aReplayThreadLeavesAloneTheSessionWhichReplacedTheOneItStopped() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    ReplicationServer replicationServer = null;
    LDAPReplicationDomain domain = null;
    ReplicationBroker publisher = null;
    try
    {
      final int rsPort = TestCaseUtils.findFreePort();
      replicationServer = createReplicationServer(rsPort, "sessionRestartTestStaleClaimDb");
      domain = MultimasterReplication.createNewDomain(newDomainCfg(baseDN, rsPort));
      domain.start();
      assertTrue(domain.isConnected());

      final Entry entry = TestCaseUtils.addEntry(
          "dn: cn=stale claim," + baseDN,
          "objectClass: top",
          "objectClass: person",
          "cn: stale claim",
          "sn: claim");
      final String uuid = getEntry(entry.getName(), 1, true).parseAttribute("entryuuid").asString();
      publisher = openReplicationSession(baseDN, PUBLISHER_ID, 100, rsPort, 1000);

      final LDAPReplicationDomain replica = domain;
      final Object serviceStateLock = serviceStateLockOf(replica);
      try
      {
        /*
         * The backend refuses the delete for longer than the replay is retried in place,
         * so the replay thread stops the session for the change to be delivered again -
         * and serves it once the session was restarted for it, so that the change is
         * applied over the session started next and asks for no restart of its own.
         */
        ShortCircuitPlugin.registerShortCircuit(OperationType.DELETE, "PreParse",
            ResultCode.UNAVAILABLE.intValue(), IN_PLACE_REPLAY_ATTEMPTS + 2);
        publisher.publish(new DeleteMsg(entry.getName(), new CSNGenerator(PUBLISHER_ID, 0).newCSN(), uuid));

        /*
         * The lock is taken and let go until it is taken while the session is stopped: the
         * replay thread stopped it under the lock and let the lock go for its wait, so the
         * claim it holds is standing, and the lock is held for the rest of that wait. The
         * restart below runs while the claim is standing however short the wait is.
         */
        final long claim;
        final long replaced;
        final long deadline = System.currentTimeMillis() + SECONDS.toMillis(30);
        while (true)
        {
          synchronized (serviceStateLock)
          {
            if (!replica.isConnected())
            {
              claim = sessionGenerationOf(replica);
              // Something else restarts the session while the replay thread waits: the
              // attributes the external changelog includes are changed.
              replica.changeConfig(newTreeSet("cn"), new TreeSet<String>());
              replaced = sessionGenerationOf(replica);
              break;
            }
          }
          assertTrue(System.currentTimeMillis() < deadline, "the failed replay never stopped the session");
          Thread.sleep(5);
        }
        assertNotEquals(replaced, claim, "the restart left the generation of the session where it was");
        assertTrue(replica.isConnected(), "the restart left the domain without a session");

        // The change is delivered again over the session which replaced the stopped one,
        // and applied; the replay thread comes back from its wait and finds its claim stale.
        assertNull(getEntry(entry.getName(), 30000, false),
            "the change the session was stopped for was not delivered again over the session which replaced it");
        new TestTimer.Builder().maxSleep(30, SECONDS).sleepTimes(100, MILLISECONDS).toTimer()
            .repeatUntilSuccess(new CallableVoid()
        {
          @Override
          public void call() throws Exception
          {
            assertFalse(isRecoveringFromAReplayFailure(replica),
                "the replay thread never came back from the restart it asked for");
          }
        });

        synchronized (serviceStateLock)
        {
          assertEquals(sessionGenerationOf(replica), replaced,
              "the replay thread started the session which replaced the one it stopped,"
                  + " as if its claim on the stopped one were still current");
        }
      }
      finally
      {
        ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse");
      }
    }
    finally
    {
      if (publisher != null)
      {
        publisher.stop();
      }
      if (domain != null)
      {
        MultimasterReplication.deleteDomain(baseDN);
      }
      remove(replicationServer);
    }
  }

  /**
   * Applies a configuration change which changes the attributes the external changelog
   * includes. The domain hands it to {@code ExternalChangelogDomain}, which asks for the
   * session to be restarted so that the replication server hears the new list.
   *
   * @return the result of the change, so that each test says whether the restart it asked
   *         for was to be run or refused
   */
  private void changeEclIncludes(LDAPReplicationDomain domain, DomainFakeCfg domainCfg)
  private ConfigChangeResult changeEclIncludes(LDAPReplicationDomain domain, DomainFakeCfg domainCfg)
      throws Exception
  {
    final SortedSet<String> eclIncludes = new TreeSet<>();
@@ -127,10 +331,49 @@

    final ConfigChangeResult ccr = domain.applyConfigurationChange(domainCfg);
    assertEquals(ccr.getResultCode(), ResultCode.SUCCESS, ccr.getMessages().toString());
    // the restart the change asked for was refused, and said so rather than reported as applied
    assertTrue(ccr.adminActionRequired(), "the refused restart was reported as fully applied");
    // the change did reach the external changelog configuration of the domain
    assertThat(domain.getEclIncludes()).contains("cn");
    return ccr;
  }

  /**
   * The attributes the replication server was told the domain includes in the external
   * changelog, by the session it holds for the domain right now.
   */
  private static Set<String> eclIncludesHeardBy(ReplicationServer replicationServer, DN baseDN)
  {
    final DataServerHandler ds =
        replicationServer.getReplicationServerDomain(baseDN).getConnectedDSs().get(DS_ID);
    assertNotNull(ds, "the replication server holds no session of the domain");
    return ds.toDSInfo().getEclIncludes();
  }

  private static Object serviceStateLockOf(LDAPReplicationDomain domain) throws Exception
  {
    // Declared where the session lives, next to disableService()/enableService()
    final Field serviceStateLock = ReplicationDomain.class.getDeclaredField("serviceStateLock");
    serviceStateLock.setAccessible(true);
    return serviceStateLock.get(domain);
  }

  /** Read under {@code serviceStateLock}, as {@code getSessionGeneration()} asks. */
  private static long sessionGenerationOf(LDAPReplicationDomain domain) throws Exception
  {
    final Method getSessionGeneration = ReplicationDomain.class.getDeclaredMethod("getSessionGeneration");
    getSessionGeneration.setAccessible(true);
    return (Long) getSessionGeneration.invoke(domain);
  }

  /**
   * Whether a replay thread of the domain is restarting the session for a change it could
   * not apply: set by the thread which took that recovery on, and cleared once the restart
   * it asked for is done with - run, or declined on a stale claim.
   */
  private static boolean isRecoveringFromAReplayFailure(LDAPReplicationDomain domain) throws Exception
  {
    final Field replayFailureRecovery = LDAPReplicationDomain.class.getDeclaredField("replayFailureRecovery");
    replayFailureRecovery.setAccessible(true);
    return ((AtomicBoolean) replayFailureRecovery.get(domain)).get();
  }

  private DomainFakeCfg newDomainCfg(DN baseDN, int rsPort)

 opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java

@@ -238,6 +238,85 @@
  }

  /**
   * The session generation is the identity of a session: a replay thread which stopped the
   * session and let the lock go compares the generation it read then with the one it reads
   * when it comes back, and starts the session only if the two are the same. Both halves
   * of a restart have to move it, or a session stopped or started by something else in the
   * meantime would look like the one that thread stopped.
   */
  @Test
  public void everyStopAndEveryStartOfTheSessionIsCounted() throws Exception
  {
    final DN testService = DN.valueOf("o=test");
    ReplicationServer replServer = null;
    FakeReplicationDomain domain = null;
    try
    {
      final int replServerPort = TestCaseUtils.findFreePort();
      replServer = createReplicationServer(1, replServerPort, "ReplicationDomainTestDbGeneration", 100);
      domain = new FakeReplicationDomain(testService, 2, newTreeSet("localhost:" + replServerPort), 1000, 1);

      final long live = sessionGenerationOf(domain);
      domain.disableService();
      final long stopped = sessionGenerationOf(domain);
      assertNotEquals(stopped, live, "stopping the session left its generation where it was");

      domain.enableService();
      final long started = sessionGenerationOf(domain);
      assertNotEquals(started, stopped, "starting the session left its generation where it was");
    }
    finally
    {
      disable(domain);
      remove(replServer);
    }
  }

  /**
   * What {@code restartSession()} of the LDAP domain does, minus the wait: the session is
   * stopped, the generation is read as the claim on it, and the lock is let go. A
   * {@link ReplicationDomain#restartService()} run by something else in the meantime - a
   * configuration change - has to leave that claim stale, or the thread which comes back
   * from its wait could not tell the session it stopped from the one which replaced it.
   */
  @Test
  public void aClaimOnAStoppedSessionIsStaleOnceSomethingElseRestartedIt() throws Exception
  {
    final DN testService = DN.valueOf("o=test");
    ReplicationServer replServer = null;
    FakeReplicationDomain domain = null;
    try
    {
      final int replServerPort = TestCaseUtils.findFreePort();
      replServer = createReplicationServer(1, replServerPort, "ReplicationDomainTestDbStaleClaim", 100);
      domain = new FakeReplicationDomain(testService, 2, newTreeSet("localhost:" + replServerPort), 1000, 1);

      domain.disableService();
      final long claim = sessionGenerationOf(domain);

      domain.restartService();

      assertNotEquals(sessionGenerationOf(domain), claim,
          "a restart of the session by something else left the generation where it was,"
              + " so the claim of the thread which stopped it still looks current");
    }
    finally
    {
      disable(domain);
      remove(replServer);
    }
  }

  /** Read under the lock, as {@link ReplicationDomain#getSessionGeneration()} asks. */
  private static long sessionGenerationOf(ReplicationDomain domain)
  {
    synchronized (domain.serviceStateLock)
    {
      return domain.getSessionGeneration();
    }
  }

  /**
   * Publish performance test.
   * The test loops calling the publish methods of the ReplicationDomain.
   * It should not be enabled by default as it will use a lot of time.