From c2f7a14064a1bcd170eb25cf8ff7ec57df132dbd Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Tue, 22 Sep 2026 10:29:51 +0000
Subject: [PATCH] [#1040] Leave the session to a total update when a configuration change asks for a restart (#1042)
---
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ConfigChangeDuringImportTest.java | 436 ++++++++++++++++++++++++++++++++++++++++++++++++
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ExternalChangelogDomain.java | 16 +
opendj-server-legacy/src/messages/org/opends/messages/replication.properties | 7
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java | 54 ++++-
opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java | 19 +
5 files changed, 506 insertions(+), 26 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ExternalChangelogDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ExternalChangelogDomain.java
index 9342606..e600381 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ExternalChangelogDomain.java
+++ b/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;
}
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java
index 6ec3909..1467c35 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java
+++ b/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(
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java
index 5538de1..3dd9cfd 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java
@@ -3487,8 +3487,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 +3506,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 +4004,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)
diff --git a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
index f81040f..3129b6e 100644
--- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
+++ b/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 \
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ConfigChangeDuringImportTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ConfigChangeDuringImportTest.java
new file mode 100644
index 0000000..4d0008b
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ConfigChangeDuringImportTest.java
@@ -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')));
+ }
+}
--
Gitblit v1.10.0