ReplicationTestCase.java: Change entryList and configEntryList from LinkedL...

Jean-Noel Rouvignac

03.59.2013 cbf6bfd149ce305652be0aac68d210778b5cbba6

ReplicationTestCase.java:
Change entryList and configEntryList from LinkedList to Set and renamed them to entriesToCleanup and configEntriesToCleanup.
Extracted method connect() and waitForSpecificMsg() + Changed the return type of the existing waitForSpecificMsg() methods.
In configureReplication(), added 2 String parameters + extracted method addSynchroServerEntry() and addConfigEntry().
Simplified code a lot.

*Test[Case].java:
Consequence of the changes to ReplicationTestCase.
Factorized code with ReplicationTestCase.
Various cleanups:
- used assertj
- used StaticUtils.close()
- made fields private
- used foreach
- used assertEquals() instead of assertTrue()
- removed try/catch/fail test anti pattern.
- let exceptions go up the stack

ProtocolWindowTest.java:
Extracted method searchNbMonitorEntries().

 opends/tests/unit-tests-testng/src/server/org/opends/server/replication/ChangeNumberControlPluginTestCase.java

@@ -34,6 +34,7 @@
import org.opends.server.TestCaseUtils;
import org.opends.server.tools.LDAPModify;
import org.opends.server.types.DN;
import org.opends.server.util.StaticUtils;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@@ -80,7 +81,6 @@
        + "ds-cfg-replication-port: " + replServerPort + "\n"
        + "ds-cfg-replication-db-directory: ChangeNumberControlDbTest\n"
        + "ds-cfg-replication-server-id: 103\n";
    replServerEntry = TestCaseUtils.entryFromLdifString(replServerLdif);

    // suffix synchronized
    String testName = "changeNumberControlPluginTestCase";
@@ -93,9 +93,8 @@
        + "ds-cfg-replication-server: localhost:" + replServerPort + "\n"
        + "ds-cfg-server-id: 1\n"
        + "ds-cfg-receive-status: true\n";
    synchroServerEntry = TestCaseUtils.entryFromLdifString(synchroServerLdif);

    configureReplication();
    configureReplication(replServerLdif, synchroServerLdif);
  }

  @DataProvider(name = "operations")
@@ -152,16 +151,25 @@

  private boolean isCsnLinePresent(String file) throws Exception {
    FileReader fr = new FileReader(file);
    try
    {
    BufferedReader br = new BufferedReader(fr);
    String line = null;
    boolean found = false;
    while ((line = br.readLine()) != null) {
      while ((line = br.readLine()) != null)
      {
      if (line.contains(INFO_CHANGE_NUMBER_CONTROL_RESULT.get("%s","%s")
                            .toString().split("%s")[1])) {
            .toString().split("%s")[1]))
        {
        found = true;
      }
    }
    return (found);
      return found;
    }
    finally
    {
      StaticUtils.close(fr);
    }
  }

}

 opends/tests/unit-tests-testng/src/server/org/opends/server/replication/GenerationIdTest.java

@@ -31,6 +31,7 @@
import java.net.SocketTimeoutException;
import java.util.*;

import org.assertj.core.api.Assertions;
import org.opends.messages.Category;
import org.opends.messages.Message;
import org.opends.messages.Severity;
@@ -368,11 +369,7 @@
      // have been called
      assertNull(synchroServerEntry);

      synchroServerEntry = TestCaseUtils.entryFromLdifString(synchroServerLdif);
      DirectoryServer.getConfigHandler().addEntry(synchroServerEntry, null);
      assertNotNull(DirectoryServer.getConfigEntry(synchroServerEntry.getDN()),
        "Unable to add the synchronized server");
      configEntryList.add(synchroServerEntry.getDN());
      addSynchroServerEntry(synchroServerLdif);

      int waitCo=0;
      LDAPReplicationDomain doToco=null;
