[#1038] Pin the session lock and generation of #974, and the restart a conf...

Valery Kharseko

yesterday 918aed143217bf25967a2832d567588ae2c5c10c

[#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/LDAPReplicationDomainConfigChangeTest.java

@@ -241,6 +241,58 @@
    }
  }

  /**
   * The fractional twin of the test above: the fractional configuration is what the session
   * filters the changes it receives on, and a change of it restarts the session. A domain
   * which owns its session is given no restart, and the configuration is applied all the
   * same - the session its enable() starts filters on it - and the change says a session
   * is waiting for it.
   */
  @Test
  public void fractionalConfigurationIsAppliedToADomainWhichOwnsItsSession() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    try
    {
      final SortedSet<String> replServers = unstartedReplicationServer();
      final LDAPReplicationDomain domain = startDomain(new DomainFakeCfg(baseDN, SERVER_ID, replServers));
      assertFalse(domain.getFractionalConfig().isFractional());

      // Disabled is what an online import leaves the domain: it owns its session, so this
      // change is applied to it without a session being restarted for it.
      domain.disable();
      waitForListenerThread(baseDN, false);

      final DomainFakeCfg fractional = new DomainFakeCfg(baseDN, SERVER_ID, replServers);
      // The same as the configuration in place, or the broker would restart the session for
      // the heartbeat interval and the change would no longer be the fractional one alone.
      fractional.setHeartbeatInterval(HEARTBEAT_INTERVAL_IN_MS);
      fractional.getFractionalExclude().add("*:description");
      final ConfigChangeResult ccr = domain.applyConfigurationChange(fractional);

      assertEquals(ccr.getResultCode(), ResultCode.SUCCESS, ccr.getMessages().toString());
      assertFalse(hasListenerThread(baseDN),
          "the change started a session on a domain which was disabled for a total update");
      final LDAPReplicationDomain.FractionalConfig applied = domain.getFractionalConfig();
      assertTrue(applied.isFractional(),
          "the fractional configuration was dropped although the change reported success");
      assertTrue(applied.isFractionalExclusive(),
          "the fractional configuration was applied in the wrong mode");
      assertTrue(applied.getFractionalAllClassesAttributes().contains("description"),
          "the attribute the change excludes was dropped: " + applied.getFractionalAllClassesAttributes());
      assertTrue(ccr.adminActionRequired(),
          "the fractional configuration is what a session filters on, and this domain was"
              + " given no session to filter over");

      // Left as a total update leaves it: enabled back, on the configuration it was given.
      domain.enable();
    }
    finally
    {
      MultimasterReplication.deleteDomain(baseDN);
    }
  }

  @Test
  public void changeIsRefusedWhenTheExternalChangelogDomainRejectsIt() throws Exception
  {

 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)

 opendj-server-legacy/src/test/java/org/opends/server/replication/service/ReplicationDomainTest.java

@@ -238,6 +238,85 @@
  }

  /**
   * The session generation is the identity of a session: a replay thread which stopped the
   * session and let the lock go compares the generation it read then with the one it reads
   * when it comes back, and starts the session only if the two are the same. Both halves
   * of a restart have to move it, or a session stopped or started by something else in the
   * meantime would look like the one that thread stopped.
   */
  @Test
  public void everyStopAndEveryStartOfTheSessionIsCounted() throws Exception
  {
    final DN testService = DN.valueOf("o=test");
    ReplicationServer replServer = null;
    FakeReplicationDomain domain = null;
    try
    {
      final int replServerPort = TestCaseUtils.findFreePort();
      replServer = createReplicationServer(1, replServerPort, "ReplicationDomainTestDbGeneration", 100);
      domain = new FakeReplicationDomain(testService, 2, newTreeSet("localhost:" + replServerPort), 1000, 1);

      final long live = sessionGenerationOf(domain);
      domain.disableService();
      final long stopped = sessionGenerationOf(domain);
      assertNotEquals(stopped, live, "stopping the session left its generation where it was");

      domain.enableService();
      final long started = sessionGenerationOf(domain);
      assertNotEquals(started, stopped, "starting the session left its generation where it was");
    }
    finally
    {
      disable(domain);
      remove(replServer);
    }
  }

  /**
   * What {@code restartSession()} of the LDAP domain does, minus the wait: the session is
   * stopped, the generation is read as the claim on it, and the lock is let go. A
   * {@link ReplicationDomain#restartService()} run by something else in the meantime - a
   * configuration change - has to leave that claim stale, or the thread which comes back
   * from its wait could not tell the session it stopped from the one which replaced it.
   */
  @Test
  public void aClaimOnAStoppedSessionIsStaleOnceSomethingElseRestartedIt() throws Exception
  {
    final DN testService = DN.valueOf("o=test");
    ReplicationServer replServer = null;
    FakeReplicationDomain domain = null;
    try
    {
      final int replServerPort = TestCaseUtils.findFreePort();
      replServer = createReplicationServer(1, replServerPort, "ReplicationDomainTestDbStaleClaim", 100);
      domain = new FakeReplicationDomain(testService, 2, newTreeSet("localhost:" + replServerPort), 1000, 1);

      domain.disableService();
      final long claim = sessionGenerationOf(domain);

      domain.restartService();

      assertNotEquals(sessionGenerationOf(domain), claim,
          "a restart of the session by something else left the generation where it was,"
              + " so the claim of the thread which stopped it still looks current");
    }
    finally
    {
      disable(domain);
      remove(replServer);
    }
  }

  /** Read under the lock, as {@link ReplicationDomain#getSessionGeneration()} asks. */
  private static long sessionGenerationOf(ReplicationDomain domain)
  {
    synchronized (domain.serviceStateLock)
    {
      return domain.getSessionGeneration();
    }
  }

  /**
   * Publish performance test.
   * The test loops calling the publish methods of the ReplicationDomain.
   * It should not be enabled by default as it will use a lot of time.