external-software/github_OpenIdentityPlatform_OpenDJ.git

			@@ -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)