@@ -399,8 +396,7 @@
  {
    {
      // suffix synchronized
      String synchroServerStringDN = "cn=" + testName + ", cn=domains," +
      SYNCHRO_PLUGIN_DN;
      String synchroServerStringDN = "cn=" + testName + ", cn=domains," + SYNCHRO_PLUGIN_DN;
      // Must have called connectServer1ToChangelog previously
      assertNotNull(synchroServerEntry);

@@ -417,7 +413,7 @@
        "Unable to delete the synchronized domain");
      synchroServerEntry = null;

      configEntryList.remove(configEntryList.indexOf(synchroServerDN));
      configEntriesToCleanup.remove(synchroServerDN);

      LDAPReplicationDomain replDomainToDis = null;
      try
@@ -751,13 +747,11 @@

      // Broker 2 and 3 should receive 1 change status message to order them
      // to enter the bad gen id status
      ChangeStatusMsg csMsg = (ChangeStatusMsg)waitForSpecificMsg(broker2,
        ChangeStatusMsg.class.getName());
      ChangeStatusMsg csMsg = waitForSpecificMsg(broker2, ChangeStatusMsg.class);
      assertEquals(csMsg.getRequestedStatus(), ServerStatus.BAD_GEN_ID_STATUS,
          "Broker 2 connection is expected to receive 1 ChangeStatusMsg"
              + " to enter the bad gen id status" + csMsg);
      csMsg = (ChangeStatusMsg)waitForSpecificMsg(broker3,
        ChangeStatusMsg.class.getName());
      csMsg = waitForSpecificMsg(broker3, ChangeStatusMsg.class);
      assertEquals(csMsg.getRequestedStatus(), ServerStatus.BAD_GEN_ID_STATUS,
          "Broker 2 connection is expected to receive 1 ChangeStatusMsg"
              + " to enter the bad gen id status" + csMsg);
@@ -867,11 +861,11 @@

      debugInfo("Verify that DS2 receives the add message stored in RS1 DB");
      msg = broker2.receive();
      assertTrue(msg instanceof AddMsg, "Expected to receive an AddMsg but received: " + msg);
      Assertions.assertThat(msg).isInstanceOf(AddMsg.class);

      debugInfo("Verify that DS3 receives the add message stored in RS1 DB");
      msg = broker3.receive();
      assertTrue(msg instanceof AddMsg, "Expected to receive an AddMsg but received: " + msg);
      Assertions.assertThat(msg).isInstanceOf(AddMsg.class);

      debugInfo("DS2 is publishing a change and RS1 must store this change, DS3 must receive it.");
      emsg = createAddMsg();
@@ -882,6 +876,7 @@

      /* expected */
      msg = broker3.receive();
      Assertions.assertThat(msg).isInstanceOf(AddMsg.class);
      AddMsg rcvmsg = (AddMsg)msg;
      assertEquals(rcvmsg.getCSN(), emsg.getCSN());


 opends/tests/unit-tests-testng/src/server/org/opends/server/replication/InitOnLineTest.java

@@ -200,8 +200,8 @@
        "ds-task-initialize-replica-server-id: all");
  }

  /** Tests that entries have been written in the db */
  private void testEntriesInDb()
  /** Tests that entries have been written in the db. */
  private void testEntriesInDb() throws Exception
  {
    log("TestEntriesInDb");
    short found = 0;
@@ -215,15 +215,7 @@

      log("Search Entry: " + dn);

      DN entryDN = null;
      try
      {
        entryDN = DN.decode(dn);
      }
      catch(Exception e)
      {
        log("TestEntriesInDb/" + e);
      }
      DN entryDN = DN.decode(dn);

      try
      {
@@ -348,16 +340,13 @@
  {
    for (String ldifEntry : updatedEntries)
    {
      Entry entry = TestCaseUtils.entryFromLdifString(ldifEntry);
      addTestEntryToDB(entry);
      // They will be removed at the end of the test
      entryList.addLast(entry.getDN());
      addTestEntryToDB(TestCaseUtils.entryFromLdifString(ldifEntry));
    }
    log("addTestEntriesToDB : " + updatedEntries.length
        + " successfully added to DB");
  }

  private void addTestEntryToDB(Entry entry)
  private void addTestEntryToDB(final Entry entry)
  {
    AddOperation addOp =
        new AddOperationBasis(connection, InternalClientConnection
@@ -371,8 +360,7 @@
      log("addEntry: Failed" + addOp.getResultCode());
    }

    // They will be removed at the end of the test
    entryList.addLast(entry.getDN());
    entriesToCleanup.add(entry.getDN());
  }

  /**
@@ -602,14 +590,9 @@

    TestCaseUtils.clearJEBackend(false, "userRoot", EXAMPLE_DN);

    synchroServerEntry = TestCaseUtils.entryFromLdifString(synchroServerLdif);
    DirectoryServer.getConfigHandler().addEntry(synchroServerEntry, null);
    assertNotNull(DirectoryServer.getConfigEntry(synchroServerEntry.getDN()),
        "Unable to add the synchronized server");
    configEntryList.add(synchroServerEntry.getDN());
    addSynchroServerEntry(synchroServerLdif);

    replDomain = LDAPReplicationDomain.retrievesReplicationDomain(baseDN);

    assertTrue(!replDomain.ieRunning(),
        "ReplicationDomain: Import/Export is not expected to be running");
  }
@@ -1268,8 +1251,7 @@
      addTask(taskInit, ResultCode.SUCCESS, null);

      waitTaskState(taskInit, TaskState.STOPPED_BY_ERROR,
        ERR_NO_REACHABLE_PEER_IN_THE_DOMAIN.get(
            baseDN.toString(), "20"));
          ERR_NO_REACHABLE_PEER_IN_THE_DOMAIN.get(baseDN.toString(), "20"));

      // Test 2
      taskInit = TestCaseUtils.makeEntry(
@@ -1349,18 +1331,21 @@
    String testCase = "InitializeStopped";
    fail(testCase + " NYI");
  }

  @Test(enabled=false)
  public void initializeTargetStopped() throws Exception
  {
    String testCase = "InitializeTargetStopped";
    fail(testCase + " NYI");
  }

  @Test(enabled=false)
  public void initializeCompressed() throws Exception
  {
    String testCase = "InitializeStopped";
    fail(testCase + " NYI");
  }

  @Test(enabled=false)
  public void initializeTargetEncrypted() throws Exception
  {
@@ -1429,8 +1414,7 @@
      ErrorMsg msg = new ErrorMsg(server1ID, 1, Message.EMPTY);
      server2.publish(msg);

      waitTaskState(taskInit, TaskState.STOPPED_BY_ERROR,
        null);
      waitTaskState(taskInit, TaskState.STOPPED_BY_ERROR, null);

      log("Successfully ending " + testCase);
    } finally
@@ -1481,8 +1465,6 @@

  /**
   * Clean up the environment.
   *
   * @throws Exception If the environment could not be set up.
   */
  @AfterClass
  @Override

 opends/tests/unit-tests-testng/src/server/org/opends/server/replication/ProtocolWindowTest.java

@@ -28,10 +28,10 @@
package org.opends.server.replication;

import java.net.SocketTimeoutException;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;

import org.assertj.core.api.Assertions;
import org.opends.messages.Category;
import org.opends.messages.Message;
import org.opends.messages.Severity;
@@ -42,7 +42,6 @@
import org.opends.server.core.ModifyOperation;
import org.opends.server.protocols.internal.InternalClientConnection;
import org.opends.server.protocols.internal.InternalSearchOperation;
import org.opends.server.protocols.ldap.LDAPFilter;
import org.opends.server.replication.protocol.AddMsg;
import org.opends.server.replication.protocol.ReplicationMsg;
import org.opends.server.replication.server.ReplServerFakeConfiguration;
@@ -55,6 +54,7 @@

import static org.opends.server.TestCaseUtils.*;
import static org.opends.server.loggers.ErrorLogger.*;
import static org.opends.server.types.SearchScope.*;
import static org.testng.Assert.*;

/**
@@ -70,7 +70,7 @@
  /**
   * A "person" entry
   */
  protected Entry personEntry;
  private Entry personEntry;
  private int replServerPort;


@@ -112,7 +112,7 @@
    DirectoryServer.getConfigHandler().addEntry(repDomainEntry, null);
    assertNotNull(DirectoryServer.getConfigEntry(repDomainEntry.getDN()),
          "Unable to add the synchronized server");
    configEntryList.add(repDomainEntry.getDN());
    configEntriesToCleanup.add(repDomainEntry.getDN());

    ReplicationBroker broker = openReplicationSession(baseDN, 12,
        WINDOW_SIZE, replServerPort, 1000, true);
@@ -125,8 +125,8 @@
       * configuration is working.
       */
      Thread.sleep(2000);
      assertTrue(checkWindows(WINDOW_SIZE));
      assertTrue(checkChangelogQueueSize(REPLICATION_QUEUE_SIZE));
      assertEquals(checkWindows(WINDOW_SIZE), 3);
      assertEquals(checkChangelogQueueSize(REPLICATION_QUEUE_SIZE), 2);

      // Create an Entry (add operation) that will be later used in the test.
      Entry tmp = personEntry.duplicate(false);
@@ -142,8 +142,7 @@

      // Check if the client has received the MSG
      ReplicationMsg msg = broker.receive();
      assertTrue(msg instanceof AddMsg,
        "The received Replication message is not an ADD msg");
      Assertions.assertThat(msg).isInstanceOf(AddMsg.class);
      AddMsg addMsg =  (AddMsg) msg;

      Operation receivedOp = addMsg.createOperation(connection);
@@ -206,20 +205,20 @@
    }
  }

  private int searchNbMonitorEntries(String filterString) throws Exception
  {
    InternalSearchOperation op = connection.processSearch("cn=monitor", WHOLE_SUBTREE, filterString);
    assertEquals(op.getResultCode(), ResultCode.SUCCESS);
    return op.getEntriesSent();
  }

  /**
   * Check that the ReplicationServer queue size has correctly been configured
   * by reading the monitoring information.
   * @throws LDAPException
   */
  private boolean checkChangelogQueueSize(int changelog_queue_size)
          throws LDAPException
  private int checkChangelogQueueSize(int changelog_queue_size) throws Exception
  {
    InternalSearchOperation op = connection.processSearch(
        ByteString.valueOf("cn=monitor"),
        SearchScope.WHOLE_SUBTREE, LDAPFilter.decode(
            "(max-waiting-changes=" +  changelog_queue_size + ")"));
    assertEquals(op.getResultCode(), ResultCode.SUCCESS);
    return (op.getEntriesSent() == 2);
    return searchNbMonitorEntries("(max-waiting-changes=" + changelog_queue_size + ")");
  }

  /**
@@ -227,14 +226,9 @@
   * by reading the monitoring information and checking
   * that we do have 2 entries with the configured max-rcv-window.
   */
  private boolean checkWindows(int windowSize) throws LDAPException
  private int checkWindows(int windowSize) throws Exception
  {
    InternalSearchOperation op = connection.processSearch(
        ByteString.valueOf("cn=monitor"),
        SearchScope.WHOLE_SUBTREE,
        LDAPFilter.decode("(max-rcv-window=" + windowSize + ")"));
    assertEquals(op.getResultCode(), ResultCode.SUCCESS);
    return (op.getEntriesSent() == 3);
    return searchNbMonitorEntries("(max-rcv-window=" + windowSize + ")");
  }

  /**
@@ -245,36 +239,11 @@
   */
  private void searchUpdateSent() throws Exception
  {
    InternalSearchOperation op = connection.processSearch(
        ByteString.valueOf("cn=monitor"),
        SearchScope.WHOLE_SUBTREE,
        LDAPFilter.decode("(sent-updates=" + WINDOW_SIZE + ")"));
    assertEquals(searchNbMonitorEntries("(sent-updates=" + WINDOW_SIZE + ")"), 1);

    assertEquals(op.getResultCode(), ResultCode.SUCCESS);
    assertEquals(op.getEntriesSent(), 1,
        "Entries#=" + op.getEntriesSent());

    op = connection.processSearch(
        ByteString.valueOf("cn=monitor"),
        SearchScope.WHOLE_SUBTREE,
        LDAPFilter.decode("(missing-changes=" +
            (REPLICATION_QUEUE_SIZE + WINDOW_SIZE) + ")"));
    assertEquals(op.getResultCode(), ResultCode.SUCCESS);

    Iterator<SearchResultEntry> entriesit = op.getSearchEntries().iterator();
    while(entriesit.hasNext())
    {
      SearchResultEntry e = entriesit.next();
      Iterator<Attribute> attit = e.getAttributes().iterator();
      while (attit.hasNext())
      {
        Attribute attr = attit.next();
        logError(Message.raw(Category.SYNC, Severity.INFORMATION,
        e.getDN() + "= " + attr.getName() + " " + attr.iterator()
        .next().getValue().toString()));
      }
    }
    assertEquals(op.getEntriesSent(), 1, "Entries#=" + op.getEntriesSent());
    final int nb = searchNbMonitorEntries(
        "(missing-changes=" + (REPLICATION_QUEUE_SIZE + WINDOW_SIZE) + ")");
    assertEquals(nb, 1);
  }

  /**

 opends/tests/unit-tests-testng/src/server/org/opends/server/replication/ReSyncTest.java

@@ -51,6 +51,7 @@
/**
 * Test re-synchronization after after backup/restore and LDIF import.
 */
