From 918aed143217bf25967a2832d567588ae2c5c10c Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Tue, 22 Sep 2026 10:44:48 +0000
Subject: [PATCH] [#1038] Pin the session lock and generation of #974, and the restart a configuration change does run (#1043)
---
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/SessionRestartTest.java | 259 ++++++++++++++++++++++++++++++++++++++++++++++++++-
1 files changed, 251 insertions(+), 8 deletions(-)
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
index 04a9e5a..96127cd 100644
--- 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
@@ -15,32 +15,52 @@
*/
package org.opends.server.replication.plugin;
+import static java.util.concurrent.TimeUnit.*;
import static org.assertj.core.api.Assertions.*;
import static org.opends.server.TestCaseUtils.*;
+import static org.opends.server.replication.plugin.LDAPReplicationDomain.*;
+import static org.opends.server.util.CollectionUtils.*;
import static org.testng.Assert.*;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.forgerock.opendj.config.server.ConfigChangeResult;
import org.forgerock.opendj.ldap.DN;
import org.forgerock.opendj.ldap.ResultCode;
import org.opends.server.TestCaseUtils;
+import org.opends.server.plugins.ShortCircuitPlugin;
import org.opends.server.replication.ReplicationTestCase;
+import org.opends.server.replication.common.CSNGenerator;
+import org.opends.server.replication.protocol.DeleteMsg;
+import org.opends.server.replication.server.DataServerHandler;
import org.opends.server.replication.server.ReplServerFakeConfiguration;
import org.opends.server.replication.server.ReplicationServer;
+import org.opends.server.replication.service.ReplicationBroker;
+import org.opends.server.replication.service.ReplicationDomain;
+import org.opends.server.types.Entry;
+import org.opends.server.types.OperationType;
+import org.opends.server.util.TestTimer;
+import org.opends.server.util.TestTimer.CallableVoid;
import org.testng.annotations.Test;
/**
- * Tests that a configuration change does not start the session of a domain which stopped
- * its own session: a domain which is shutting down, or whose data is being replaced, owns
- * its session and is the one which brings it back.
+ * Tests which sessions a configuration change restarts. A domain which stopped its own
+ * session - one which is shutting down, or whose data is being replaced - owns it and is the
+ * one which brings it back; one which is running its session is given a new one, and the
+ * replication server hears the change over it.
*/
@SuppressWarnings("javadoc")
public class SessionRestartTest extends ReplicationTestCase
{
private static final int RS_ID = 601;
private static final int DS_ID = 1;
+ /** The replica whose changes the domain under test replays. */
+ private static final int PUBLISHER_ID = 2;
private static final int GROUP_ID = 1;
@Test
@@ -63,10 +83,12 @@
domain.disable();
assertFalse(domain.isConnected());
- changeEclIncludes(domain, domainCfg);
+ final ConfigChangeResult ccr = changeEclIncludes(domain, domainCfg);
assertFalse(domain.isConnected(),
"a configuration change started the session of a disabled domain");
+ // the restart the change asked for was refused, and said so rather than reported as applied
+ assertTrue(ccr.adminActionRequired(), "the refused restart was reported as fully applied");
}
finally
{
@@ -97,10 +119,12 @@
domain.shutdown();
assertFalse(domain.isConnected());
- changeEclIncludes(domain, domainCfg);
+ final ConfigChangeResult ccr = changeEclIncludes(domain, domainCfg);
assertFalse(domain.isConnected(),
"a configuration change started the session of a domain which has shut down");
+ // the restart the change asked for was refused, and said so rather than reported as applied
+ assertTrue(ccr.adminActionRequired(), "the refused restart was reported as fully applied");
}
finally
{
@@ -113,11 +137,191 @@
}
/**
+ * The twin of the two above: a domain which is running its session is given a new one by
+ * the change, and the replication server is what tells. The attributes the external
+ * changelog includes are stored in the domain before the session is restarted for them,
+ * so the domain says the change is applied whether or not the restart ran; the
+ * replication server hears the list once, in the {@code StartSessionMsg} of a session, and
+ * a {@code DataServerHandler} which carries the new list is a session started after the
+ * change.
+ */
+ @Test
+ public void aConfigurationChangeRestartsTheSessionOfALiveDomain() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ ReplicationServer replicationServer = null;
+ LDAPReplicationDomain domain = null;
+ try
+ {
+ final int rsPort = TestCaseUtils.findFreePort();
+ replicationServer = createReplicationServer(rsPort, "sessionRestartTestLiveDb");
+
+ final DomainFakeCfg domainCfg = newDomainCfg(baseDN, rsPort);
+ domain = MultimasterReplication.createNewDomain(domainCfg);
+ domain.start();
+ assertTrue(domain.isConnected());
+ // What the session which is running told the replication server: no attribute at all.
+ assertThat(eclIncludesHeardBy(replicationServer, baseDN)).doesNotContain("cn");
+
+ final ConfigChangeResult ccr = changeEclIncludes(domain, domainCfg);
+
+ // the restart was run, so there is nothing to tell the administrator to wait for
+ assertFalse(ccr.adminActionRequired(),
+ "a restart which was run was reported as refused: " + ccr.getMessages());
+ final ReplicationServer rs = replicationServer;
+ new TestTimer.Builder().maxSleep(5, SECONDS).sleepTimes(100, MILLISECONDS).toTimer()
+ .repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ assertThat(eclIncludesHeardBy(rs, baseDN))
+ .as("the replication server was never told the new list, so no session was"
+ + " started after the change")
+ .contains("cn");
+ }
+ });
+ }
+ finally
+ {
+ if (domain != null)
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ }
+ remove(replicationServer);
+ }
+ }
+
+ /**
+ * A replay thread which could not apply a change stops the session so that the change
+ * is delivered again, and waits out a backoff before it starts the session back. The
+ * session it stopped is its claim; a configuration change which stops and starts the
+ * session while it waits leaves that claim stale, and the thread which comes back from
+ * its wait leaves the session which replaced the one it stopped alone.
+ * <p>
+ * What a stale claim which is not declined does to the session is nothing today: the
+ * broker and the listener thread are up already, and starting them again is a no-op. The
+ * one thing which tells a declined claim from one which was acted on is the generation of
+ * the session, which counts every start: the thread which acted on its stale claim leaves
+ * it one past where the restart which replaced its session left it.
+ */
+ @Test
+ public void aReplayThreadLeavesAloneTheSessionWhichReplacedTheOneItStopped() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ ReplicationServer replicationServer = null;
+ LDAPReplicationDomain domain = null;
+ ReplicationBroker publisher = null;
+ try
+ {
+ final int rsPort = TestCaseUtils.findFreePort();
+ replicationServer = createReplicationServer(rsPort, "sessionRestartTestStaleClaimDb");
+ domain = MultimasterReplication.createNewDomain(newDomainCfg(baseDN, rsPort));
+ domain.start();
+ assertTrue(domain.isConnected());
+
+ final Entry entry = TestCaseUtils.addEntry(
+ "dn: cn=stale claim," + baseDN,
+ "objectClass: top",
+ "objectClass: person",
+ "cn: stale claim",
+ "sn: claim");
+ final String uuid = getEntry(entry.getName(), 1, true).parseAttribute("entryuuid").asString();
+ publisher = openReplicationSession(baseDN, PUBLISHER_ID, 100, rsPort, 1000);
+
+ final LDAPReplicationDomain replica = domain;
+ final Object serviceStateLock = serviceStateLockOf(replica);
+ try
+ {
+ /*
+ * The backend refuses the delete for longer than the replay is retried in place,
+ * so the replay thread stops the session for the change to be delivered again -
+ * and serves it once the session was restarted for it, so that the change is
+ * applied over the session started next and asks for no restart of its own.
+ */
+ ShortCircuitPlugin.registerShortCircuit(OperationType.DELETE, "PreParse",
+ ResultCode.UNAVAILABLE.intValue(), IN_PLACE_REPLAY_ATTEMPTS + 2);
+ publisher.publish(new DeleteMsg(entry.getName(), new CSNGenerator(PUBLISHER_ID, 0).newCSN(), uuid));
+
+ /*
+ * The lock is taken and let go until it is taken while the session is stopped: the
+ * replay thread stopped it under the lock and let the lock go for its wait, so the
+ * claim it holds is standing, and the lock is held for the rest of that wait. The
+ * restart below runs while the claim is standing however short the wait is.
+ */
+ final long claim;
+ final long replaced;
+ final long deadline = System.currentTimeMillis() + SECONDS.toMillis(30);
+ while (true)
+ {
+ synchronized (serviceStateLock)
+ {
+ if (!replica.isConnected())
+ {
+ claim = sessionGenerationOf(replica);
+ // Something else restarts the session while the replay thread waits: the
+ // attributes the external changelog includes are changed.
+ replica.changeConfig(newTreeSet("cn"), new TreeSet<String>());
+ replaced = sessionGenerationOf(replica);
+ break;
+ }
+ }
+ assertTrue(System.currentTimeMillis() < deadline, "the failed replay never stopped the session");
+ Thread.sleep(5);
+ }
+ assertNotEquals(replaced, claim, "the restart left the generation of the session where it was");
+ assertTrue(replica.isConnected(), "the restart left the domain without a session");
+
+ // The change is delivered again over the session which replaced the stopped one,
+ // and applied; the replay thread comes back from its wait and finds its claim stale.
+ assertNull(getEntry(entry.getName(), 30000, false),
+ "the change the session was stopped for was not delivered again over the session which replaced it");
+ new TestTimer.Builder().maxSleep(30, SECONDS).sleepTimes(100, MILLISECONDS).toTimer()
+ .repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ assertFalse(isRecoveringFromAReplayFailure(replica),
+ "the replay thread never came back from the restart it asked for");
+ }
+ });
+
+ synchronized (serviceStateLock)
+ {
+ assertEquals(sessionGenerationOf(replica), replaced,
+ "the replay thread started the session which replaced the one it stopped,"
+ + " as if its claim on the stopped one were still current");
+ }
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse");
+ }
+ }
+ finally
+ {
+ if (publisher != null)
+ {
+ publisher.stop();
+ }
+ if (domain != null)
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ }
+ remove(replicationServer);
+ }
+ }
+
+ /**
* Applies a configuration change which changes the attributes the external changelog
* includes. The domain hands it to {@code ExternalChangelogDomain}, which asks for the
* session to be restarted so that the replication server hears the new list.
+ *
+ * @return the result of the change, so that each test says whether the restart it asked
+ * for was to be run or refused
*/
- private void changeEclIncludes(LDAPReplicationDomain domain, DomainFakeCfg domainCfg)
+ private ConfigChangeResult changeEclIncludes(LDAPReplicationDomain domain, DomainFakeCfg domainCfg)
throws Exception
{
final SortedSet<String> eclIncludes = new TreeSet<>();
@@ -127,10 +331,49 @@
final ConfigChangeResult ccr = domain.applyConfigurationChange(domainCfg);
assertEquals(ccr.getResultCode(), ResultCode.SUCCESS, ccr.getMessages().toString());
- // the restart the change asked for was refused, and said so rather than reported as applied
- assertTrue(ccr.adminActionRequired(), "the refused restart was reported as fully applied");
// the change did reach the external changelog configuration of the domain
assertThat(domain.getEclIncludes()).contains("cn");
+ return ccr;
+ }
+
+ /**
+ * The attributes the replication server was told the domain includes in the external
+ * changelog, by the session it holds for the domain right now.
+ */
+ private static Set<String> eclIncludesHeardBy(ReplicationServer replicationServer, DN baseDN)
+ {
+ final DataServerHandler ds =
+ replicationServer.getReplicationServerDomain(baseDN).getConnectedDSs().get(DS_ID);
+ assertNotNull(ds, "the replication server holds no session of the domain");
+ return ds.toDSInfo().getEclIncludes();
+ }
+
+ private static Object serviceStateLockOf(LDAPReplicationDomain domain) throws Exception
+ {
+ // Declared where the session lives, next to disableService()/enableService()
+ final Field serviceStateLock = ReplicationDomain.class.getDeclaredField("serviceStateLock");
+ serviceStateLock.setAccessible(true);
+ return serviceStateLock.get(domain);
+ }
+
+ /** Read under {@code serviceStateLock}, as {@code getSessionGeneration()} asks. */
+ private static long sessionGenerationOf(LDAPReplicationDomain domain) throws Exception
+ {
+ final Method getSessionGeneration = ReplicationDomain.class.getDeclaredMethod("getSessionGeneration");
+ getSessionGeneration.setAccessible(true);
+ return (Long) getSessionGeneration.invoke(domain);
+ }
+
+ /**
+ * Whether a replay thread of the domain is restarting the session for a change it could
+ * not apply: set by the thread which took that recovery on, and cleared once the restart
+ * it asked for is done with - run, or declined on a stale claim.
+ */
+ private static boolean isRecoveringFromAReplayFailure(LDAPReplicationDomain domain) throws Exception
+ {
+ final Field replayFailureRecovery = LDAPReplicationDomain.class.getDeclaredField("replayFailureRecovery");
+ replayFailureRecovery.setAccessible(true);
+ return ((AtomicBoolean) replayFailureRecovery.get(domain)).get();
}
private DomainFakeCfg newDomainCfg(DN baseDN, int rsPort)
--
Gitblit v1.10.0