- * Holding it costs something, and knowingly: {@code enableService()} connects to the - * replication servers under this lock, so a shutdown, an import or a configuration - * change which arrives while a replay thread is bringing the session back waits for - * that connect - up to the configured connection timeout when the replication servers - * are unreachable, which is the same outage that failed the replay. Every one of those - * stops the session as its first act, so what they wait for is a session which is about - * to be stopped again. The wait between the stop and the start is deliberately left - * outside the lock, so the waiting is bounded by a connect rather than by the backoff. - *
- * It comes after the configuration backend's update lock and never before it: a write to - * the domain configuration entry holds that lock while it calls - * {@link #applyConfigurationChange(ReplicationDomainCfg)}, which takes this one. So - * nothing may write a configuration entry while holding this lock - that is why neither - * the state {@link #disable()} saves nor the generationId {@link #enable()} stores falls - * back to the domain configuration entry when the base entry of the suffix is missing. - */ - private final Object serviceStateLock = new Object(); - /** - * Bumped every time the session of this domain is stopped or started under - * {@link #serviceStateLock}. A replay thread which stopped the session only starts it - * back if this still is the session it stopped: a configuration change, or the end of - * an import, may have started another one while it was waiting for the backend to - * recover. - *
- * It does not count the sessions {@code changeConfig()} and {@code readAssuredConfig()} - * stop and start, which they do without knowing about it: they run under the lock, so a - * replay thread never observes one of theirs, but a session it stopped may well have - * been replaced by one of theirs while it was waiting. That is why the guard in - * {@link #restartSession(boolean)} reads {@code isListenerShuttingDown()} as well - a - * session started outside this counter leaves it untouched, and only the listener says - * that one is running. - */ - @GuardedBy("serviceStateLock") - private long sessionGeneration; /** * Set by {@link #restartService()} when it left the session of this domain alone, so * that the configuration change which asked for the restart can say so. @@ -896,22 +856,21 @@ private void readFractionalConfig(ReplicationDomainCfg configuration, return; } - // Disable service if configuration changed - final boolean needRestart = needReconnection && allowReconnection; /* * The session is stopped, the configuration it depends on is changed and the session * is started again under the lock which the replay thread restarting the session after * a failed replay holds too: a session brought up in the middle of this would be * reading a fractional configuration which is half way through being changed. The - * pair has to be atomic, which the lock inside disableService()/enableService() does - * not make it. + * stop, the change and the start have to be atomic together, which taking the lock + * inside each of disableService()/enableService() does not make them. */ synchronized (serviceStateLock) { + // Disable service if configuration changed + final boolean needRestart = needReconnection && allowReconnection; if (needRestart) { disableService(); - sessionGeneration++; } else if (needReconnection) { @@ -943,7 +902,6 @@ else if (needReconnection) if (needRestart) { enableService(); - sessionGeneration++; } } } @@ -2513,7 +2471,6 @@ public void shutdown() synchronized (serviceStateLock) { disableService(); - sessionGeneration++; } } @@ -3338,13 +3295,13 @@ private void restartSession(boolean wait) final long stoppedSession; synchronized (serviceStateLock) { - if (shutdown.get() || disabled) + if (ownsItsSession()) { // The domain is going away or is being imported into: it owns its session. return; } disableService(); - stoppedSession = ++sessionGeneration; + stoppedSession = getSessionGeneration(); } if (wait) { @@ -3358,21 +3315,18 @@ private void restartSession(boolean wait) } synchronized (serviceStateLock) { - if (shutdown.get() || disabled - || sessionGeneration != stoppedSession || !isListenerShuttingDown()) + if (ownsItsSession() || getSessionGeneration() != stoppedSession) { /* * The domain went away while this thread was waiting, or the session was stopped * and started again by something else - a configuration change, the end of an * import - in the meantime: the session this thread stopped is gone, so it has - * nothing left to start. The generation says a session was started under this - * lock; the listener says one is running, which is what a restart made outside it - * leaves behind. + * nothing left to start. Every stop and every start of a session is counted, so + * the generation alone tells one session from another. */ return; } enableService(); - sessionGeneration++; } } @@ -4052,7 +4006,6 @@ public void disable() */ disabled = true; disableService(); // This will cut the session and wake up the listener - sessionGeneration++; awaitReplayDrained(); state.save(); state.clearInMemory(); @@ -4220,7 +4173,6 @@ public void enable() try { enableService(); - sessionGeneration++; started = true; } finally @@ -5019,23 +4971,6 @@ private LocalizableMessage configChangeFailed(ReplicationDomainCfg domCfg, Excep domCfg.getBaseDN(), stackTraceToSingleLineString(e)); } - /** - * {@inheritDoc} - *
- * Taken under {@link #serviceStateLock} like every other configuration change: this one
- * comes from the external changelog domain - from the entry of its own, or from
- * {@link #applyECLConfiguration} - and it restarts the session as well.
- */
- @Override
- public void changeConfig(Set
+ * Holding it costs something, and knowingly: {@link #enableService()} connects to the
+ * replication servers under this lock, so a shutdown, an import or a configuration
+ * change which arrives while a replay thread is bringing the session back waits for
+ * that connect - up to the configured connection timeout when the replication servers
+ * are unreachable, which is the same outage that failed the replay. Every one of those
+ * stops the session as its first act, so what they wait for is a session which is
+ * about to be stopped again. A wait between a stop and a start belongs outside the
+ * lock, so that the waiting is bounded by a connect rather than by a backoff.
+ *
+ * It comes after the configuration backend's update lock and never before it: a write
+ * to the configuration entry of a domain holds that lock while it calls the domain's
+ * configuration change listener, which takes this one. So nothing may write a
+ * configuration entry while holding this lock - that is why neither the state a domain
+ * saves on its way down nor the generationId it stores on its way up falls back to the
+ * domain configuration entry when the base entry of the suffix is missing.
+ */
+ protected final Object serviceStateLock = new Object();
+ /**
+ * Bumped every time {@link #disableService()} stops the session of this domain or
+ * {@link #enableService()} starts it, both of them under {@link #serviceStateLock}. It
+ * is the identity of the session: a thread which stops one and lets the lock go - a
+ * replay thread waiting out a backoff before it asks for the change it could not apply
+ * again - only starts it back if this still is the session it stopped, since a
+ * configuration change or the end of an import may have started another one while it
+ * was waiting.
+ *
+ * The starts at domain startup are not counted: {@link #startPublishService()} from
+ * the constructor of the domain and {@link #startListenService()} from its start bring
+ * the two halves of the first session up before any replay thread exists to hold a
+ * claim on it.
+ */
+ @GuardedBy("serviceStateLock")
+ private long sessionGeneration;
/**
* The generationId for this replication domain. It is made of a hash of the
@@ -3294,34 +3333,38 @@ else if (processUpdate(updateMsg)
* It can be useful to disable the Replication Service when the
* repository where the replicated information is stored becomes
* temporarily unavailable and replicated updates can therefore not
- * be replayed during a while. This method is not MT safe.
+ * be replayed during a while.
*/
- public void disableService()
+ public final void disableService()
{
- synchronized (sessionLock)
+ synchronized (serviceStateLock)
{
- /*
- * Stop the broker first in order to prevent the listener from reconnecting - see OPENDJ-457.
- */
- if (broker != null)
+ synchronized (sessionLock)
{
- broker.stop();
- }
-
- // Stop the listener thread
- if (listenerThread != null)
- {
- listenerThread.initiateShutdown();
- try
+ /*
+ * Stop the broker first in order to prevent the listener from reconnecting - see OPENDJ-457.
+ */
+ if (broker != null)
{
- listenerThread.join();
+ broker.stop();
}
- catch (InterruptedException e)
+
+ // Stop the listener thread
+ if (listenerThread != null)
{
- // Give up waiting.
+ listenerThread.initiateShutdown();
+ try
+ {
+ listenerThread.join();
+ }
+ catch (InterruptedException e)
+ {
+ // Give up waiting.
+ }
+ listenerThread = null;
}
- listenerThread = null;
}
+ sessionGeneration++;
}
}
@@ -3338,6 +3381,24 @@ protected final boolean isListenerShuttingDown()
return tmp == null || tmp.isShutdownInitiated();
}
+ /**
+ * Returns the generation of the session of this domain: bumped by
+ * {@link #disableService()} and {@link #enableService()} under
+ * {@link #serviceStateLock}, not by the starts at domain startup (see
+ * {@code sessionGeneration}).
+ *
+ * It only says anything while {@link #serviceStateLock} is held, and is meant to be
+ * read under the lock which stopped a session and read again under the lock which
+ * starts it back: the session stopped is gone when the two differ.
+ *
+ * @return the generation of the session of this domain
+ */
+ @GuardedBy("serviceStateLock")
+ protected final long getSessionGeneration()
+ {
+ return sessionGeneration;
+ }
+
/**
* Restart the Replication service after a {@link #disableService()}.
*
@@ -3348,34 +3409,56 @@ protected final boolean isListenerShuttingDown()
* If some data have changed in the repository during the period of time when
* the Replication Service was disabled, this {@link ServerState} should
* therefore be updated by the Replication Domain subclass before calling this
- * method. This method is not MT safe.
+ * method.
*/
- public void enableService()
+ public final void enableService()
{
- synchronized (sessionLock)
+ synchronized (serviceStateLock)
{
- broker.start();
- startListenService();
+ synchronized (sessionLock)
+ {
+ broker.start();
+ startListenService();
+ }
+ /*
+ * Counted once the session really is up: a start which threw leaves the generation
+ * where it was, so the thread which stopped this session still owns it and may try
+ * to bring it back.
+ */
+ sessionGeneration++;
}
}
/**
* Change some ReplicationDomain parameters.
+ *
+ * The change and the restart it may call for are taken together under
+ * {@link #serviceStateLock}, as {@link #readAssuredConfig(ReplicationDomainCfg, boolean)}
+ * takes its own: a session brought up between the two would negotiate the broker
+ * properties which are half way through being changed.
*
* @param config
* The new configuration that this domain should now use.
*/
protected void changeConfig(ReplicationDomainCfg config)
{
- if (broker != null && broker.changeConfig(config))
+ synchronized (serviceStateLock)
{
- restartService();
+ if (broker != null && broker.changeConfig(config))
+ {
+ restartService();
+ }
}
}
/**
* Applies a configuration change to the attributes which should be included
* in the ECL.
+ *
+ * Taken under {@link #serviceStateLock} like every other configuration change: this one
+ * comes from the external changelog domain - from the entry of its own, or from the
+ * domain configuration change which reads that entry - and it restarts the session as
+ * well, so the attributes and the restart go together.
*
* @param includeAttributes
* attributes to be included with all change records.
@@ -3385,11 +3468,14 @@ protected void changeConfig(ReplicationDomainCfg config)
public void changeConfig(Set
- * 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()}.
+ * The pair is taken under {@link #serviceStateLock}, so that nothing starts a session
+ * 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()}.
*/
protected void restartService()
{
- disableService();
- enableService();
+ synchronized (serviceStateLock)
+ {
+ disableService();
+ enableService();
+ }
}
/**
@@ -3874,32 +3966,41 @@ public CSN getLastLocalChange()
protected void readAssuredConfig(ReplicationDomainCfg config,
boolean allowReconnection)
{
- // Disconnect if required: changing configuration values before
- // disconnection would make assured replication used immediately and
- // disconnection could cause some timeouts error.
- final boolean needReconnection = needReconnection(config);
- final boolean needRestart = needReconnection && allowReconnection;
- if (needRestart)
- {
- disableService();
- }
- else if (needReconnection)
- {
- onSessionRestartSuppressed();
- }
/*
- * Stored whether or not the session was restarted for it, as the fractional
- * 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.
+ * The stop, the change and the start are taken together under serviceStateLock: a
+ * session brought up in between, by a replay thread restarting the session after a
+ * failed replay, would negotiate an assured configuration which is half way through
+ * being changed.
*/
- assuredConfig = config;
- if (needRestart)
+ synchronized (serviceStateLock)
{
- enableService();
+ // Disconnect if required: changing configuration values before
+ // disconnection would make assured replication used immediately and
+ // disconnection could cause some timeouts error.
+ final boolean needReconnection = needReconnection(config);
+ final boolean needRestart = needReconnection && allowReconnection;
+ if (needRestart)
+ {
+ disableService();
+ }
+ else if (needReconnection)
+ {
+ onSessionRestartSuppressed();
+ }
+ /*
+ * Stored whether or not the session was restarted for it, as the fractional
+ * 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.
+ */
+ assuredConfig = config;
+ if (needRestart)
+ {
+ enableService();
+ }
}
}
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/LDAPReplicationDomainConfigChangeTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/LDAPReplicationDomainConfigChangeTest.java
index 3d68bf475b..50dd0177a7 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/LDAPReplicationDomainConfigChangeTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/LDAPReplicationDomainConfigChangeTest.java
@@ -45,6 +45,7 @@
import org.opends.server.core.ModifyOperation;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.common.AssuredMode;
+import org.opends.server.replication.service.ReplicationDomain;
import org.testng.annotations.Test;
/**
@@ -541,7 +542,8 @@ private static boolean waitForBlockedOn(Thread thread, Object monitor) throws Ex
private static Object serviceStateLockOf(LDAPReplicationDomain domain) throws Exception
{
- final Field serviceStateLock = LDAPReplicationDomain.class.getDeclaredField("serviceStateLock");
+ // Declared where the session lives, next to disableService()/enableService()
+ final Field serviceStateLock = ReplicationDomain.class.getDeclaredField("serviceStateLock");
serviceStateLock.setAccessible(true);
return serviceStateLock.get(domain);
}
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartTest.java
new file mode 100644
index 0000000000..04a9e5ab95
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartTest.java
@@ -0,0 +1,149 @@
+/*
+ * 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 org.assertj.core.api.Assertions.*;
+import static org.opends.server.TestCaseUtils.*;
+import static org.testng.Assert.*;
+
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+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.replication.ReplicationTestCase;
+import org.opends.server.replication.server.ReplServerFakeConfiguration;
+import org.opends.server.replication.server.ReplicationServer;
+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.
+ */
+@SuppressWarnings("javadoc")
+public class SessionRestartTest extends ReplicationTestCase
+{
+ private static final int RS_ID = 601;
+ private static final int DS_ID = 1;
+ private static final int GROUP_ID = 1;
+
+ @Test
+ public void aConfigurationChangeDoesNotStartTheSessionOfADisabledDomain() 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, "sessionRestartTestDisabledDb");
+
+ final DomainFakeCfg domainCfg = newDomainCfg(baseDN, rsPort);
+ domain = MultimasterReplication.createNewDomain(domainCfg);
+ domain.start();
+ assertTrue(domain.isConnected());
+
+ // The data this domain replicates is about to be replaced by an import or a restore.
+ domain.disable();
+ assertFalse(domain.isConnected());
+
+ changeEclIncludes(domain, domainCfg);
+
+ assertFalse(domain.isConnected(),
+ "a configuration change started the session of a disabled domain");
+ }
+ finally
+ {
+ if (domain != null)
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ }
+ remove(replicationServer);
+ }
+ }
+
+ @Test
+ public void aConfigurationChangeDoesNotStartTheSessionOfAShutDownDomain() 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, "sessionRestartTestShutdownDb");
+
+ final DomainFakeCfg domainCfg = newDomainCfg(baseDN, rsPort);
+ domain = MultimasterReplication.createNewDomain(domainCfg);
+ domain.start();
+ assertTrue(domain.isConnected());
+
+ domain.shutdown();
+ assertFalse(domain.isConnected());
+
+ changeEclIncludes(domain, domainCfg);
+
+ assertFalse(domain.isConnected(),
+ "a configuration change started the session of a domain which has shut down");
+ }
+ finally
+ {
+ 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.
+ */
+ private void changeEclIncludes(LDAPReplicationDomain domain, DomainFakeCfg domainCfg)
+ throws Exception
+ {
+ final SortedSet