@SuppressWarnings("javadoc")
public class ReSyncTest extends ReplicationTestCase
{
  /** The tracer object for the debug logger */
@@ -106,7 +107,6 @@
        + "ds-cfg-replication-port:" + replServerPort + "\n"
        + "ds-cfg-replication-db-directory: ReSyncTest\n"
        + "ds-cfg-replication-server-id: 104\n";
    replServerEntry = TestCaseUtils.entryFromLdifString(replServerLdif);

    // suffix synchronized
    String reSyncTest = "reSyncTest";
@@ -118,9 +118,8 @@
        + "ds-cfg-base-dn: " + EXAMPLE_DN + "\n"
        + "ds-cfg-replication-server: localhost:"+ replServerPort + "\n"
        + "ds-cfg-server-id: 123\n";
    synchroServerEntry = TestCaseUtils.entryFromLdifString(domainLdif);

    configureReplication();
    configureReplication(replServerLdif, domainLdif);

    // Give some time to the replication to setup
    Thread.sleep(1000);
@@ -140,17 +139,15 @@
   */
  private ResultCode addEntry(String entryString) throws Exception
  {
    Entry entry;
    AddOperationBasis addOp;
    entry = TestCaseUtils.entryFromLdifString(entryString);
    addOp = new AddOperationBasis(connection,
    Entry entry = TestCaseUtils.entryFromLdifString(entryString);
    AddOperationBasis addOp = new AddOperationBasis(connection,
       InternalClientConnection.nextOperationID(), InternalClientConnection
       .nextMessageID(), null, entry.getDN(), entry.getObjectClasses(),
       entry.getUserAttributes(), entry.getOperationalAttributes());
    addOp.setInternalOperation(true);
    addOp.run();

    entryList.add(entry.getDN());
    entriesToCleanup.add(entry.getDN());
    return addOp.getResultCode();
  }

@@ -170,7 +167,8 @@
    // Delete the entry we are going to use to make sure that
    // we do test something.

    connection.processDelete(DN.decode("dc=fooUniqueName1," + EXAMPLE_DN));
    DN entryDN = DN.decode("dc=fooUniqueName1," + EXAMPLE_DN);
    connection.processDelete(entryDN);

    task("dn: ds-task-id=" + UUID.randomUUID()
        +  ",cn=Scheduled Tasks,cn=Tasks\n"
@@ -197,10 +195,10 @@

    debugInfo("testResyncAfterRestore: restore done");

   if (getEntry(DN.decode("dc=fooUniqueName1," + EXAMPLE_DN), 30000, true) == null)
     fail("The Directory has not been resynchronized after the restore.");
    assertNotNull(getEntry(entryDN, 30000, true),
        "The Directory has not been resynchronized after the restore.");

   connection.processDelete(DN.decode("dc=fooUniqueName1," + EXAMPLE_DN));
    connection.processDelete(entryDN);
  }

  /**
@@ -218,7 +216,8 @@

    // delete the entry we are going to use to make sure that
    // we do test something.
    connection.processDelete(DN.decode("dc=fooUniqueName2," + EXAMPLE_DN));
    DN entryDN = DN.decode("dc=fooUniqueName2," + EXAMPLE_DN);
    connection.processDelete(entryDN);

    String path = reSyncTempDir.getAbsolutePath() + File.pathSeparator +
            "ReSynchTest";
@@ -249,14 +248,12 @@

    debugInfo("testResyncAfterImport: import done");

   if (getEntry(DN.decode("dc=fooUniqueName2," + EXAMPLE_DN), 30000, true) == null)
     fail("The Directory has not been resynchronized after the restore.");
    assertNotNull(getEntry(entryDN, 30000, true),
        "The Directory has not been resynchronized after the restore.");
  }

  /**
   * Clean up the environment.
   *
   * @throws Exception If the environment could not be set up.
   */
  @AfterClass
  @Override

 opends/tests/unit-tests-testng/src/server/org/opends/server/replication/ReplicationTestCase.java

@@ -45,7 +45,6 @@
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.protocols.internal.InternalClientConnection;
import org.opends.server.protocols.internal.InternalSearchOperation;
import org.opends.server.protocols.ldap.LDAPFilter;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.plugin.GenerationIdChecksum;
import org.opends.server.replication.plugin.LDAPReplicationDomain;
@@ -68,7 +67,9 @@
import static org.opends.server.config.ConfigConstants.*;
import static org.opends.server.loggers.ErrorLogger.*;
import static org.opends.server.loggers.debug.DebugLogger.*;
import static org.opends.server.util.StaticUtils.*;
import static org.opends.server.protocols.internal.InternalClientConnection.*;
import static org.opends.server.types.ResultCode.*;
import static org.opends.server.types.SearchScope.*;
import static org.testng.Assert.*;

/**
@@ -100,14 +101,13 @@
  */
  protected InternalClientConnection connection;

  /**
   * Created entries that need to be deleted for cleanup
   */
  protected LinkedList<DN> entryList = new LinkedList<DN>();
  protected LinkedList<DN> configEntryList = new LinkedList<DN>();
  /** Created entries that will be deleted on class cleanup. */
  protected final Set<DN> entriesToCleanup = new HashSet<DN>();
  /** Created config entries that will be deleted on class cleanup. */
  protected final Set<DN> configEntriesToCleanup = new HashSet<DN>();

  /** Replicated suffix (replication domain). */
  protected Entry synchroServerEntry;

  protected Entry replServerEntry;

  /**
@@ -236,14 +236,18 @@
    ReplicationBroker broker = new ReplicationBroker(replicationDomain,
        state, baseDN, serverId, window_size,
        generationId, 100000, getReplSessionSecurity(), (byte)1, 500);
    List<String> servers = new ArrayList<String>(1);
    servers.add("localhost:" + port);
    broker.start(servers);
    connect(broker, port, timeout);
    return broker;
  }

  private void connect(ReplicationBroker broker, int port, int timeout) throws Exception
  {
    broker.start(Collections.singletonList("localhost:" + port));
    // give some time to the broker to connect to the replicationServer.
    checkConnection(30, broker, port);

    if (timeout != 0)
      broker.setSoTimeout(timeout);
    checkConnection(30, broker, port); // give some time to the broker to connect
                                       // to the replicationServer.
    return broker;
  }

  /**
@@ -251,44 +255,30 @@
   * replication server. Waits for connection to be ok up to secTimeout seconds
   * before failing.
   */
  protected void checkConnection(int secTimeout, ReplicationBroker rb, int rsPort)
  protected void checkConnection(int secTimeout, ReplicationBroker rb, int rsPort) throws Exception
  {
    int nSec = 0;

    // Go out of the loop only if connection is verified or if timeout occurs
    while (true)
    {
      // Test connection
      boolean connected = rb.isConnected();

      if (connected)
      if (rb.isConnected())
      {
        // Connection verified
        TRACER.debugInfo("checkConnection: connection of broker "
          + rb.getServerId() + " to RS " + rb.getRsGroupId()
          + " obtained after " + nSec + " seconds.");
        return;
      }

      // Sleep 1 second
      try
      {
        Thread.sleep(1000);
        rb.start();
      } catch (InterruptedException ex)
      {
        fail("Error sleeping " + stackTraceToSingleLineString(ex));
      }
      nSec++;

      if (nSec > secTimeout)
      {
        // Timeout reached, end with error
        fail("checkConnection: DS " + rb.getServerId() + " is not connected to "
      assertTrue(nSec <= secTimeout,
          "checkConnection: DS " + rb.getServerId() + " is not connected to "
           + "the RS port " + rsPort + " after " + secTimeout + " seconds.");
      }
    }
  }

  /**
   * Open a replicationServer session to the local ReplicationServer
@@ -315,57 +305,36 @@
    ReplicationBroker broker = new ReplicationBroker(null,
        state, baseDN, serverId, window_size, generationId,
        100000, getReplSessionSecurity(), (byte)1, 500);
    List<String> servers = new ArrayList<String>(1);
    servers.add("localhost:" + port);
    broker.start(servers);
    checkConnection(30, broker, port);
    if (timeout != 0)
      broker.setSoTimeout(timeout);

    connect(broker, port, timeout);
    return broker;
  }

  protected void deleteEntry(DN dn)
  {
    try
  protected void deleteEntry(DN dn) throws Exception
    {
    if (dn.getParent().getRDN().toString().equalsIgnoreCase("cn=domains"))
      deleteEntry(DN.decode("cn=external changelog,"+dn.toString()));
    }
    catch(Exception e)
    {}
      deleteEntry(DN.decode("cn=external changelog," + dn));

    DeleteOperationBasis op = new DeleteOperationBasis(connection,
        InternalClientConnection.nextOperationID(), InternalClientConnection.nextMessageID(),
        null, dn);
    op.run();
    if ((op.getResultCode() != ResultCode.SUCCESS) &&
        (op.getResultCode() != ResultCode.NO_SUCH_OBJECT))
    {
      fail("Delete entry " + dn +
          " failed: " + op.getResultCode().getResultCodeName());
    }
    assertTrue(op.getResultCode() == SUCCESS || op.getResultCode() == NO_SUCH_OBJECT,
        "Delete entry " + dn + " failed: " + op.getResultCode().getResultCodeName());
  }

  /**
   * suppress all the config entries created by the tests in this class
   */
  protected void cleanConfigEntries()
  protected void cleanConfigEntries() throws Exception
  {
    logError(Message.raw(Category.SYNC, Severity.NOTICE,
    "ReplicationTestCase/Cleaning config entries"));
    logError(Message.raw(Category.SYNC, Severity.NOTICE, "ReplicationTestCase/Cleaning config entries"));

    try
    for (DN dn : configEntriesToCleanup)
    {
      while (true)
      {
        DN dn = configEntryList.removeLast();
        deleteEntry(dn);
      }
    }
    catch (NoSuchElementException e) {
      // done
    }
    configEntriesToCleanup.clear();

    synchroServerEntry = null;
    replServerEntry = null;
  }
@@ -373,23 +342,15 @@
  /**
   * suppress all the real entries created by the tests in this class
   */
  protected void cleanRealEntries()
  protected void cleanRealEntries() throws Exception
  {
    logError(Message.raw(Category.SYNC, Severity.NOTICE,
    "ReplicationTestCase/Cleaning entries"));
    logError(Message.raw(Category.SYNC, Severity.NOTICE, "ReplicationTestCase/Cleaning entries"));

    // Delete entries
    try
    for (DN dn : entriesToCleanup)
    {
      while (true)
      {
        DN dn = entryList.removeLast();
        deleteEntry(dn);
      }
    }
    catch (NoSuchElementException e) {
      // done
    }
    entriesToCleanup.clear();
  }

  /**
@@ -406,10 +367,7 @@
    removeReplicationServerDB();

    cleanConfigEntries();
    configEntryList = new LinkedList<DN>();

    cleanRealEntries();
    entryList = new LinkedList<DN>();

    // Clear the test backend (TestCaseUtils.TEST_ROOT_DN_STRING)
    // (in case our test created some entries in it)
@@ -429,7 +387,7 @@
   * - replication changes backend object
   * This method checks for existence of anything of that type.
   */
  protected void paranoiaCheck()
  protected void paranoiaCheck() throws Exception
  {
    logError(Message.raw(Category.SYNC, Severity.NOTICE,
      "Performing paranoia check"));
@@ -440,7 +398,7 @@

    // Be sure that no replication server instance is left
    List<ReplicationServer> allRSInstances = ReplicationServer.getAllInstances();
    assertTrue(allRSInstances.size() == 0, "Some replication servers left: " + allRSInstances);
    assertEquals(allRSInstances.size(), 0, "Some replication servers left: " + allRSInstances);

    // Check for config entries for replication domain
    assertNoConfigEntriesWithFilter("(objectclass=ds-cfg-replication-domain)",
@@ -521,20 +479,14 @@
   * @param errorMsg The error message to display if a config entry is found
   */
  private void assertNoConfigEntriesWithFilter(String filter, String errorMsg)
  {
    try
      throws Exception
    {
      // Search for matching entries in config backend
      InternalSearchOperation op = connection.processSearch(
        ByteString.valueOf("cn=config"),
        SearchScope.WHOLE_SUBTREE,
        LDAPFilter.decode(filter));

      assertEquals(op.getResultCode(), ResultCode.SUCCESS,
        op.getErrorMessage().toString());
    InternalSearchOperation op = connection.processSearch("cn=config", WHOLE_SUBTREE, filter);
    assertEquals(op.getResultCode(), ResultCode.SUCCESS, op.getErrorMessage() .toString());

      // Check that no entries have been found
      LinkedList<SearchResultEntry> entries = op.getSearchEntries();
    List<SearchResultEntry> entries = op.getSearchEntries();
      assertNotNull(entries);
      StringBuilder sb = new StringBuilder();
      for (SearchResultEntry entry : entries)
@@ -543,34 +495,33 @@
        sb.append(' ');
      }
      assertEquals(entries.size(), 0, errorMsg + ":\n" + sb);
    } catch (Exception e)
    {
      fail("assertNoConfigEntriesWithFilter: could not search config backend" +
        "with filter: " + filter + ": " + e.getMessage());
    }
  }

  /**
   * Configure the replication for this test.
   */
  protected void configureReplication() throws Exception
  protected void configureReplication(String replServerEntryLdif,
      String synchroServerEntryLdif) throws Exception
  {
    if (replServerEntry != null)
    {
      // Add the replication server
      DirectoryServer.getConfigHandler().addEntry(replServerEntry, null);
      assertNotNull(DirectoryServer.getConfigEntry(replServerEntry.getDN()),
       "Unable to add the replication server");
      configEntryList.add(replServerEntry.getDN());
    replServerEntry = TestCaseUtils.entryFromLdifString(replServerEntryLdif);
    addConfigEntry(replServerEntry, "Unable to add the replication server");
    addSynchroServerEntry(synchroServerEntryLdif);
    }

    if (synchroServerEntry != null)
  protected void addSynchroServerEntry(String synchroServerEntryLdif)
      throws Exception
    {
      // We also have a replicated suffix (replication domain)
      DirectoryServer.getConfigHandler().addEntry(synchroServerEntry, null);
      assertNotNull(DirectoryServer.getConfigEntry(synchroServerEntry.getDN()),
          "Unable to add the synchronized server");
      configEntryList.add(synchroServerEntry.getDN());
    synchroServerEntry = TestCaseUtils.entryFromLdifString(synchroServerEntryLdif);
    addConfigEntry(synchroServerEntry, "Unable to add the synchronized server");
  }

  private void addConfigEntry(Entry configEntry, String errorMessage) throws Exception
  {
    if (configEntry != null)
    {
      DirectoryServer.getConfigHandler().addEntry(configEntry, null);
      assertNotNull(DirectoryServer.getConfigEntry(configEntry.getDN()), errorMessage);
      configEntriesToCleanup.add(configEntry.getDN());
    }
  }

@@ -591,18 +542,13 @@
    {
      if (count++>0)
        Thread.sleep(100);
      op = connection.processSearch(
          ByteString.valueOf("cn=replication,cn=monitor"),
                                    SearchScope.WHOLE_SUBTREE,
                                    LDAPFilter.decode(monitorFilter));
      op = connection.processSearch("cn=replication,cn=monitor", WHOLE_SUBTREE, monitorFilter);
    }
    while (op.getSearchEntries().isEmpty() && (count<100));
    if (op.getSearchEntries().isEmpty())
      throw new Exception("Could not read monitoring information");
    assertFalse(op.getSearchEntries().isEmpty(), "Could not read monitoring information");

    SearchResultEntry entry = op.getSearchEntries().getFirst();
    AttributeType attrType =
         DirectoryServer.getDefaultAttributeType(attr);
    AttributeType attrType = DirectoryServer.getDefaultAttributeType(attr);
    return entry.getAttributeValue(attrType, IntegerSyntax.DECODER).longValue();
  }

@@ -741,15 +687,10 @@
  {
    Entry taskEntry = TestCaseUtils.makeEntry(task);

    InternalClientConnection connection =
         InternalClientConnection.getRootConnection();
    InternalClientConnection connection = getRootConnection();

    // Add the task.
    AddOperation addOperation =
         connection.processAdd(taskEntry.getDN(),
                               taskEntry.getObjectClasses(),
                               taskEntry.getUserAttributes(),
                               taskEntry.getOperationalAttributes());
    AddOperation addOperation = connection.processAdd(taskEntry);
    assertEquals(addOperation.getResultCode(), ResultCode.SUCCESS,
                 "Add of the task definition was not successful");

@@ -822,23 +763,13 @@
   * result code is not SUCCESS
   */
  protected void addTask(Entry taskEntry, ResultCode expectedResult,
      Message errorMessage)
      Message errorMessage) throws Exception
  {
    try
    {
      TRACER.debugInfo("AddTask/" + taskEntry);

      // Change config of DS to launch the total update task
      InternalClientConnection connection =
        InternalClientConnection.getRootConnection();

      // Add the task.

      AddOperation addOperation =
        connection.processAdd(taskEntry.getDN(),
            taskEntry.getObjectClasses(),
            taskEntry.getUserAttributes(),
            taskEntry.getOperationalAttributes());
      AddOperation addOperation = getRootConnection().processAdd(taskEntry);

      assertEquals(addOperation.getResultCode(), expectedResult,
          "Result of ADD operation of the task is: "
@@ -849,15 +780,10 @@

      if (expectedResult != ResultCode.SUCCESS)
      {
        assertTrue(addOperation.getErrorMessage().toString().
            startsWith(errorMessage.toString()),
            "Error MsgID of the task <"
            + addOperation.getErrorMessage()
            + "> equals <"
            + errorMessage + ">");
        assertTrue(addOperation.getErrorMessage().toString().startsWith(errorMessage.toString()),
            "Error MsgID of the task <" + addOperation.getErrorMessage() + "> equals <" + errorMessage + ">");
        TRACER.debugInfo("Create config task: <"+ errorMessage.getDescriptor().getId()
                + addOperation.getErrorMessage() + ">");

      }
      else
      {
@@ -865,14 +791,10 @@
      }

      // Entry will be removed at the end of the test
      entryList.addLast(taskEntry.getDN());
      entriesToCleanup.add(taskEntry.getDN());

      TRACER.debugInfo("AddedTask/" + taskEntry.getDN());
    }
    catch(Exception e)
    {
      fail("Exception when adding task:"+ e.getMessage());
    }
  }

  protected void waitTaskState(Entry taskEntry, TaskState expectedTaskState,
@@ -913,10 +835,8 @@
    if ((taskState != TaskState.COMPLETED_SUCCESSFULLY)
        && (taskState != TaskState.RUNNING))
    {
      if (logMessages.size() == 0)
      {
        fail("No log messages were written to the task entry on a failed task");
      }
      assertTrue(logMessages.size() != 0,
          "No log messages were written to the task entry on a failed task");
    }
    if (logMessages.size() != 0)
    {
@@ -945,9 +865,8 @@
  /**
   * Add to the current DB the entries necessary to the test
   */
  protected void addTestEntriesToDB(String... ldifEntries)
  protected void addTestEntriesToDB(String... ldifEntries) throws Exception
  {
    try
    {
      // Change config of DS to launch the total update task
      InternalClientConnection connection =
@@ -979,10 +898,6 @@
        }
      }
    }
    catch(Exception e)
    {
      fail("addEntries Exception:"+ e.getMessage() + " " + stackTraceToSingleLineString(e));
    }
  }

  /**
@@ -1010,8 +925,8 @@
        Entry newEntry = DirectoryServer.getEntry(dn);
        if (newEntry != null)
        {
          List<Attribute> tmpAttrList = newEntry.getAttribute("entryuuid");
          for (AttributeValue val : tmpAttrList.get(0))
          Attribute attribute = newEntry.getAttribute("entryuuid").get(0);
          for (AttributeValue val : attribute)
          {
            found = val.getValue().toString();
            break;
@@ -1032,7 +947,7 @@
  /**
   * Utility method : removes a domain deleting the passed config entry
   */
  protected void removeDomain(Entry... domainCfgEntries)
  protected void removeDomain(Entry... domainCfgEntries) throws Exception
  {
    for (Entry entry : domainCfgEntries)
    {
@@ -1050,38 +965,8 @@
   * @param msgType Class of the message we are waiting for.
   * @return The expected message if it comes in time or fails (assertion).
   */
  protected static ReplicationMsg waitForSpecificMsg(Session session, String msgType) {
    int timeOut = 5000; // 5 seconds max to wait for the desired message
    long startTime = System.currentTimeMillis();
    long curTime = startTime;
    int nMsg = 0;
    while ((curTime - startTime) <= timeOut)
    {
      ReplicationMsg replMsg = null;
      try
      {
        replMsg = session.receive();
      } catch (Exception ex)
      {
        fail("Exception waiting for " + msgType + " message : " +
          ex.getClass().getName()  + " : " + ex.getMessage());
      }
      // Get message type
      String rcvMsgType = replMsg.getClass().getName();
      if (rcvMsgType.equals(msgType))
      {
        // Ok, got it, let's return the expected message
        return replMsg;
      }
      TRACER.debugInfo("waitForSpecificMsg received : " + replMsg);
      nMsg++;
      curTime = System.currentTimeMillis();
    }
    // Timeout
    fail("Failed to receive an expected " + msgType +
      " message after 5 seconds : also received " + nMsg +
      " other messages during wait time.");
    return null;
  protected static <T extends ReplicationMsg> T waitForSpecificMsg(Session session, Class<T> msgType) {
    return waitForSpecificMsg(session, null, msgType);
  }

  /**
@@ -1091,7 +976,15 @@
   * @param msgType Class of the message we are waiting for.
   * @return The expected message if it comes in time or fails (assertion).
   */
  protected static ReplicationMsg waitForSpecificMsg(ReplicationBroker broker, String msgType) {
  protected static <T extends ReplicationMsg> T waitForSpecificMsg(ReplicationBroker broker, Class<T> msgType) {
    return waitForSpecificMsg(null, broker, msgType);
  }

  protected static <T extends ReplicationMsg> T waitForSpecificMsg(Session session, ReplicationBroker broker, Class<T> msgType)
  {
    assertTrue(session != null || broker != null, "One of Session or ReplicationBroker parameter must not be null");
    assertTrue(session == null || broker == null, "Only one of Session or ReplicationBroker parameter must not be null");

    int timeOut = 5000; // 5 seconds max to wait for the desired message
    long startTime = System.currentTimeMillis();
    long curTime = startTime;
@@ -1101,27 +994,34 @@
      ReplicationMsg replMsg = null;
      try
      {
        replMsg = broker.receive();
      } catch (Exception ex)
        if (session != null)
      {
        fail("Exception waiting for " + msgType + " message : " +
          ex.getClass().getName()  + " : " + ex.getMessage());
          replMsg = session.receive();
      }
      // Get message type
      String rcvMsgType = replMsg.getClass().getName();
      if (rcvMsgType.equals(msgType))
        else if (broker != null)
        {
          replMsg = broker.receive();
        }
      }
      catch (Exception ex)
      {
        fail("Exception waiting for " + msgType + " message : "
            + ex.getClass().getName() + " : " + ex.getMessage());
      }

      if (replMsg.equals(msgType.getClass()))
      {
        // Ok, got it, let's return the expected message
        return replMsg;
        return (T) replMsg;
      }
      TRACER.debugInfo("waitForSpecificMsg received : " + replMsg);
      nMsg++;
      curTime = System.currentTimeMillis();
    }
    // Timeout
    fail("Failed to receive an expected " + msgType +
      " message after 5 seconds : also received " + nMsg +
      " other messages during wait time.");
    fail("Failed to receive an expected " + msgType
        + " message after 5 seconds : also received " + nMsg
        + " other messages during wait time.");
    return null;
  }
}

 opends/tests/unit-tests-testng/src/server/org/opends/server/replication/SchemaReplicationTest.java

@@ -32,6 +32,7 @@
import java.util.ArrayList;
import java.util.List;

import org.assertj.core.api.Assertions;
import org.opends.messages.Category;
import org.opends.messages.Message;
import org.opends.messages.Severity;
@@ -92,7 +93,6 @@
        + "ds-cfg-replication-port: " + replServerPort + "\n"
        + "ds-cfg-replication-db-directory: SchemaReplicationTest\n"
        + "ds-cfg-replication-server-id: 105\n";
    replServerEntry = TestCaseUtils.entryFromLdifString(replServerLdif);

    // suffix synchronized
    String testName = "schemaReplicationTest";
@@ -104,9 +104,8 @@
        + "ds-cfg-base-dn: cn=schema\n"
        + "ds-cfg-replication-server: localhost:" + replServerPort + "\n"
        + "ds-cfg-server-id: 1\n";
    synchroServerEntry = TestCaseUtils.entryFromLdifString(domainLdif);

    configureReplication();
    configureReplication(replServerLdif, domainLdif);
  }

  /**
@@ -140,30 +139,21 @@
      modOp.setInternalOperation(true);
      modOp.run();

      ResultCode code = modOp.getResultCode();
      assertTrue(code.equals(ResultCode.SUCCESS),
                 "The original operation failed: " + code.getResultCodeName());
      assertEquals(modOp.getResultCode(), ResultCode.SUCCESS,
          "The original operation failed");

      // See if the client has received the msg
      ReplicationMsg msg = broker.receive();

      assertTrue(msg instanceof ModifyMsg,
                 "The received replication message is not a MODIFY msg");
      Assertions.assertThat(msg).isInstanceOf(ModifyMsg.class);
      ModifyMsg modMsg = (ModifyMsg) msg;

      Operation receivedOp = modMsg.createOperation(connection);
      assertEquals(modMsg.getDN(), baseDN,
                 "The received message is not for cn=schema");

      assertTrue(receivedOp instanceof ModifyOperation,
                 "The received replication message is not a MODIFY msg");
      assertEquals(modMsg.getDN(), baseDN, "The received message is not for cn=schema");
      Assertions.assertThat(receivedOp).isInstanceOf(ModifyOperation.class);
      ModifyOperation receivedModifyOperation = (ModifyOperation) receivedOp;

      List<RawModification> rcvdRawMods =
        receivedModifyOperation.getRawModifications();

      this.rcvdMods = new ArrayList<Modification>();
      for (RawModification m : rcvdRawMods)
      for (RawModification m : receivedModifyOperation.getRawModifications())
      {
        this.rcvdMods.add(m.toModification());
      }
@@ -183,15 +173,12 @@
      modOp.setInternalOperation(true);
      modOp.run();

      code = modOp.getResultCode();
      assertTrue(code.equals(ResultCode.SUCCESS),
                 "The original operation failed" + code.getResultCodeName());
      assertEquals(modOp.getResultCode(), ResultCode.SUCCESS,
          "The original operation failed");

      // See if the client has received the msg
      msg = broker.receive();

      assertTrue(msg instanceof ModifyMsg,
                 "The received replication message is not a MODIFY msg");
      Assertions.assertThat(msg).isInstanceOf(ModifyMsg.class);
    }
    finally
    {
@@ -227,9 +214,7 @@
      boolean found = checkEntryHasAttribute(baseDN, "attributetypes",
        "( 2.5.44.77.33 NAME 'dummy' )",
        10000, true);

      if (found == false)
        fail("The modification has not been correctly replayed.");
      assertTrue(found, "The modification has not been correctly replayed.");
    }
    finally
    {
@@ -272,24 +257,16 @@

      // receive the message on the broker side.
      ReplicationMsg msg = broker.receive();

      assertTrue(msg instanceof ModifyMsg,
        "The received replication message is not a MODIFY msg");
      Assertions.assertThat(msg).isInstanceOf(ModifyMsg.class);
      ModifyMsg modMsg = (ModifyMsg) msg;

      Operation receivedOp = modMsg.createOperation(connection);
      assertEquals(modMsg.getDN(), baseDN,
        "The received message is not for cn=schema");

      assertTrue(receivedOp instanceof ModifyOperation,
        "The received replication message is not a MODIFY msg");
      assertEquals(modMsg.getDN(), baseDN, "The received message is not for cn=schema");
      Assertions.assertThat(receivedOp).isInstanceOf(ModifyOperation.class);
      ModifyOperation receivedModifyOperation = (ModifyOperation) receivedOp;

      List<RawModification> rcvdRawMods =
        receivedModifyOperation.getRawModifications();

      this.rcvdMods = new ArrayList<Modification>();
      for (RawModification m : rcvdRawMods)
      for (RawModification m : receivedModifyOperation.getRawModifications())
      {
        this.rcvdMods.add(m.toModification());
      }
@@ -327,11 +304,8 @@
        {
          break;
        }
        if (count++ > 50)
        {
          fail("The Schema persistentState (CSN:" + stateStr
        assertTrue(count++ <= 50, "The Schema persistentState (CSN:" + stateStr
              + ") has not been saved to " + path + " : " + fileStr);
        }
        TestCaseUtils.sleep(100);
      }
    } finally

 opends/tests/unit-tests-testng/src/server/org/opends/server/replication/StressTest.java

@@ -58,15 +58,16 @@
@SuppressWarnings("javadoc")
public class StressTest extends ReplicationTestCase
{

  private static final String REPLICATION_STRESS_TEST =
    "Replication Stress Test";

  private BrokerReader reader = null;
  private BrokerReader reader;

  /**
   * A "person" entry
   */
  protected Entry personEntry;
  private Entry personEntry;

  private int replServerPort;

@@ -143,14 +144,7 @@
        thread.join();
      }

      int rcvCount = reader.getCount();

      if (rcvCount != TOTAL_MESSAGES)
      {
        fail("some messages were lost : expected : " +TOTAL_MESSAGES +
            " received : " + rcvCount);
      }

      assertEquals(reader.getCount(), TOTAL_MESSAGES, "some messages were lost");
    }
    finally {
      DirectoryServer.deregisterMonitorProvider(monitor);
@@ -193,7 +187,6 @@
        + "ds-cfg-replication-port: " + replServerPort + "\n"
        + "ds-cfg-replication-db-directory: StressTest\n"
        + "ds-cfg-replication-server-id: 106\n";
    replServerEntry = TestCaseUtils.entryFromLdifString(replServerLdif);

    // suffix synchronized
    String testName = "stressTest";
@@ -205,7 +198,6 @@
        + "ds-cfg-base-dn: ou=People," + TEST_ROOT_DN_STRING + "\n"
        + "ds-cfg-replication-server: localhost:" + replServerPort + "\n"
        + "ds-cfg-server-id: 1\n" + "ds-cfg-receive-status: true\n";
    synchroServerEntry = TestCaseUtils.entryFromLdifString(synchroServerLdif);

    String personLdif = "dn: uid=user.1,ou=People," + TEST_ROOT_DN_STRING + "\n"
        + "objectClass: top\n" + "objectClass: person\n"
@@ -223,7 +215,7 @@
        + "userPassword: password\n" + "initials: AA\n";
    personEntry = TestCaseUtils.entryFromLdifString(personLdif);

    configureReplication();
    configureReplication(replServerLdif, synchroServerLdif);
  }

  private class BrokerWriter extends Thread

 opends/tests/unit-tests-testng/src/server/org/opends/server/replication/UpdateOperationTest.java

@@ -47,7 +47,6 @@
import org.opends.server.replication.service.ReplicationBroker;
import org.opends.server.schema.DirectoryStringSyntax;
import org.opends.server.types.*;
import org.opends.server.util.StaticUtils;
import org.opends.server.util.TimeThread;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.DataProvider;
@@ -83,7 +82,7 @@
  /**
   * A "person" entry
   */
  protected Entry personEntry;
  private Entry personEntry;
  private int replServerPort;
  private String domain1uid;
  private String domain2uid;
@@ -129,7 +128,6 @@
        + "ds-cfg-replication-port: " + replServerPort + "\n"
        + "ds-cfg-replication-db-directory: UpdateOperationTest\n"
        + "ds-cfg-replication-server-id: 107\n";
    replServerEntry = TestCaseUtils.entryFromLdifString(replServerLdif);

    // suffix synchronized
    String testName = "updateOperationTest";
@@ -142,7 +140,6 @@
        + "ds-cfg-replication-server: localhost:" + replServerPort + "\n"
        + "ds-cfg-server-id: "+ domainSid +"\n"
        + "ds-cfg-receive-status: true\n";
    synchroServerEntry = TestCaseUtils.entryFromLdifString(synchroServerLdif);

    String personLdif = "dn: uid=user.1,ou=People," + TEST_ROOT_DN_STRING + "\n"
        + "objectClass: top\n" + "objectClass: person\n"
@@ -238,7 +235,7 @@
        + "objectClass:domain\n"
        + "dc:domain3");

    configureReplication();
    configureReplication(replServerLdif, synchroServerLdif);
  }

