mirror of https://github.com/OpenIdentityPlatform/OpenDJ.git
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java
@@ -369,46 +369,6 @@ */ private final AtomicInteger consecutiveSessionRestarts = new AtomicInteger(); /** * Serialises the session of this domain being stopped and started again: the replay * thread which restarts it after a failed replay must not race the domain being * disabled for an import or a restore, or it would bring a broker and a listener * thread back up on a domain which is supposed to be down. * <p> * 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. * <p> * 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. * <p> * 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. * <p> @@ -907,22 +867,21 @@ 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) { @@ -954,7 +913,6 @@ if (needRestart) { enableService(); sessionGeneration++; } } } @@ -2525,7 +2483,6 @@ synchronized (serviceStateLock) { disableService(); sessionGeneration++; } } @@ -3709,13 +3666,13 @@ 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) { @@ -3729,21 +3686,18 @@ } 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++; } } @@ -4423,7 +4377,6 @@ */ disabled = true; disableService(); // This will cut the session and wake up the listener sessionGeneration++; awaitReplayDrained(); state.save(); state.clearInMemory(); @@ -4591,7 +4544,6 @@ try { enableService(); sessionGeneration++; started = true; } finally @@ -5390,23 +5342,6 @@ domCfg.getBaseDN(), stackTraceToSingleLineString(e)); } /** * {@inheritDoc} * <p> * 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<String> includeAttributes, Set<String> includeAttributesForDeletes) { synchronized (serviceStateLock) { super.changeConfig(includeAttributes, includeAttributesForDeletes); } } @Override public boolean isConfigurationChangeAcceptable( ReplicationDomainCfg configuration, List<LocalizableMessage> unacceptableReasons) opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java
@@ -46,6 +46,7 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import net.jcip.annotations.GuardedBy; import net.jcip.annotations.Immutable; import org.forgerock.i18n.LocalizableMessage; @@ -368,6 +369,44 @@ * session of this ReplicationDomain. */ private final Object sessionLock = new Object(); /** * Serialises the stopping and the starting of the session of this domain, so that a * pair of them is atomic: a session stopped so that the configuration it reads can be * changed must not be brought back in the middle of that change by something else. * <p> * 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. * <p> * 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. * <p> * 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 @@ * 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 broker first in order to prevent the listener from reconnecting - see OPENDJ-457. */ if (broker != null) { broker.stop(); } // Stop the listener thread if (listenerThread != null) { listenerThread.initiateShutdown(); try // Stop the listener thread if (listenerThread != null) { listenerThread.join(); listenerThread.initiateShutdown(); try { listenerThread.join(); } catch (InterruptedException e) { // Give up waiting. } listenerThread = null; } catch (InterruptedException e) { // Give up waiting. } listenerThread = null; } sessionGeneration++; } } @@ -3339,6 +3382,24 @@ } /** * 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}). * <p> * 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()}. * <p> * The Replication Service will restart from the point indicated by the @@ -3348,34 +3409,56 @@ * 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. * <p> * 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. * <p> * 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 @@ public void changeConfig(Set<String> includeAttributes, Set<String> includeAttributesForDeletes) { final boolean attrsModified = setEclIncludes( getServerId(), includeAttributes, includeAttributesForDeletes); if (attrsModified && broker != null) synchronized (serviceStateLock) { restartService(); final boolean attrsModified = setEclIncludes( getServerId(), includeAttributes, includeAttributesForDeletes); if (attrsModified && broker != null) { restartService(); } } } @@ -3397,14 +3483,20 @@ * Stops the session of this domain and starts it again, so that it comes up on the * configuration which has just changed. * <p> * 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(); } } /** @@ -3884,32 +3976,41 @@ 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(); } } }
@@ -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 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); } opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartTest.java
New file @@ -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<String> eclIncludes = new TreeSet<>(); eclIncludes.add("cn"); domainCfg.setExternalChangelogDomain( new ExternalChangelogDomainFakeCfg(true, eclIncludes, new TreeSet<String>())); final ConfigChangeResult ccr = domain.applyConfigurationChange(domainCfg); assertEquals(ccr.getResultCode(), ResultCode.SUCCESS, ccr.getMessages().toString()); // the restart the change asked for was refused, and said so rather than reported as applied assertTrue(ccr.adminActionRequired(), "the refused restart was reported as fully applied"); // the change did reach the external changelog configuration of the domain assertThat(domain.getEclIncludes()).contains("cn"); } private DomainFakeCfg newDomainCfg(DN baseDN, int rsPort) { final SortedSet<String> replServers = new TreeSet<>(); replServers.add("localhost:" + rsPort); return new DomainFakeCfg(baseDN, DS_ID, replServers, GROUP_ID); } private ReplicationServer createReplicationServer(int rsPort, String dbDir) throws Exception { final ReplServerFakeConfiguration conf = new ReplServerFakeConfiguration( rsPort, dbDir, 0, RS_ID, 0, 100, new TreeSet<String>(), GROUP_ID, 1000, 5000); return new ReplicationServer(conf); } }
