external-software/github_OpenIdentityPlatform_OpenDJ.git

			@@ -46,9 +46,9 @@
			import java.util.LinkedList;
			import java.util.List;
			import java.util.Map;
			import java.util.NavigableMap;
			import java.util.NoSuchElementException;
			import java.util.Set;
			import java.util.SortedMap;
			import java.util.StringTokenizer;
			import java.util.TreeMap;
			import java.util.concurrent.BlockingQueue;
			@@ -57,8 +57,8 @@
			import java.util.concurrent.atomic.AtomicBoolean;
			import java.util.concurrent.atomic.AtomicInteger;
			import java.util.concurrent.atomic.AtomicLong;
			import java.util.concurrent.atomic.AtomicReference;
			import java.util.concurrent.locks.ReentrantReadWriteLock;
			import java.util.function.Consumer;

			import net.jcip.annotations.GuardedBy;

			@@ -129,6 +129,7 @@
			import org.opends.server.replication.protocol.ModifyMsg;
			import org.opends.server.replication.protocol.OperationContext;
			import org.opends.server.replication.protocol.RoutableMsg;
			import org.opends.server.replication.protocol.Session;
			import org.opends.server.replication.protocol.UpdateMsg;
			import org.opends.server.replication.service.DSRSShutdownSync;
			import org.opends.server.replication.service.ReplicationBroker;
			@@ -221,11 +222,14 @@
			{
			private final CSN startCSN;
			private final CSN endCSN;
			/** What the search found, for the round which runs it alone: see {@link #replayOperations}. */
			private final NavigableMap<CSN, FakeOperation> found;

			public ScanSearchListener(CSN startCSN, CSN endCSN)
			public ScanSearchListener(CSN startCSN, CSN endCSN, NavigableMap<CSN, FakeOperation> found)
			{
			this.startCSN = startCSN;
			this.endCSN = endCSN;
			this.found = found;
			}

			@Override
			@@ -234,19 +238,22 @@
			throws DirectoryException
			{
			// Build the list of Operations that happened on this entry after startCSN
			// and before endCSN and add them to the replayOperations list
			// and up to endCSN and add them to the list of the round.
			// CSNs are ordered by time first, so the changes of the other replicas which this one
			// replayed fall in the range as well: they are left out, it is not for this replica to
			// publish them.
			// The end is part of the range, as it is of the search: the next range starts after it, so
			// a change numbered exactly the end would be in neither.
			Iterable<FakeOperation> updates =
			EntryHistorical.generateFakeOperations(searchEntry);

			for (FakeOperation op : updates)
			{
			CSN csn = op.getCSN();
			if (csn.isNewerThan(startCSN) && csn.isOlderThan(endCSN))
			if (csn.getServerId() == getServerId()
			&& csn.isNewerThan(startCSN) && !csn.isNewerThan(endCSN))
			{
			synchronized (replayOperations)
			{
			replayOperations.put(csn, op);
			}
			found.put(csn, op);
			}
			}
			}
			@@ -564,7 +571,25 @@
			* replication server and that the ServerState is not updated too early.
			*/
			private final PendingChanges pendingChanges;
			private final AtomicReference<RSUpdater> rsUpdater = new AtomicReference<>(null);
			/**
			* Guards {@link #rsUpdater} and the start a session hands over to it, and is held as a recovery
			* starts, ends, and as its publisher leaves: a session which needs a recovery either hands its
			* start to the publisher which runs - which then neither ends the recovery nor leaves before it
			* has run from it - or finds none and starts one. Nothing is done under it which takes another
			* lock: a session starts under the lock of the connect phase of the broker, and a recovery ends
			* under the lock of the pending changes.
			*/
			private final Object recoveryLock = new Object();
			/** The missing change publisher which runs, if any: under {@link #recoveryLock}. */
			private RSUpdater rsUpdater;
			/**
			* How many missing change publishers in a row failed - their search did not go through, or they
			* threw - since a recovery last ended: the wait before the session restart each of them asks for
			* is computed from it. It is a count of its own rather than the one of the session restarts a
			* failed replay asks for, which a change replayed resets: the replication of the other replicas
			* into this one goes on while a recovery keeps failing.
			*/
			private final AtomicInteger consecutiveRecoveryFailures = new AtomicInteger();

			/**
			* It contain the updates that were done on other servers, transmitted by the
			@@ -600,10 +625,43 @@
			private volatile boolean importingData;

			/**
			* This list is used to temporary store operations that needs to be replayed
			* at session establishment time.
			* The local deletes, kept aside for the recoveries: a deleted entry leaves no historical
			* information to publish the delete again from.
			* <p>
			* A delete stays here once it is published, and once a recovery is over: the replication server
			* of a later session may miss it. The list holds the last {@link #MAX_DELETES_KEPT_ASIDE} deletes,
			* and a delete which no replication server has taken yet is not dropped without a trace. What the
			* search of a round of a recovery finds is the round's own, so that it neither takes room here
			* nor stays after the round.
			*/
			private final SortedMap<CSN, FakeOperation> replayOperations = new TreeMap<>();
			private final NavigableMap<CSN, FakeDelOperation> replayOperations = new TreeMap<>();
			/** How many local deletes {@link #replayOperations} keeps at most. */
			private static final int MAX_DELETES_KEPT_ASIDE = 10000;
			/** How many local deletes {@link #replayOperations} keeps at most: fewer in the tests which fill it. */
			private volatile int maxDeletesKeptAside = MAX_DELETES_KEPT_ASIDE;

			/**
			* Run by {@link #buildAndPublishMissingChanges(CSN, ReplicationBroker)} after the search of
			* each of its rounds, before it chooses what the round publishes. Only there for the tests,
			* which do local changes there - a change done while a round searches is one the search may or
			* may not have seen, and nothing else puts a change in that gap on cue - or fail the search,
			* which a local backend which works does not do on cue either.
			*/
			private volatile Consumer<InternalSearchOperation> missingChangesSearchHook;
			/**
			* Run by the missing change publisher after each of its runs, before it takes the start a session
			* handed over or leaves. Only there for the tests which start a session at that point - once a run
			* has ended the recovery and before its publisher has left - where nothing else holds it on cue.
			*/
			private volatile Runnable missingChangesPublisherRunHook;
			/**
			* Run by a round of a recovery which publishes the local changes the recovery kept, once it has
			* published them and before it ends the recovery, handed whether the round holds the local changes
			* back. Only there for the tests which do local changes at that point, or start a session there -
			* one which hands its start over while the round holds the local changes back - where nothing else
			* holds the round on cue: it searches nothing.
			*/
			private volatile Consumer<Boolean> keptChangesPublishedHook;

			/** Published by a configuration change, read by the changelog threads without a lock. */
			private volatile ExternalChangelogDomain eclDomain;
			@@ -790,18 +848,56 @@
			}
			}

			/** Where a missing change publisher starts, and the session it publishes on. */
			private static final class RecoveryStart
			{
			/** The newest change of this replica the replication server of the session holds, or less. */
			private final CSN csn;
			/** The session the publisher publishes on: its broker refuses the messages for another one. */
			private final Session session;

			private RecoveryStart(CSN csn, Session session)
			{
			this.csn = csn;
			this.session = session;
			}
			}

			/**
			* The thread that is responsible to update the RS to which this domain is
			* connected in case it is late and there is no RS which is up to date.
			* <p>
			* One runs at a time. A session which starts while it runs hands it its start, and the publisher
			* goes back there and publishes on that session from it: its broker refuses what it publishes for
			* the session before, and the replication server of the new one - which may hold less than the one
			* before - gets each change it misses, in order. The recovery is not ended while a start is handed
			* over, and the publisher does not leave before it has run from it.
			*/
			private class RSUpdater extends DirectoryThread
			{
			private final CSN startCSN;
			/** Where the next run of this publisher starts: under {@link #recoveryLock}, {@code null} once taken. */
			private RecoveryStart nextStart;

			protected RSUpdater(CSN replServerMaxCSN)
			protected RSUpdater(RecoveryStart start)
			{
			super("Replica DS(" + getServerId() + ") missing change publisher for domain \"" + getBaseDN() + "\"");
			this.startCSN = replServerMaxCSN;
			this.nextStart = start;
			}

			/**
			* Hands over the start of a session which started while this publisher runs, and ends the wait of
			* a failed run for the session restart: this session is the next one. Under {@link #recoveryLock}.
			*/
			private void handOver(RecoveryStart start)
			{
			nextStart = start;
			recoveryLock.notifyAll();
			}

			/** Whether a session which started while this publisher runs handed it its start: under {@link #recoveryLock}. */
			private boolean hasAStartHandedOver()
			{
			return nextStart != null;
			}

			@Override
			@@ -811,49 +907,145 @@
			// let's send them to him.
			logger.trace(DEBUG_GOING_TO_SEARCH_FOR_CHANGES);

			/*
			* Get all the changes that have not been seen by this
			* replication server and publish them.
			*/
			boolean startTheSessionAgain = false;
			try
			{
			if (buildAndPublishMissingChanges(startCSN, broker))
			for (RecoveryStart start = takeStart(); start != null; start = takeStart())
			{
			logger.trace(DEBUG_CHANGES_SENT);
			synchronized(replayOperations)
			startTheSessionAgain = false;
			try
			{
			replayOperations.clear();
			/*
			* Get all the changes that have not been seen by this
			* replication server and publish them.
			* The deletes kept aside are not dropped once the recovery is over: a later session may
			* reach a replication server which misses some of them, and they have nothing else to be
			* published from.
			*/
			if (buildAndPublishMissingChanges(start.csn, broker, start.session, this))
			{
			logger.trace(DEBUG_CHANGES_SENT);
			consecutiveRecoveryFailures.set(0);
			}
			// Otherwise the broker refused a change - it reaches no replication server, its session
			// was replaced, or it was stopped - and the publisher runs from the start the session
			// which replaced it handed over, or the recovery of the session it opens next publishes
			// the change; or this domain is going away.
			}
			catch (Exception e)
			{
			/*
			* The search of a round did not go through, or the round threw. Nothing it did not
			* publish is lost - a round whose search failed publishes nothing, and the deletes kept
			* aside and the local changes the recovery keeps stay where they are - and the recovery
			* is not over: the local changes are still left to it. No recovery of this session would
			* publish them, as a session which stays up starts no other, so the session is started
			* again, and the recovery of the next one publishes them. A domain which is going away
			* needs none, and one which is disabled starts a session of its own when it is enabled
			* back.
			*
			* The backend is left some time to recover first, with the session up meanwhile, so that
			* the replication of the other replicas into this one goes on: the restart is asked for
			* once the wait is over, and not with the backoff of the restart a failed replay asks
			* for, which stops the session before it waits. This publisher waits registered: a
			* session which comes meanwhile hands it its start, which ends the wait, and is the next
			* session the publisher runs from, with no restart.
			*/
			if (!shutdown.get() && !disabled)
			{
			final int failures = consecutiveRecoveryFailures.incrementAndGet();
			logger.error(ERR_MISSING_CHANGES_NOT_PUBLISHED, getBaseDN(),
			e instanceof MissingChangesSearchFailedException ? e.getMessage() : stackTraceToSingleLineString(e),
			sessionRestartBackoffInMs(failures));
			waitForAStartOrTheBackoff(failures);
			startTheSessionAgain = true;
			}
			}
			final Runnable runHook = missingChangesPublisherRunHook;
			if (runHook != null)
			{
			runHook.run();
			}
			}
			else
			{
			/*
			* An error happened trying to search for the updates
			* This server will start accepting again new updates but
			* some inconsistencies will stay between servers.
			* Log an error for the repair tool
			* that will need to re-synchronize the servers.
			*/
			logger.error(ERR_CANNOT_RECOVER_CHANGES, getBaseDN());
			}
			}
			catch (Exception e)
			{
			/*
			* An error happened trying to search for the updates
			* This server will start accepting again new updates but
			* some inconsistencies will stay between servers.
			* Log an error for the repair tool
			* that will need to re-synchronize the servers.
			*/
			logger.error(ERR_CANNOT_RECOVER_CHANGES, getBaseDN());
			}
			finally
			{
			broker.setRecoveryRequired(false);
			// RSUpdater thread has finished its work, let's remove it from memory
			// so another RSUpdater thread can be started if needed.
			rsUpdater.compareAndSet(this, null);
			/*
			* Left without taking a start only on an error, which goes on up from here: the recovery is
			* not over, and no session would start another - the start a session handed over meanwhile
			* included - so the session is started again, as after a run which failed.
			*/
			synchronized (recoveryLock)
			{
			if (rsUpdater == this)
			{
			rsUpdater = null;
			broker.setRecoveryRequired(false);
			startTheSessionAgain = true;
			}
			}
			if (startTheSessionAgain && !shutdown.get() && !disabled)
			{
			/*
			* Asked for once this publisher is gone, so that the session the restart brings up starts a
			* publisher of its own. The state checkpointer runs it within a second, or once a total
			* update is over. The domain is checked again, as it may have been disabled or shut down
			* during the wait.
			*/
			sessionRestarts.request(SessionRestart.NOW);
			}
			}
			}

			/**
			* Waits before the session restart a failed run asks for: one second more for each failure in a
			* row, up to ten. A session which hands its start over meanwhile ends the wait, as it is the next
			* session; so does the domain as it is shut down or disabled.
			*/
			private void waitForAStartOrTheBackoff(int failures)
			{
			final long until = monotonicNowInMs() + sessionRestartBackoffInMs(failures);
			synchronized (recoveryLock)
			{
			for (long left = until - monotonicNowInMs();
			left > 0 && !hasAStartHandedOver() && !shutdown.get() && !disabled;
			left = until - monotonicNowInMs())
			{
			try
			{
			recoveryLock.wait(left);
			}
			catch (InterruptedException e)
			{
			Thread.currentThread().interrupt();
			return;
			}
			}
			}
			}

			/**
			* Takes the start of the next run of this publisher, or has it leave if there is none - under
			* the lock a session hands its start over under, so that a session either finds this publisher
			* still to run from its start, or finds none and starts one.
			*
			* @return the start of the next run, or {@code null} once this publisher has left
			*/
			private RecoveryStart takeStart()
			{
			synchronized (recoveryLock)
			{
			final RecoveryStart start = nextStart;
			nextStart = null;
			if (start == null \|\| shutdown.get())
			{
			// RSUpdater thread has finished its work, let's remove it from memory
			// so another RSUpdater thread can be started if needed.
			rsUpdater = null;
			broker.setRecoveryRequired(false);
			return null;
			}
			return start;
			}
			}
			}
			@@ -1926,16 +2118,28 @@
			ctx = new DeleteContext(csn, modifiedEntryUUID);
			deleteOperation.setAttachment(SYNCHROCONTEXT, ctx);

			final int maxDeletes = maxDeletesKeptAside;
			FakeDelOperation dropped = null;
			synchronized (replayOperations)
			{
			int size = replayOperations.size();
			if (size >= 10000)
			if (maxDeletes > 0)
			{
			replayOperations.remove(replayOperations.firstKey());
			if (replayOperations.size() >= maxDeletes)
			{
			dropped = replayOperations.pollFirstEntry().getValue();
			}
			replayOperations.put(csn, new FakeDelOperation(deleteOperation.getEntryDN(), csn, modifiedEntryUUID));
			}
			FakeOperation op = new FakeDelOperation(
			deleteOperation.getEntryDN(), csn, modifiedEntryUUID);
			replayOperations.put(csn, op);
			}
			/*
			* The oldest delete goes first: it is the least likely one to be missed by a replication server.
			* One which no replication server has taken - done while none took the changes, or while a
			* recovery which kept too many of them runs - is published by nobody from now on, and the
			* replicas which miss it keep the entry.
			*/
			if (dropped != null && !dropped.isPublished())
			{
			logger.error(ERR_UNPUBLISHED_DELETE_DROPPED, getBaseDN(), dropped.getDN(), dropped.getCSN(), maxDeletes);
			}
			}

			@@ -2376,6 +2580,36 @@
			}

			/**
			* {@inheritDoc}
			* <p>
			* A local delete the broker wrote is marked as published among the deletes kept aside: dropping
			* it from there later loses nothing a replication server has not taken.
			*/
			@Override
			public boolean publish(UpdateMsg msg)
			{
			final boolean published = super.publish(msg);
			if (published && msg instanceof DeleteMsg)
			{
			deleteKeptAsidePublished(msg.getCSN());
			}
			return published;
			}

			/** Marks the delete kept aside with the provided CSN, if there is one, as taken by a replication server. */
			private void deleteKeptAsidePublished(CSN csn)
			{
			synchronized (replayOperations)
			{
			final FakeDelOperation delete = replayOperations.get(csn);
			if (delete != null)
			{
			delete.published();
			}
			}
			}

			/**
			* Check if an operation must be synchronized.
			* Also update the list of pending changes and the server RUV
			* @param op the operation
			@@ -2473,6 +2707,19 @@
			}
			else if (!op.isSynchronizationOperation() && curCSN != null)
			{
			/*
			* A delete is kept aside as its conflicts are resolved, before the access control, the plugins
			* and the backend have had their say: one which did not go through is no change, and a
			* recovery would publish it as one - the replicas it reached would delete an entry this one
			* keeps.
			*/
			if (op.getOperationType() == OperationType.DELETE)
			{
			synchronized (replayOperations)
			{
			replayOperations.remove(curCSN);
			}
			}
			// Remove an unsuccessful non-replication operation from the pending
			// changes list.
			pendingChanges.remove(curCSN);
			@@ -2634,10 +2881,14 @@
			*/
			awaitReplayDrained();

			final RSUpdater rsUpdater = this.rsUpdater.get();
			if (rsUpdater != null)
			synchronized (recoveryLock)
			{
			rsUpdater.initiateShutdown();
			if (rsUpdater != null)
			{
			rsUpdater.initiateShutdown();
			}
			// A publisher which waits to have the session started again has nothing left to wait for.
			recoveryLock.notifyAll();
			}

			// stop the thread in charge of flushing the ServerState.
			@@ -4472,16 +4723,15 @@
			*/
			private void waitBeforeSessionRestart(int restarts, long wakes)
			{
			final long until = monotonicNowInMs()
			+ Math.min(REPLAY_RETRY_DELAY_IN_MS * restarts, MAX_REPLAY_RETRY_DELAY_IN_MS);
			final long until = monotonicNowInMs() + sessionRestartBackoffInMs(restarts);
			try
			{
			/*
			* Waited on a monitor rather than slept through, so that a domain which is going
			* away or is being disabled is not waited out: the thread which holds this wait is
			* a replay thread the shutdown of the pool joins, or the state checkpointer the
			* shutdown of the domain waits for. The session it would start back is one the
			* domain is stopping anyway.
			* shutdown of the domain waits for - or a missing change publisher whose recovery
			* failed. The session it would start back is one the domain is stopping anyway.
			*
			* The wake is counted rather than only checked for by its flag: a domain which was
			* disabled and enabled back has its flag cleared again, and a wait which read the
			@@ -4510,6 +4760,18 @@
			}

			/**
			* Returns how long the session is left before it is started again, after the provided number
			* of restarts in a row.
			*
			* @param restarts how many times in a row the session was restarted already
			* @return the wait, in milliseconds
			*/
			private static long sessionRestartBackoffInMs(int restarts)
			{
			return Math.min(REPLAY_RETRY_DELAY_IN_MS * restarts, MAX_REPLAY_RETRY_DELAY_IN_MS);
			}

			/**
			* Has the next session restarts of this domain fail where they would start the session
			* again.
			* <p>
			@@ -5339,6 +5601,12 @@
			* BackendConfigManager's write lock when a backend is being deregistered.
			*/
			disabled = true;
			// A missing change publisher which waits to have the session started again has nothing left
			// to wait for: the domain starts a session of its own when it is enabled back.
			synchronized (recoveryLock)
			{
			recoveryLock.notifyAll();
			}
			disableService(); // This will cut the session and wake up the listener
			awaitReplayDrained();
			state.save();
			@@ -5502,13 +5770,159 @@
			{
			state.clearInMemory();
			state.loadState();
			bringAStateWithNoChangeOfThisReplicaUpToTheData();
			getGenerator().adjust(state.getMaxCSN(getServerId()));
			forgetTheLocalChangesTheDataLacks();

			// Retrieves the generation ID associated with the data imported
			generationId = loadGenerationId();
			}

			/**
			* Drops the deletes kept aside, and the local changes a recovery kept, which the data the
			* ServerState was just read from does not reflect (issue #1221): it was restored from a backup, or
			* imported, since they were done, and a recovery which published them would have the other
			* replicas delete or change entries which this one holds as they were.
			* <p>
			* What comes after the newest change of this replica the state holds is not in the data: the state
			* read back covers the changes the historical information of the entries holds, whether the save
			* which was to write them wrote or not (see {@link #bringAStateWithNoChangeOfThisReplicaUpToTheData()})
			* - except for a delete, which leaves no historical information, and which the state misses too when
			* it could not be saved: a delete of the base entry takes the saved state away with it. A delete is
			* therefore dropped only if its entry is in the data. A domain enabled back on its own data loses
			* none of them, unless the search which brings its state up to the data does not go through, which
			* {@code ERR_CANNOT_RECOVER_CHANGES} tells.
			*/
			private void forgetTheLocalChangesTheDataLacks()
			{
			final CSN newest = state.getMaxCSN(getServerId());

			// Looked up without the locks of the lists: a local operation holds the lock of its entry, which
			// the lookup may need, while it waits for the lock of the pending changes.
			final List<FakeDelOperation> deletes;
			synchronized (replayOperations)
			{
			deletes = new ArrayList<>((newest != null ? replayOperations.tailMap(newest, false) : replayOperations).values());
			}
			final List<CSN> deletesTheDataLacks = new ArrayList<>();
			for (FakeDelOperation delete : deletes)
			{
			if (holdsTheEntry(delete.getEntryUUID()))
			{
			deletesTheDataLacks.add(delete.getCSN());
			}
			}
			synchronized (replayOperations)
			{
			replayOperations.keySet().removeAll(deletesTheDataLacks);
			}

			final List<CSN> keptChangesTheDataLacks = new ArrayList<>();
			for (LDAPUpdateMsg kept : pendingChanges.keptChangesNewerThan(newest))
			{
			if (!(kept instanceof DeleteMsg) \|\| holdsTheEntry(kept.getEntryUUID()))
			{
			keptChangesTheDataLacks.add(kept.getCSN());
			}
			}
			pendingChanges.dropKeptChanges(keptChangesTheDataLacks);
			}

			/**
			* Brings a ServerState just read back which holds no change of this replica up to the changes of
			* this replica the data holds, from the first local change the recovery which runs kept.
			* <p>
			* The state read back may miss the last changes of this replica: the save which was to write them
			* did not - a save gives its turn up to the one in flight, and a write may fail. The state is
			* brought up to the changes the historical information of the entries holds past the newest change
			* of this replica it read as it is loaded, but one which read none is left as it is, as that search
			* would go through every entry this replica ever changed. Such a state is read back once the base
			* entry, which holds the saved state, was deleted and added back, and no save wrote the state into it
			* since: the local changes a recovery kept meanwhile would all be taken for changes the data does not
			* reflect, and the add of the base entry dropped, while the deletes which removed the suffix are
			* published. The search here starts at the first change the recovery kept, which bounds it.
			*/
			private void bringAStateWithNoChangeOfThisReplicaUpToTheData()
			{
			if (state.getMaxCSN(getServerId()) != null)
			{
			return;
			}
			final List<LDAPUpdateMsg> kept = pendingChanges.keptChangesNewerThan(null);
			if (kept.isEmpty())
			{
			return;
			}
			final CSN newest = newestChangeOfThisReplicaInTheDataFrom(kept.get(0).getCSN());
			if (newest != null)
			{
			state.update(newest);
			}
			}

			/**
			* Returns the newest change of this replica the historical information of the entries holds, among
			* the changes from the provided one on.
			*
			* @return the change, or {@code null} if the data holds none, or the search did not go through: the
			* changes are then taken for changes the data does not hold, as they are when the search which
			* brings the state up to them as it is loaded does not go through
			*/
			private CSN newestChangeOfThisReplicaInTheDataFrom(CSN from)
			{
			final InternalSearchOperation op;
			try
			{
			op = searchForChangedEntries(getBaseDN(), from, null);
			}
			catch (Exception e)
			{
			logger.traceException(e);
			logger.error(ERR_CANNOT_RECOVER_CHANGES, getBaseDN());
			return null;
			}
			if (op.getResultCode() != ResultCode.SUCCESS)
			{
			// A suffix whose base entry is gone holds no change.
			if (op.getResultCode() != ResultCode.NO_SUCH_OBJECT)
			{
			logger.error(ERR_CANNOT_RECOVER_CHANGES, getBaseDN());
			}
			return null;
			}
			CSN newest = null;
			for (SearchResultEntry entry : op.getSearchEntries())
			{
			for (FakeOperation change : EntryHistorical.generateFakeOperations(entry))
			{
			final CSN csn = change.getCSN();
			if (csn.getServerId() == getServerId() && (newest == null \|\| csn.isNewerThan(newest)))
			{
			newest = csn;
			}
			}
			}
			return newest;
			}

			/**
			* Tells whether the data of this domain holds the entry with the provided entryUUID. A lookup
			* which did not go through counts as one which found it: a delete then is not published, rather
			* than published over an entry which may be there.
			*/
			private boolean holdsTheEntry(String entryUUID)
			{
			try
			{
			return findEntryDN(entryUUID) != null;
			}
			catch (SearchFailedException e)
			{
			return true;
			}
			}

			/**
			* Enable back the domain after a previous disable.
			* The domain will connect back to a replication Server and
			* will recreate threads to listen for messages from the Synchronization
			@@ -6769,6 +7183,7 @@
			* changes.
			*/
			CSN replServerMaxCSN = rsState.getCSN(getServerId());
			final CSN ourMaxCSN = state.getMaxCSN(getServerId());

			// we don't want to update from here (a DS) an empty RS because
			// normally the RS should have been updated by other RSes except for
			@@ -6776,20 +7191,32 @@
			// ... hence the RS we are connected to should not be empty
			// ... or if it is empty, it is due to a voluntary reset
			// and we don't want to update it with our changes that could be huge.
			if (replServerMaxCSN != null && replServerMaxCSN.getSeqnum() != 0)
			final boolean replServerHoldsChanges = replServerMaxCSN != null && replServerMaxCSN.getSeqnum() != 0;
			if (replServerHoldsChanges
			&& ourMaxCSN != null
			&& !ourMaxCSN.isOlderThanOrEqualTo(replServerMaxCSN))
			{
			CSN ourMaxCSN = state.getMaxCSN(getServerId());
			if (ourMaxCSN != null
			&& !ourMaxCSN.isOlderThanOrEqualTo(replServerMaxCSN))
			{
			pendingChanges.setRecovering(true);
			broker.setRecoveryRequired(true);
			final RSUpdater rsUpdater = new RSUpdater(replServerMaxCSN);
			if (this.rsUpdater.compareAndSet(null, rsUpdater))
			{
			rsUpdater.start();
			}
			}
			startMissingChangesPublisher(replServerMaxCSN, true);
			}
			else if (pendingChanges.isRecovering())
			{
			/*
			* A missing change publisher which failed - its broker reached no replication server, or
			* was stopped, or its search failed and it had the session started again - left the
			* pending changes in recovery, and no recovery of this session would end it, as this
			* replication server misses no change of this replica: the local changes would be
			* recorded in the state and published by nobody. A publisher is started all the same,
			* from what the replication server holds: it finds nothing to publish, or the changes
			* done since this check, and ends the recovery under the lock the local changes are
			* pushed under. The recovery is not ended here, as this runs under the lock of the
			* connect phase of the broker, which a change being pushed waits for holding the lock of
			* the pending changes.
			*
			* A publisher of the previous session which still runs is handed this start: it publishes
			* on this session from there, and ends the recovery then.
			*/
			startMissingChangesPublisher(replServerHoldsChanges ? replServerMaxCSN
			: ourMaxCSN != null ? ourMaxCSN : new CSN(TimeThread.getTime(), 0, getServerId()), false);
			}
			} catch (Exception e)
			{
			@@ -6798,133 +7225,539 @@
			}

			/**
			* Has the changes of this replica which come after the provided CSN published on the session
			* which starts: starts the thread which publishes them, or hands the start over to the one which
			* runs, which then publishes on this session from there - a publisher which kept going from where
			* it had got to on the session before would leave the replication server of this one without the
			* changes it misses before that point, and it would take none of them once it had a later one.
			* <p>
			* Decided under the lock a recovery ends under, and its publisher leaves under: a publisher which
			* was about to end the recovery or to leave runs from this start instead.
			*
			* @param startCSN
			* the newest change of this replica the replication server of the session holds, or the
			* CSN after which a recovery left by an earlier session is to be ended
			* @param replServerMisses
			* whether the replication server misses changes of this replica: the recovery starts,
			* rather than only being ended if a failed publisher left one
			*/
			private void startMissingChangesPublisher(CSN startCSN, boolean replServerMisses)
			{
			final RecoveryStart start = new RecoveryStart(startCSN, broker.getSession());
			synchronized (recoveryLock)
			{
			if (replServerMisses)
			{
			pendingChanges.setRecovering(true);
			broker.setRecoveryRequired(true);
			}
			else if (!pendingChanges.isRecovering())
			{
			// The publisher of an earlier session ended it since it was checked.
			return;
			}
			if (rsUpdater == null)
			{
			rsUpdater = new RSUpdater(start);
			rsUpdater.start();
			}
			else
			{
			rsUpdater.handOver(start);
			}
			}
			}

			/**
			* Build the list of changes that have been processed by this server after the
			* CSN given as a parameter and publish them using the given session.
			* <p>
			* The changes done before the recovery began are rebuilt from the historical information of
			* their entries, in rounds, each one from where the round before it stopped, up to the last of
			* them: each of them has its entry written by then, so the search of a round finds it, or a later
			* change of the same values, which supersedes it. The local changes pushed since the recovery began
			* are kept by the pending changes, with their messages, and they are published next, in order. The
			* recovery ends with the round which publishes up to the last local change, no change having been
			* pushed meanwhile; once a round has caught up with the local changes and some were pushed while it
			* ran, the round after it holds them back, so that the recovery ends with it. The local changes
			* wait for as long as publishing the ones pushed during the round before takes, and for no search.
			* <p>
			* A recovery which kept too many of the local changes keeps none, and searches for them as well,
			* from where it has got to: a round then publishes up to the last change done when it began, or up
			* to the end of its search window, and a change done while it runs is left to the round after it.
			* The search may have missed it, and publishing a later change in its place would have the next
			* round search past it - while a replication server takes no change older than the newest one it
			* holds of this replica. Once such a round has caught up with the local changes and some were done
			* while it ran, the rounds after it hold them back up to the last one.
			* <p>
			* A round counts as published only what the broker did not refuse: when it refuses a change - it
			* reaches no replication server, its session is not the one the recovery publishes on anymore, or
			* it was stopped - the recovery fails, and the changes are left to the start the session which
			* replaced it hands over, or to the next session, which starts a publisher whether its
			* replication server misses changes of this replica or not, so that the recovery ends.
			* <p>
			* A round whose search does not go through publishes nothing, and the recovery fails with it: what
			* the search found may be some of the changes only, and a replication server which took them would
			* take none of the ones it missed, which are older. The caller has the session started again, for
			* the recovery of the next one to publish them.
			* <p>
			* The deletes kept aside are left where they are, the ones a round publishes included: the
			* replication server of a later session may miss them.
			*
			* @param startCSN
			* The CSN where we need to start the search
			* @param session
			* The session to use to publish the changes
			* @return A boolean indicating he success of the operation.
			* The broker to publish the changes with
			* @param recoverySession
			* The session to publish the changes on: the broker refuses them for another one
			* @param publisher
			* The missing change publisher which runs the recovery, which a session which starts
			* meanwhile hands its start over to, or {@code null} if none does
			* @return {@code true} once the recovery is over, {@code false} if the broker refused a change, a
			* session which started meanwhile handed its start over, or the domain was shut down
			* @throws Exception
			* if an Exception happens during the search.
			* if the search of a round did not go through, or a round threw: the changes the round
			* did not publish are left to the next recovery
			*/
			boolean buildAndPublishMissingChanges(CSN startCSN, ReplicationBroker session)
			throws Exception
			private boolean buildAndPublishMissingChanges(CSN startCSN, ReplicationBroker session,
			Session recoverySession, RSUpdater publisher) throws Exception
			{
			// Trim the changes in replayOperations that are older than the startCSN.
			synchronized (replayOperations)
			{
			Iterator<CSN> it = replayOperations.keySet().iterator();
			while (it.hasNext())
			{
			if (shutdown.get())
			{
			return false;
			}
			if (it.next().isNewerThan(startCSN))
			{
			break;
			}
			it.remove();
			}
			}

			CSN lastRetrievedChange;
			InternalSearchOperation op;
			final MissingChanges missingChanges = new MissingChanges(session, recoverySession, publisher);
			CSN currentStartCSN = startCSN;
			do
			boolean holdBackLocalChanges = false;
			while (true)
			{
			if (shutdown.get())
			if (shutdown.get() \|\| missingChanges.startHandedOver())
			{
			return false;
			}

			lastRetrievedChange = null;
			// We can't do the search in one go because we need to store the results
			// so that we are sure we send the operations in order and because the
			// list might be large.
			// So we search by interval of 10 seconds and store the results in the
			// replayOperations list so that they are sorted before sending them.
			long missingChangesDelta = currentStartCSN.getTime() + 10000;
			CSN endCSN = new CSN(missingChangesDelta, 0xffffffff, getServerId());

			ScanSearchListener listener =
			new ScanSearchListener(currentStartCSN, endCSN);
			op = searchForChangedEntries(getBaseDN(), currentStartCSN, endCSN,
			listener);

			// Publish and remove all the changes from the replayOperations list
			// that are older than the endCSN.
			final List<FakeOperation> opsToSend = new LinkedList<>();
			synchronized (replayOperations)
			final CSN fromCSN = currentStartCSN;
			final MissingChangesRound round = holdBackLocalChanges
			? pendingChanges.holdBackLocalChangesWhile(() -> missingChanges.publishUntilCaughtUp(fromCSN))
			: missingChanges.publish(fromCSN, false);
			if (round == null \|\| !round.succeeded)
			{
			Iterator<FakeOperation> itOp = replayOperations.values().iterator();
			while (itOp.hasNext())
			{
			if (shutdown.get())
			{
			return false;
			}
			FakeOperation fakeOp = itOp.next();
			if (fakeOp.getCSN().isNewerThan(endCSN) // sanity check
			\|\| !state.cover(fakeOp.getCSN())
			// do not look for replay operations in the future
			\|\| currentStartCSN.isNewerThan(now()))
			{
			break;
			}
			return false;
			}
			if (round.endedRecovery)
			{
			return true;
			}
			holdBackLocalChanges = round.caughtUp;
			currentStartCSN = round.publishedUpTo;
			}
			}

			lastRetrievedChange = fakeOp.getCSN();
			opsToSend.add(fakeOp);
			itOp.remove();
			/**
			* Publishes the changes of this replica which come after the provided CSN, as
			* {@link #buildAndPublishMissingChanges(CSN, ReplicationBroker, Session, RSUpdater)} does, on the
			* session the provided broker publishes on, and with no missing change publisher a session could
			* hand a start over to.
			*
			* @param startCSN
			* The CSN where we need to start the search
			* @param session
			* The broker to publish the changes with
			* @return {@code true} once the recovery is over, {@code false} if the broker refused a change or
			* the domain was shut down
			* @throws Exception
			* if the search of a round did not go through, or a round threw
			*/
			@VisibleForTesting
			boolean buildAndPublishMissingChanges(CSN startCSN, ReplicationBroker session) throws Exception
			{
			return buildAndPublishMissingChanges(startCSN, session, session.getSession(), null);
			}

			/** What one round of {@link LDAPReplicationDomain#buildAndPublishMissingChanges(CSN, ReplicationBroker)} did. */
			private static final class MissingChangesRound
			{
			/** Every change of this replica up to this CSN is published, or has nothing left to publish it from. */
			private final CSN publishedUpTo;
			/** Whether the round went as far as the last local change: neither its search window nor the search of the recovery ended before. */
			private final boolean caughtUp;
			/** Whether the broker refused none of the changes the round published: a failed search throws. */
			private final boolean succeeded;
			/** Whether the round ended the recovery: it caught up, and no change was done while it ran. */
			private final boolean endedRecovery;

			private MissingChangesRound(CSN publishedUpTo, boolean caughtUp, boolean succeeded,
			boolean endedRecovery)
			{
			this.publishedUpTo = publishedUpTo;
			this.caughtUp = caughtUp;
			this.succeeded = succeeded;
			this.endedRecovery = endedRecovery;
			}
			}

			/** One run of {@link LDAPReplicationDomain#buildAndPublishMissingChanges(CSN, ReplicationBroker, Session, RSUpdater)}. */
			private final class MissingChanges
			{
			private final ReplicationBroker recoveryBroker;
			private final Session recoverySession;
			private final RSUpdater publisher;

			private MissingChanges(ReplicationBroker recoveryBroker, Session recoverySession, RSUpdater publisher)
			{
			this.recoveryBroker = recoveryBroker;
			this.recoverySession = recoverySession;
			this.publisher = publisher;
			}

			/** Whether a session which started meanwhile handed its start over to the publisher of this run. */
			private boolean startHandedOver()
			{
			if (publisher == null)
			{
			return false;
			}
			synchronized (recoveryLock)
			{
			return publisher.hasAStartHandedOver();
			}
			}

			/**
			* Ends the recovery, under the lock a session starts one under, unless a session which started
			* meanwhile handed its start over: its replication server may miss changes this run published to
			* the one before, and the recovery goes on from that start.
			*/
			private void endRecovery(Runnable recovered)
			{
			synchronized (recoveryLock)
			{
			if (publisher == null \|\| !publisher.hasAStartHandedOver())
			{
			recovered.run();
			broker.setRecoveryRequired(false);
			}
			}
			}

			for (FakeOperation opToSend : opsToSend)
			/**
			* Runs the rounds of a recovery which holds the local changes back, up to the last local change.
			* <p>
			* The changes done while the round before ran may go further than one search window: a round
			* which stopped at the end of its window and let the local changes go again would leave the next
			* one as far behind. Nothing is done while these rounds run, so the last of them ends the
			* recovery.
			*
			* @return the last round, or {@code null} if the domain was shut down meanwhile
			*/
			private MissingChangesRound publishUntilCaughtUp(CSN fromCSN) throws Exception
			{
			MissingChangesRound round = publish(fromCSN, true);
			while (round != null && round.succeeded && !round.caughtUp)
			{
			if (shutdown.get())
			{
			return false;
			return null;
			}
			session.publishRecovery(opToSend.generateMessage());
			round = publish(round.publishedUpTo, true);
			}
			return round;
			}

			/**
			* Runs one round: publishes the changes done before the recovery began which come after
			* {@code fromCSN}, rebuilt from the historical information, up to the end of a search window;
			* past them, the local changes the recovery kept; or, if it kept too many to keep any, the
			* changes rebuilt from the historical information up to the last local change.
			*
			* @param heldBack
			* whether the round runs while the local changes are held back
			* @return the round, or {@code null} if the domain was shut down meanwhile
			* @throws MissingChangesSearchFailedException
			* if the search did not go through: the round published nothing
			*/
			private MissingChangesRound publish(CSN fromCSN, boolean heldBack) throws Exception
			{
			if (importingData)
			{
			return refusedForTheImport(fromCSN);
			}
			final PendingChanges.KeptChanges keeping = pendingChanges.keptChanges();
			if (keeping == null)
			{
			return publishMissingChanges(fromCSN, null);
			}
			if (keeping.searchUpTo != null && fromCSN.isOlderThan(keeping.searchUpTo))
			{
			return publishMissingChanges(fromCSN, keeping.searchUpTo);
			}
			final List<LDAPUpdateMsg> kept = pendingChanges.keptChangesAfter(fromCSN);
			if (kept == null)
			{
			// It kept too many since.
			return publishMissingChanges(fromCSN, null);
			}
			return publishKeptChanges(fromCSN, kept, heldBack);
			}

			/**
			* A total update replaces the data of this replica: what a recovery would publish meanwhile was
			* done on the data it replaces, and the changes it kept which the new data does not reflect are
			* dropped once the import is over. The round is refused, and the session started again after the
			* import has a recovery of its own, from the state the new data holds.
			*/
			private MissingChangesRound refusedForTheImport(CSN fromCSN)
			{
			if (logger.isDebugEnabled())
			{
			logger.debug(LocalizableMessage.raw("publish loop from " + fromCSN
			+ " stopped: a total update replaces the data of this replica"));
			}
			return new MissingChangesRound(fromCSN, false, false, false);
			}

			/**
			* Runs a round which publishes the changes of this replica which come after {@code fromCSN},
			* rebuilt from the historical information: up to {@code searchUpTo}, where the search of the
			* recovery ends, or - if it is {@code null} - up to the last change done when the round begins,
			* and in either case up to the end of the search window of the round if that comes first. A
			* round which went up to the last change done ends the recovery if no change was done since.
			*
			* @return the round, or {@code null} if the domain was shut down meanwhile
			* @throws MissingChangesSearchFailedException
			* if the search did not go through: the round published nothing
			*/
			private MissingChangesRound publishMissingChanges(CSN fromCSN, CSN searchUpTo) throws Exception
			{
			// Read before the search: every change up to it has its entry written, so the search sees it.
			final CSN lastChange = searchUpTo != null ? searchUpTo : getLastLocalChange();
			// We can't do the search in one go because we need to store the results
			// so that we are sure we send the operations in order and because the
			// list might be large.
			// So we search by interval of 10 seconds and store the results in a
			// list of the round so that they are sorted before sending them.
			// The greatest sequence number reads as the greatest both as a CSN and as the text the index
			// of the historical information compares, where 0xffffffff would read as -1 and as the
			// greatest: the changes of the last millisecond fell then out of this window and of the next.
			final CSN endCSN = new CSN(fromCSN.getTime() + 10000, Integer.MAX_VALUE, getServerId());
			final boolean reachesLastChange = lastChange == null \|\| !lastChange.isNewerThan(endCSN);
			final CSN bound = !reachesLastChange ? endCSN : lastChange != null ? lastChange : fromCSN;
			// Where the search of the recovery ends, the local changes it kept come next: only a round which
			// went up to the last local change catches up with them.
			final boolean caughtUp = reachesLastChange && searchUpTo == null;

			final NavigableMap<CSN, FakeOperation> found = new TreeMap<>();
			final InternalSearchOperation op =
			searchForChangedEntries(getBaseDN(), fromCSN, endCSN, new ScanSearchListener(fromCSN, endCSN, found));
			final Consumer<InternalSearchOperation> searchHook = missingChangesSearchHook;
			if (searchHook != null)
			{
			searchHook.accept(op);
			}
			/*
			* A search which did not go through may have found some of the changes only, and a replication
			* server which took them would take none of the ones it missed, which are older: the round
			* publishes nothing. A search which found no base entry went through: the base entry of the
			* domain is gone, and no entry is left below it to hold a change, so the deletes kept aside are
			* all there is - and they go out as in any other round, or the deletes which removed the suffix
			* would wait for it to come back. The same result comes for a base DN with no backend - one an
			* import has taken away - and that search found nothing of what is there.
			*/
			final ResultCode resultCode = op.getResultCode();
			if (!resultCode.equals(ResultCode.SUCCESS)
			&& !(resultCode.equals(ResultCode.NO_SUCH_OBJECT) && baseEntryIsAbsentFromALiveBackend()))
			{
			throw new MissingChangesSearchFailedException(fromCSN, op);
			}

			if (lastRetrievedChange != null)
			/*
			* Publish what the search found and the deletes kept aside which come after fromCSN, up to the
			* bound, in order. The deletes stay kept aside, the ones before fromCSN as well: the replication
			* server of this session has them, but the one of a later session may not, and a delete has
			* nothing else to be published from.
			*/
			final NavigableMap<CSN, FakeOperation> opsToSend = new TreeMap<>(found.headMap(bound, true));
			if (bound.isNewerThan(fromCSN))
			{
			if (logger.isDebugEnabled())
			synchronized (replayOperations)
			{
			logger.debug(LocalizableMessage.raw("publish loop"
			+ " >=" + currentStartCSN + " <=" + endCSN
			+ " nentries=" + op.getEntriesSent()
			+ " result=" + op.getResultCode()
			+ " lastRetrievedChange=" + lastRetrievedChange));
			opsToSend.putAll(replayOperations.subMap(fromCSN, false, bound, true));
			}
			currentStartCSN = lastRetrievedChange;
			}
			else
			{
			if (logger.isDebugEnabled())
			{
			logger.debug(LocalizableMessage.raw("publish loop"
			+ " >=" + currentStartCSN + " <=" + endCSN
			+ " nentries=" + op.getEntriesSent()
			+ " result=" + op.getResultCode()
			+ " no changes"));
			}
			currentStartCSN = endCSN;
			}
			} while (pendingChanges.recoveryUntil(currentStartCSN)
			&& op.getResultCode().equals(ResultCode.SUCCESS));

			return op.getResultCode().equals(ResultCode.SUCCESS);
			for (FakeOperation opToSend : opsToSend.values())
			{
			if (shutdown.get())
			{
			return null;
			}
			if (importingData)
			{
			return refusedForTheImport(fromCSN);
			}
			if (!recoveryBroker.publishRecovery(opToSend.generateMessage(), recoverySession))
			{
			return refused(fromCSN, opToSend.getCSN());
			}
			if (opToSend instanceof FakeDelOperation)
			{
			((FakeDelOperation) opToSend).published();
			}
			}

			final CSN publishedUpTo = bound.isNewerThan(fromCSN) ? bound : fromCSN;
			if (logger.isDebugEnabled())
			{
			logger.debug(LocalizableMessage.raw("publish loop"
			+ " >=" + fromCSN + " <=" + endCSN
			+ " nentries=" + op.getEntriesSent()
			+ " result=" + op.getResultCode()
			+ " published=" + opsToSend.size()
			+ " upTo=" + publishedUpTo));
			}
			final boolean endedRecovery = caughtUp && !pendingChanges.recoveryUntil(publishedUpTo, this::endRecovery);
			return new MissingChangesRound(publishedUpTo, caughtUp, true, endedRecovery);
			}

			/**
			* Runs a round which publishes the local changes the recovery kept after {@code fromCSN}, in
			* order, and ends the recovery if no change was pushed since it took them.
			*
			* @param heldBack
			* whether the round runs while the local changes are held back
			* @return the round, or {@code null} if the domain was shut down meanwhile
			*/
			private MissingChangesRound publishKeptChanges(CSN fromCSN, List<LDAPUpdateMsg> kept, boolean heldBack)
			{
			CSN publishedUpTo = fromCSN;
			for (LDAPUpdateMsg msg : kept)
			{
			if (shutdown.get())
			{
			return null;
			}
			if (importingData)
			{
			return refusedForTheImport(fromCSN);
			}
			if (!recoveryBroker.publishRecovery(msg, recoverySession))
			{
			return refused(fromCSN, msg.getCSN());
			}
			publishedUpTo = msg.getCSN();
			if (msg instanceof DeleteMsg)
			{
			deleteKeptAsidePublished(msg.getCSN());
			}
			}
			if (logger.isDebugEnabled())
			{
			logger.debug(LocalizableMessage.raw("publish loop of the kept changes >" + fromCSN
			+ " published=" + kept.size() + " upTo=" + publishedUpTo));
			}
			final Consumer<Boolean> publishedHook = keptChangesPublishedHook;
			if (publishedHook != null)
			{
			publishedHook.accept(heldBack);
			}
			final boolean endedRecovery = !pendingChanges.recoveryUntil(publishedUpTo, this::endRecovery);
			if (!endedRecovery && heldBack && !startHandedOver())
			{
			/*
			* Nothing is pushed while the local changes are held back, so the changes kept reach the last
			* local change - unless one which should have been kept is not there. Rather than go round
			* for ever under the lock every local change waits for, the recovery keeps none from here on,
			* and searches for them in the historical information.
			*/
			if (logger.isDebugEnabled())
			{
			logger.debug(LocalizableMessage.raw("the changes kept up to " + publishedUpTo
			+ " do not reach the last local change: the recovery searches for the ones after them"));
			}
			pendingChanges.keepNoMoreChanges();
			return new MissingChangesRound(publishedUpTo, false, true, false);
			}
			return new MissingChangesRound(publishedUpTo, true, true, endedRecovery);
			}

			/**
			* The broker refused a change - it reaches no replication server, its session is not the one
			* of this run anymore, or it was stopped: the changes from it are left where they are, to the
			* start a session which replaced it hands over, or to the recovery of the next session.
			*/
			private MissingChangesRound refused(CSN fromCSN, CSN csn)
			{
			if (logger.isDebugEnabled())
			{
			logger.debug(LocalizableMessage.raw("publish loop refused " + csn
			+ ": the changes from it are left to the start a session hands over, or to the recovery of the next one"));
			}
			return new MissingChangesRound(fromCSN, false, false, false);
			}
			}

			private static CSN now()
			/**
			* Thrown by a round of {@link LDAPReplicationDomain#buildAndPublishMissingChanges(CSN, ReplicationBroker)}
			* whose search did not go through.
			*/
			private static final class MissingChangesSearchFailedException extends Exception
			{
			// ensure now() will always come last with isNewerThan() test,
			// even though the timestamp, or the timestamp and seqnum would be the same
			return new CSN(TimeThread.getTime(), Integer.MAX_VALUE, Integer.MAX_VALUE);
			private static final long serialVersionUID = 1L;

			private MissingChangesSearchFailedException(CSN fromCSN, InternalSearchOperation search)
			{
			super("the search of the changes after " + fromCSN + " ended with "
			+ search.getResultCode().getName() + " " + search.getErrorMessage());
			}
			}

			/**
			* Sets what {@link #buildAndPublishMissingChanges(CSN, ReplicationBroker)} runs after the search
			* of each of its rounds, before it chooses what the round publishes.
			* <p>
			* Only there for the tests which do local changes in that gap - the changes a round may publish
			* are bounded by what was done when its search began, and a change done in between is the one
			* a bound taken any later lets through without the search having found it - or which fail the
			* search, by setting its result code or by throwing.
			*
			* @param hook what to run there, handed the search of the round, or {@code null} to run nothing
			*/
			@VisibleForTesting
			void setMissingChangesSearchHook(Consumer<InternalSearchOperation> hook)
			{
			missingChangesSearchHook = hook;
			}

			/**
			* Sets how many local deletes this domain keeps aside for its recoveries at most.
			*
			* @param maxDeletes
			* how many local deletes are kept aside at most
			*/
			@VisibleForTesting
			void setMaxDeletesKeptAside(int maxDeletes)
			{
			maxDeletesKeptAside = maxDeletes;
			}

			/**
			* Sets what the missing change publisher runs after each of its runs, before it takes the start a
			* session handed over or leaves.
			*
			* @param hook what to run there, or {@code null} to run nothing
			*/
			@VisibleForTesting
			void setMissingChangesPublisherRunHook(Runnable hook)
			{
			missingChangesPublisherRunHook = hook;
			}

			/**
			* Sets what a round of a recovery which publishes the local changes the recovery kept runs once it
			* has published them, before it ends the recovery.
			*
			* @param hook what to run there, handed whether the round holds the local changes back, or
			* {@code null} to run nothing
			*/
			@VisibleForTesting
			void setKeptChangesPublishedHook(Consumer<Boolean> hook)
			{
			keptChangesPublishedHook = hook;
			}

			/**