  /**
@@ -415,10 +412,7 @@
      // Check that the modify has been replayed.
      boolean found = checkEntryHasAttribute(personWithUUIDEntry.getDN(),
          "telephonenumber", "01 02 45", 10000, true);
      if (!found)
      {
        fail("The first modification was not replayed.");
      }
      assertTrue(found, "The first modification was not replayed.");

      // Simulate loss of heartbeats.
      HeartbeatThread.setHeartbeatsDisabled(true);
@@ -436,10 +430,7 @@
      found = checkEntryHasAttribute(personWithUUIDEntry.getDN(),
          "description", "Description was changed",
          10000, true);
      if (!found)
      {
        fail("The second modification was not replayed.");
      }
      assertTrue(found, "The second modification was not replayed.");

      // Delete the entries to clean the database.
      DeleteMsg delMsg = new DeleteMsg(personWithUUIDEntry.getDN(), gen.newCSN(), user1entryUUID);
@@ -667,8 +658,7 @@
    // check that the modify has been applied as if the entry had been renamed.
    boolean found = checkEntryHasAttribute(personWithUUIDEntry.getDN(),
                           "telephonenumber", "01 02 45", 10000, true);
    if (found == false)
     fail("The modification has not been correctly replayed.");
      assertTrue(found, "The modification has not been correctly replayed.");
    assertEquals(getMonitorDelta(), 1);

    // check that there was no administrative alert generated
@@ -696,9 +686,7 @@
    // check that the modify has been applied.
    found = checkEntryHasAttribute(personWithUUIDEntry.getDN(),
                           "uid", "AnotherUid", 10000, true);

    if (found == false)
      fail("The modification has not been correctly replayed.");
      assertTrue(found, "The modification has not been correctly replayed.");
    assertEquals(getMonitorDelta(), 1);

    /*
@@ -736,8 +724,8 @@
    Thread.sleep(2000);
    found = checkEntryHasAttribute(personWithUUIDEntry.getDN(),
                           "telephonenumber", "02 01 03 05", 10000, false);
    if (found == true)
     fail("The modification has been replayed while it should not.");
      assertFalse(found,
          "The modification has been replayed while it should not.");
    assertEquals(getMonitorDelta(), 1);

    // Check that there was no administrative alert generated
@@ -1295,16 +1283,10 @@
    return new Object[][] { { false }, {true} };
  }

  private void cleanupTest() {
    try
  private void cleanupTest() throws Exception
    {
      classCleanUp();
      setUp();
    } catch (Exception e)
    {
      fail("Test cleanup failed: " + e.getClass().getName() + " : " +
        e.getMessage() + " : " + StaticUtils.stackTraceToSingleLineString(e));
    }
  }

  /**
@@ -1452,9 +1434,7 @@

      boolean found = checkEntryHasAttribute(personWithUUIDEntry.getDN(),
          "telephonenumber", "01 02 45", 10000, true);

      if (found == false)
        fail("The modification has not been correctly replayed.");
      assertTrue(found, "The modification has not been correctly replayed.");

      // Test that replication is able to add attribute that do
      // not exist in the schema.
@@ -1467,8 +1447,7 @@

      found = checkEntryHasAttribute(
          personWithUUIDEntry.getDN(), "badattribute", "value", 10000, true);
      if (found == false)
        fail("The modification has not been correctly replayed.");
      assertTrue(found, "The modification has not been correctly replayed.");

      /*
       * Test the Reception of Modify Dn Msg
@@ -1482,9 +1461,7 @@
      broker.publish(moddnMsg);

      resultEntry = getEntry(newDN, 10000, true);

      assertNotNull(resultEntry,
      "The modify DN replication message was not applied");
      assertNotNull(resultEntry, "The modify DN replication message was not applied");

      /*
       * Test the Reception of Delete Msg

 opends/tests/unit-tests-testng/src/server/org/opends/server/replication/plugin/AssuredReplicationPluginTest.java

@@ -1056,7 +1056,7 @@
    } while (!rs.isScenarioExecuted());
  }

  private void endTest(String testcase)
  private void endTest(String testcase) throws Exception
  {
    debugInfo("Ending test " + testcase);
    if (replicationServer != null)
@@ -1509,7 +1509,7 @@
   * - assured-sr-server-not-acknowledged-updates in SR mode
   * - assured-sd-server-timeout-updates in SD mode
   */
  protected Map<Integer,Integer> getErrorsByServers(DN baseDN,
  private Map<Integer, Integer> getErrorsByServers(DN baseDN,
    AssuredMode assuredMode) throws Exception
  {
    // Find monitoring entry for requested base DN

 opends/tests/unit-tests-testng/src/server/org/opends/server/replication/plugin/HistoricalTest.java

@@ -84,7 +84,6 @@
         + "ds-cfg-replication-port: " + replServerPort + "\n"
         + "ds-cfg-replication-db-directory: HistoricalTest\n"
         + "ds-cfg-replication-server-id: 102\n";
    replServerEntry = TestCaseUtils.entryFromLdifString(replServerLdif);

    // The suffix to be synchronized.
    String synchroServerStringDN = "cn=" + testName + ", cn=domains, " +
@@ -97,9 +96,8 @@
         + "ds-cfg-replication-server: localhost:" + replServerPort + "\n"
         + "ds-cfg-server-id: 1\n"
         + "ds-cfg-receive-status: true\n";
    synchroServerEntry = TestCaseUtils.entryFromLdifString(synchroServerLdif);

    configureReplication();
    configureReplication(replServerLdif, synchroServerLdif);
  }

  /**

 opends/tests/unit-tests-testng/src/server/org/opends/server/replication/server/MonitorTest.java

@@ -35,7 +35,6 @@
import org.opends.messages.Message;
import org.opends.messages.Severity;
import org.opends.server.TestCaseUtils;
import org.opends.server.core.DirectoryServer;
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.common.CSN;
@@ -204,14 +203,9 @@
        + "ds-cfg-receive-status: true\n"
        + "ds-cfg-window-size: " + WINDOW_SIZE;

      synchroServerEntry = TestCaseUtils.entryFromLdifString(synchroServerLdif);
      DirectoryServer.getConfigHandler().addEntry(synchroServerEntry, null);
      assertNotNull(DirectoryServer.getConfigEntry(synchroServerEntry.getDN()),
        "Unable to add the synchronized server");
      configEntryList.add(synchroServerEntry.getDN());
      addSynchroServerEntry(synchroServerLdif);

      replDomain = LDAPReplicationDomain.retrievesReplicationDomain(baseDN);

      if (replDomain != null)
      {
        debugInfo("ReplicationDomain: Import/Export is running ? " +
@@ -233,7 +227,7 @@
      DN synchroServerDN = DN.decode(synchroServerStringDN);
      deleteEntry(synchroServerDN);
      synchroServerEntry = null;
      configEntryList.remove(configEntryList.indexOf(synchroServerDN));
    configEntriesToCleanup.remove(synchroServerDN);
  }

  private int getChangelogPort(int changelogID) throws Exception

 opends/tests/unit-tests-testng/src/server/org/opends/server/replication/server/ReplicationServerTest.java

@@ -123,7 +123,7 @@
  /**
   * Start the server and configure a replicationServer.
   */
  protected void configure() throws Exception
  private void configure() throws Exception
  {
    replicationServerPort = TestCaseUtils.findFreePort();

@@ -953,7 +953,7 @@
      session.publish(new WindowProbeMsg());

      // We may receive some MonitoringMsg so use filter method
      windowMsg = (WindowMsg)waitForSpecificMsg(session, WindowMsg.class.getName());
      windowMsg = waitForSpecificMsg(session, WindowMsg.class);
      assertEquals(serverwindow, windowMsg.getNumAck());
      debugInfo("Ending windowProbeTest");
    }
@@ -984,7 +984,7 @@
  /**
   * After the tests stop the replicationServer.
   */
  protected void shutdown() throws Exception
  private void shutdown() throws Exception
  {
    TestCaseUtils.dsconfig(
        "delete-replication-server",