OPENDJ-1116 Introduce abstraction for the changelog DB

Jean-Noel Rouvignac

14.14.2013 92359e9d9537c5a5f4b8d4087e5dd89c9b3a4289

 opends/src/server/org/opends/server/replication/protocol/MonitorMsg.java

@@ -32,6 +32,7 @@
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.zip.DataFormatException;

import org.opends.server.protocols.asn1.ASN1;
@@ -60,7 +61,7 @@
  static class ServerData
  {
    private ServerState state;
    private Long approxFirstMissingDate;
    private long approxFirstMissingDate;
  }

  /**
@@ -128,7 +129,7 @@
   * @param isLDAP Specifies whether the server is a LS or a RS
   */
  public void setServerState(int serverId, ServerState state,
      Long approxFirstMissingDate, boolean isLDAP)
      long approxFirstMissingDate, boolean isLDAP)
  {
    ServerData sd = new ServerData();
    sd.state = state;
@@ -166,7 +167,7 @@
   * @param serverId The provided serverId.
   * @return The approximated state.
   */
  public Long getLDAPApproxFirstMissingDate(int serverId)
  public long getLDAPApproxFirstMissingDate(int serverId)
  {
    return data.ldapStates.get(serverId).approxFirstMissingDate;
  }
@@ -177,7 +178,7 @@
   * @param serverId The provided serverId.
   * @return The approximated state.
   */
  public Long getRSApproxFirstMissingDate(int serverId)
  public long getRSApproxFirstMissingDate(int serverId)
  {
    return data.rsStates.get(serverId).approxFirstMissingDate;
  }
@@ -261,7 +262,7 @@
      {
        ServerState newState = new ServerState();
        int serverId = 0;
        Long outime = (long)0;
        long outime = 0;
        boolean isLDAPServer = false;

        asn1Reader.readStartSequence();
@@ -402,8 +403,9 @@
  private void writeServerStates(short protocolVersion, ASN1Writer writer,
      boolean writeRSStates) throws IOException
  {
    Map<Integer, ServerData> servers = writeRSStates ? data.rsStates
        : data.ldapStates;
    final Map<Integer, ServerData> servers =
        writeRSStates ? data.rsStates : data.ldapStates;
    final int seqNum = writeRSStates ? 0 : 1;
    for (Map.Entry<Integer, ServerData> server : servers.entrySet())
    {
      writer.writeStartSequence();
@@ -414,7 +416,7 @@
         * RS (0).
         */
        ChangeNumber cn = new ChangeNumber(
            server.getValue().approxFirstMissingDate, writeRSStates ? 0 : 1,
            server.getValue().approxFirstMissingDate, seqNum,
            server.getKey());
        if (protocolVersion >= ProtocolVersion.REPLICATION_PROTOCOL_V7)
        {
@@ -465,24 +467,26 @@
  @Override
  public String toString()
  {
    String stateS = "\nRState:[";
    stateS += data.replServerDbState.toString();
    stateS += "]";
    StringBuilder stateS = new StringBuilder("\nRState:[");
    stateS.append(data.replServerDbState);
    stateS.append("]");

    stateS += "\nLDAPStates:[";
    for (Integer sid : data.ldapStates.keySet())
    stateS.append("\nLDAPStates:[");
    for (Entry<Integer, ServerData> entry : data.ldapStates.entrySet())
    {
      ServerData sd = data.ldapStates.get(sid);
      stateS += "\n[LSstate("+ sid + ")=" + sd.state + "]" +
                " afmd=" + sd.approxFirstMissingDate + "]";
      ServerData sd = entry.getValue();
      stateS.append("\n[LSstate(").append(entry.getKey()).append(")=")
            .append(sd.state).append("]").append(" afmd=")
            .append(sd.approxFirstMissingDate).append("]");
    }

    stateS += "\nRSStates:[";
    for (Integer sid : data.rsStates.keySet())
    stateS.append("\nRSStates:[");
    for (Entry<Integer, ServerData> entry : data.rsStates.entrySet())
    {
      ServerData sd = data.rsStates.get(sid);
      stateS += "\n[RSState("+ sid + ")=" + sd.state + "]" +
                " afmd=" + sd.approxFirstMissingDate + "]";
      ServerData sd = entry.getValue();
      stateS.append("\n[RSState(").append(entry.getKey()).append(")=")
            .append(sd.state).append("]").append(" afmd=")
            .append(sd.approxFirstMissingDate + "]");
    }
    return getClass().getCanonicalName() +
    "[ sender=" + this.senderID +

 opends/src/server/org/opends/server/replication/server/LightweightServerHandler.java

@@ -137,16 +137,11 @@
    this.protocolVersion = protocolVersion;

    if (debugEnabled())
      TRACER.debugInfo("In " + getLocalRSMonitorInstanceName()
      TRACER.debugInfo("In " + rsDomain.getLocalRSMonitorInstanceName()
          + " LWSH for remote server " + this.serverId + " connected to:"
          + this.replServerHandler.getMonitorInstanceName() + " ()");
  }

  private String getLocalRSMonitorInstanceName()
  {
    return rsDomain.getReplicationServer().getMonitorInstanceName();
  }

  /**
   * Creates a DSInfo structure representing this remote DS.
   * @return The DSInfo structure representing this remote DS
@@ -173,7 +168,7 @@
  public void startHandler()
  {
    if (debugEnabled())
      TRACER.debugInfo("In " + getLocalRSMonitorInstanceName()
      TRACER.debugInfo("In " + rsDomain.getLocalRSMonitorInstanceName()
          + " LWSH for remote server " + this.serverId + " connected to:"
          + this.replServerHandler.getMonitorInstanceName() + " start");
    DirectoryServer.deregisterMonitorProvider(this);
@@ -186,7 +181,7 @@
  public void stopHandler()
  {
    if (debugEnabled())
      TRACER.debugInfo("In " + getLocalRSMonitorInstanceName()
      TRACER.debugInfo("In " + rsDomain.getLocalRSMonitorInstanceName()
          + " LWSH for remote server " + this.serverId + " connected to:"
          + this.replServerHandler.getMonitorInstanceName() + " stop");
    DirectoryServer.deregisterMonitorProvider(this);

 opends/src/server/org/opends/server/replication/server/MonitorData.java

@@ -27,8 +27,6 @@
 */
package org.opends.server.replication.server;

import static org.opends.server.loggers.debug.DebugLogger.*;

import java.util.Iterator;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -38,6 +36,8 @@
import org.opends.server.replication.common.ServerState;
import org.opends.server.util.TimeThread;

import static org.opends.server.loggers.debug.DebugLogger.*;

/**
 * This class defines the Monitor Data that are consolidated across the
 * whole replication topology.
@@ -338,17 +338,16 @@
  /**
   * Set the state of the LDAP server with the provided serverId.
   * @param serverId The server ID.
   * @param newFmd The first missing date.
   * @param newFmd The new first missing date.
   */
  public void setFirstMissingDate(int serverId, Long newFmd)
  public void setFirstMissingDate(int serverId, long newFmd)
  {
    if (newFmd==null) return;
    Long currentfmd = fmd.get(serverId);
    if (currentfmd==null)
    Long currentFmd = fmd.get(serverId);
    if (currentFmd == null)
    {
      fmd.put(serverId, newFmd);
    }
    else if (newFmd != 0 && (newFmd < currentfmd || currentfmd == 0))
    else if (newFmd != 0 && (newFmd < currentFmd || currentFmd == 0))
    {
      fmd.replace(serverId, newFmd);
    }

 opends/src/server/org/opends/server/replication/server/MonitoringPublisher.java

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

import java.io.IOException;
import static org.opends.server.loggers.debug.DebugLogger.debugEnabled;
import static org.opends.server.loggers.debug.DebugLogger.getTracer;

import org.opends.server.api.DirectoryThread;
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.replication.protocol.MonitorMsg;

import static org.opends.server.loggers.debug.DebugLogger.*;

/**
 * This thread regularly publishes monitoring information:
 * - it sends monitoring messages regarding the direct topology (directly
@@ -54,13 +54,13 @@
   */
  private static final DebugTracer TRACER = getTracer();

  // The domain we send monitoring for
  /** The domain we send monitoring for. */
  private final ReplicationServerDomain replicationServerDomain;

  // Sleep time (in ms) before sending new monitoring messages.
  /** Sleep time (in ms) before sending new monitoring messages. */
  private volatile long period;

  // Is the thread terminated ?
  /** Whether the thread is terminated. */
  private volatile boolean done = false;

  private final Object shutdownLock = new Object();
@@ -75,8 +75,8 @@
    long period)
  {
    super("Replication server RS("
        + replicationServerDomain.getReplicationServer()
            .getServerId() + ") monitor publisher for domain \""
        + replicationServerDomain.getLocalRSServerId()
        + ") monitor publisher for domain \""
        + replicationServerDomain.getBaseDn() + "\"");

    this.replicationServerDomain = replicationServerDomain;
@@ -114,8 +114,7 @@
        MonitorMsg monitorMsg = replicationServerDomain
            .createGlobalTopologyMonitorMsg(0, 0, monitorData);

        int localServerId = replicationServerDomain
            .getReplicationServer().getServerId();
        int localServerId = replicationServerDomain.getLocalRSServerId();
        for (ServerHandler serverHandler : replicationServerDomain
            .getConnectedDSs().values())
        {
@@ -137,20 +136,14 @@
    {
      TRACER.debugInfo("Monitoring publisher for dn "
          + replicationServerDomain.getBaseDn()
          + " in RS "
          + replicationServerDomain.getReplicationServer()
              .getServerId()
          + " in RS " + replicationServerDomain.getLocalRSServerId()
          + " has been interrupted while sleeping.");

    }

    done = true;
    TRACER.debugInfo("Monitoring publisher for dn "
        + replicationServerDomain.getBaseDn()
        + " is terminated."
        + " This is in RS "
        + replicationServerDomain.getReplicationServer()
            .getServerId());
        + replicationServerDomain.getBaseDn() + " is terminated."
        + " This is in RS " + replicationServerDomain.getLocalRSServerId());
  }


@@ -167,9 +160,9 @@

      if (debugEnabled())
      {
        TRACER.debugInfo("Shutting down monitoring publisher for dn " +
          replicationServerDomain.getBaseDn() + " in RS " +
          replicationServerDomain.getReplicationServer().getServerId());
        TRACER.debugInfo("Shutting down monitoring publisher for dn "
            + replicationServerDomain.getBaseDn()
            + " in RS " + replicationServerDomain.getLocalRSServerId());
      }
    }
  }
@@ -184,7 +177,7 @@
    {
      int FACTOR = 40; // Wait for 2 seconds before interrupting the thread
      int n = 0;
      while ((!done) && (this.isAlive()))
      while (!done && isAlive())
      {
        Thread.sleep(50);
        n++;
@@ -192,8 +185,8 @@
        {
          TRACER.debugInfo("Interrupting monitoring publisher for dn " +
            replicationServerDomain.getBaseDn() + " in RS " +
            replicationServerDomain.getReplicationServer().getServerId());
          this.interrupt();
            replicationServerDomain.getLocalRSServerId());
          interrupt();
        }
      }
    } catch (InterruptedException e)

 opends/src/server/org/opends/server/replication/server/ReplicationServerDomain.java

@@ -91,9 +91,6 @@
   * The following map contains one balanced tree for each replica ID to which
   * we are currently publishing the first update in the balanced tree is the
   * next change that we must push to this particular server.
   * <p>
   * We add new TreeSet in the HashMap when a new server register to this
   * replication server.
   */
  private final Map<Integer, DataServerHandler> directoryServers =
    new ConcurrentHashMap<Integer, DataServerHandler>();
@@ -103,9 +100,6 @@
   * we are connected (so normally all the replication servers) the first update
   * in the balanced tree is the next change that we must push to this
   * particular server.
   * <p>
   * We add new TreeSet in the HashMap when a new replication server register to
   * this replication server.
   */
  private final Map<Integer, ReplicationServerHandler> replicationServers =
    new ConcurrentHashMap<Integer, ReplicationServerHandler>();
@@ -2223,7 +2217,7 @@
    if (debugEnabled())
    {
      TRACER.debugInfo(
          "In RS " + getReplicationServer().getServerId() +
          "In RS " + getLocalRSServerId() +
          " Receiving ChangeStatusMsg from " + senderHandler.getServerId() +
          " for baseDn " + baseDn + ":\n" + csMsg);
    }
@@ -2283,8 +2277,7 @@
  {
    try
    {
      // Acquire lock on domain (see more details in comment of start() method
      // of ServerHandler)
      // Acquire lock on domain (see ServerHandler#start() for more details)
      lock();
    }
    catch (InterruptedException ex)
@@ -2317,8 +2310,7 @@
      ServerStatus oldStatus = serverHandler.getStatus();
      try
      {
        newStatus = serverHandler
            .changeStatusFromStatusAnalyzer(event);
        newStatus = serverHandler.changeStatusFromStatusAnalyzer(event);
      }
      catch (IOException e)
      {
@@ -2429,15 +2421,6 @@
  }

  /**
   * Return the associated replication server.
   * @return The replication server.
   */
  public ReplicationServer getReplicationServer()
  {
    return localReplicationServer;
  }

  /**
   * Process topology information received from a peer RS.
   * @param topoMsg The just received topo message from remote RS
   * @param handler The handler that received the message.
@@ -2453,10 +2436,9 @@
  {
    if (debugEnabled())
    {
      TRACER.debugInfo(
        "In RS " + getReplicationServer().getServerId() +
        " Receiving TopologyMsg from " + handler.getServerId() +
        " for baseDn " + baseDn + ":\n" + topoMsg);
      TRACER.debugInfo("In RS " + getLocalRSServerId()
          + " Receiving TopologyMsg from " + handler.getServerId()
          + " for baseDn " + baseDn + ":\n" + topoMsg);
    }

    try
@@ -2475,14 +2457,10 @@

    try
    {
      /*
       * Store DS connected to remote RS & update information about the peer RS
       */
      // Store DS connected to remote RS & update information about the peer RS
      handler.processTopoInfoFromRS(topoMsg);

      /*
       * Handle generation id
       */
      // Handle generation id
      if (allowResetGenId)
      {
        // Check if generation id has to be reseted
@@ -2495,17 +2473,14 @@

      if (isDifferentGenerationId(handler.getGenerationId()))
      {
        Message message = WARN_BAD_GENERATION_ID_FROM_RS.get(handler
            .getServerId(), handler.session
            .getReadableRemoteAddress(), handler.getGenerationId(),
            baseDn, getReplicationServer().getServerId(),
            generationId);
        Message message = WARN_BAD_GENERATION_ID_FROM_RS.get(
            handler.getServerId(), handler.session.getReadableRemoteAddress(),
            handler.getGenerationId(),
            baseDn, getLocalRSServerId(), generationId);
        logError(message);

        ErrorMsg errorMsg = new ErrorMsg(
            getReplicationServer().getServerId(),
            handler.getServerId(),
            message);
        ErrorMsg errorMsg =
            new ErrorMsg(getLocalRSServerId(), handler.getServerId(), message);
        handler.send(errorMsg);
      }

@@ -3411,4 +3386,28 @@
    }
    return latest;
  }

  /**
   * Return the monitor instance name of the ReplicationServer that created the
   * current instance.
   *
   * @return the monitor instance name of the ReplicationServer that created the
   *         current instance.
   */
  String getLocalRSMonitorInstanceName()
  {
    return this.localReplicationServer.getMonitorInstanceName();
  }

  /**
   * Return the serverId of the ReplicationServer that created the current
   * instance.
   *
   * @return the serverId of the ReplicationServer that created the current
   *         instance.
   */
  int getLocalRSServerId()
  {
    return this.localReplicationServer.getServerId();
  }
}

 opends/src/server/org/opends/server/replication/server/ServerHandler.java

@@ -27,10 +27,6 @@
 */
package org.opends.server.replication.server;

import static org.opends.messages.ReplicationMessages.*;
import static org.opends.server.loggers.ErrorLogger.*;
import static org.opends.server.loggers.debug.DebugLogger.*;

import java.io.IOException;
import java.util.List;
import java.util.Random;
@@ -47,26 +43,12 @@
import org.opends.server.replication.common.ChangeNumber;
import org.opends.server.replication.common.RSInfo;
import org.opends.server.replication.common.ServerStatus;
import org.opends.server.replication.protocol.AckMsg;
import org.opends.server.replication.protocol.ChangeTimeHeartbeatMsg;
import org.opends.server.replication.protocol.HeartbeatThread;
import org.opends.server.replication.protocol.ProtocolVersion;
import org.opends.server.replication.protocol.ReplicationMsg;
import org.opends.server.replication.protocol.ResetGenerationIdMsg;
import org.opends.server.replication.protocol.RoutableMsg;
import org.opends.server.replication.protocol.Session;
import org.opends.server.replication.protocol.StartECLSessionMsg;
import org.opends.server.replication.protocol.StartMsg;
import org.opends.server.replication.protocol.StartSessionMsg;
import org.opends.server.replication.protocol.TopologyMsg;
import org.opends.server.replication.protocol.UpdateMsg;
import org.opends.server.replication.protocol.WindowMsg;
import org.opends.server.replication.protocol.WindowProbeMsg;
import org.opends.server.types.Attribute;
import org.opends.server.types.Attributes;
import org.opends.server.types.DirectoryException;
import org.opends.server.types.InitializationException;
import org.opends.server.types.ResultCode;
import org.opends.server.replication.protocol.*;
import org.opends.server.types.*;

import static org.opends.messages.ReplicationMessages.*;
import static org.opends.server.loggers.ErrorLogger.*;
import static org.opends.server.loggers.debug.DebugLogger.*;

/**
 * This class defines a server handler  :
@@ -408,15 +390,12 @@
   */
  public void send(ReplicationMsg msg) throws IOException
  {
    /*
     * Some unit tests include a null domain, so avoid logging anything in that
     * case.
     */
    // avoid logging anything for unit tests that include a null domain.
    if (debugEnabled() && replicationServerDomain != null)
    {
      TRACER.debugInfo("In "
          + replicationServerDomain.getReplicationServer()
              .getMonitorInstanceName() + this + " publishes message:\n" + msg);
          + replicationServerDomain.getLocalRSMonitorInstanceName() + " "
          + this + " publishes message:\n" + msg);
    }
    session.publish(msg);
  }
@@ -427,19 +406,17 @@
   * @return The age if the older change has not yet been replicated
   *         to the server handled by this ServerHandler.
   */
  public Long getApproxFirstMissingDate()
  public long getApproxFirstMissingDate()
  {
    Long result = (long) 0;

    // Get the older CN received
    ChangeNumber olderUpdateCN = getOlderUpdateCN();
    if (olderUpdateCN != null)
    {
      // If not present in the local RS db,
      // then approximate with the older update time
      result = olderUpdateCN.getTime();
      return olderUpdateCN.getTime();
    }
    return result;
    return 0;
  }

  /**
@@ -917,9 +894,9 @@
  public void process(RoutableMsg msg)
  {
    if (debugEnabled())
      TRACER.debugInfo("In " + replicationServerDomain.getReplicationServer().
          getMonitorInstanceName() + this +
          " processes routable msg received:" + msg);
      TRACER.debugInfo("In "
          + replicationServerDomain.getLocalRSMonitorInstanceName() + " "
          + this + " processes routable msg received:" + msg);
    replicationServerDomain.process(msg, this);
  }

@@ -931,9 +908,9 @@
  public void process(ChangeTimeHeartbeatMsg msg)
  {
    if (debugEnabled())
      TRACER.debugInfo("In " + replicationServerDomain.getReplicationServer().
          getMonitorInstanceName() + this +
          " processes received msg:\n" + msg);
      TRACER.debugInfo("In "
          + replicationServerDomain.getLocalRSMonitorInstanceName() + " "
          + this + " processes received msg:\n" + msg);
    replicationServerDomain.processChangeTimeHeartbeatMsg(this, msg);
  }


 opends/src/server/org/opends/server/replication/server/StatusAnalyzer.java

@@ -27,15 +27,14 @@
 */
package org.opends.server.replication.server;

import static org.opends.server.loggers.debug.DebugLogger.debugEnabled;
import static org.opends.server.loggers.debug.DebugLogger.getTracer;

import org.opends.server.api.DirectoryThread;
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.replication.common.ServerStatus;
import org.opends.server.replication.common.StatusMachineEvent;
import org.opends.server.types.DebugLogLevel;

import static org.opends.server.loggers.debug.DebugLogger.*;

/**
 * This thread is in charge of periodically determining if the connected
 * directory servers of the domain it is associated with are late or not
@@ -59,7 +58,7 @@
  private final ReplicationServerDomain replicationServerDomain;
  private volatile int degradedStatusThreshold = -1;

  // Sleep time for the thread, in ms.
  /** Sleep time for the thread, in ms. */
  private static final int STATUS_ANALYZER_SLEEP_TIME = 5000;

  private volatile boolean done = false;
@@ -77,9 +76,9 @@
    int degradedStatusThreshold)
  {
    super("Replication server RS("
        + replicationServerDomain.getReplicationServer()
            .getServerId() + ") delay monitor for domain \""
        + replicationServerDomain.getBaseDn() + "\"");
        + replicationServerDomain.getLocalRSServerId()
        + ") delay monitor for domain \"" + replicationServerDomain.getBaseDn()
        + "\"");

    this.replicationServerDomain = replicationServerDomain;
    this.degradedStatusThreshold = degradedStatusThreshold;
@@ -98,6 +97,7 @@
        replicationServerDomain.getBaseDn());
    }

    final int localRsId = replicationServerDomain.getLocalRSServerId();
    boolean interrupted = false;
    while (!shutdown && !interrupted)
    {
@@ -132,11 +132,10 @@
        int nChanges = serverHandler.getRcvMsgQueueSize();
        if (debugEnabled())
        {
          TRACER.debugInfo("Status analyzer for dn " +
              replicationServerDomain.getBaseDn() + " DS " +
            Integer.toString(serverHandler.getServerId()) + " has " + nChanges +
            " message(s) in writer queue. This is in RS " +
            replicationServerDomain.getReplicationServer().getServerId());
          TRACER.debugInfo("Status analyzer for dn "
              + replicationServerDomain.getBaseDn() + " DS "
              + serverHandler.getServerId() + " has " + nChanges
              + " message(s) in writer queue. This is in RS " + localRsId);
        }

        // Check status to know if it is relevant to change the status. Do not
@@ -161,10 +160,10 @@
              if (interrupted)
              {
                // Finish job and let thread die
                TRACER.debugInfo("Status analyzer for dn " +
                    replicationServerDomain.getBaseDn() +
                  " has been interrupted and will die. This is in RS " +
                  replicationServerDomain.getReplicationServer().getServerId());
                TRACER.debugInfo("Status analyzer for dn "
                    + replicationServerDomain.getBaseDn()
                    + " has been interrupted and will die. This is in RS "
                    + localRsId);
                break;
              }
            }
@@ -179,10 +178,10 @@
              if (interrupted)
              {
                // Finish job and let thread die
                TRACER.debugInfo("Status analyzer for dn " +
                    replicationServerDomain.getBaseDn() +
                  " has been interrupted and will die. This is in RS " +
                  replicationServerDomain.getReplicationServer().getServerId());
                TRACER.debugInfo("Status analyzer for dn "
                    + replicationServerDomain.getBaseDn()
                    + " has been interrupted and will die. This is in RS "
                    + localRsId);
                break;
              }
            }
@@ -192,10 +191,9 @@
    }

    done = true;
    TRACER.debugInfo("Status analyzer for dn " +
        replicationServerDomain.getBaseDn() + " is terminated." +
      " This is in RS " +
      replicationServerDomain.getReplicationServer().getServerId());
    TRACER.debugInfo("Status analyzer for dn "
        + replicationServerDomain.getBaseDn() + " is terminated."
        + " This is in RS " + localRsId);
  }

  /**
@@ -211,8 +209,8 @@
      if (debugEnabled())
      {
        TRACER.debugInfo("Shutting down status analyzer for dn "
            + replicationServerDomain.getBaseDn() + " in RS "
            + replicationServerDomain.getReplicationServer().getServerId());
            + replicationServerDomain.getBaseDn()
            + " in RS " + replicationServerDomain.getLocalRSServerId());
      }
    }
  }
@@ -234,9 +232,9 @@
        if (n >= FACTOR)
        {
          TRACER.debugInfo("Interrupting status analyzer for dn " +
            replicationServerDomain.getBaseDn() + " in RS " +
            replicationServerDomain.getReplicationServer().getServerId());
          this.interrupt();
              replicationServerDomain.getBaseDn() + " in RS " +
              replicationServerDomain.getLocalRSServerId());
          interrupt();
        }
      }
    } catch (InterruptedException e)

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

@@ -27,11 +27,6 @@
 */
package org.opends.server.replication.protocol;

import static org.opends.server.TestCaseUtils.*;
import static org.opends.server.replication.protocol.OperationContext.*;
import static org.opends.server.replication.protocol.ProtocolVersion.*;
import static org.testng.Assert.*;

import java.util.*;
import java.util.zip.DataFormatException;

@@ -51,6 +46,11 @@
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;

import static org.opends.server.TestCaseUtils.*;
import static org.opends.server.replication.protocol.OperationContext.*;
import static org.opends.server.replication.protocol.ProtocolVersion.*;
import static org.testng.Assert.*;

/**
 * Test the constructors, encoders and decoders of the replication protocol
 * PDUs classes (message classes)
@@ -84,14 +84,11 @@

    Attribute attr1 = Attributes.create("description", "new value");
    Modification mod1 = new Modification(ModificationType.REPLACE, attr1);
    List<Modification> mods1 = new ArrayList<Modification>();
    mods1.add(mod1);
    List<Modification> mods1 = newList(mod1);

    Attribute attr2 = Attributes.empty("description");
    Modification mod2 = new Modification(ModificationType.DELETE, attr2);
    List<Modification> mods2 = new ArrayList<Modification>();
    mods2.add(mod1);
    mods2.add(mod2);
    List<Modification> mods2 = newList(mod1, mod2);

    AttributeBuilder builder = new AttributeBuilder(type);
    builder.add("string");
@@ -99,8 +96,7 @@
    builder.add("again");
    Attribute attr3 = builder.toAttribute();
    Modification mod3 = new Modification(ModificationType.ADD, attr3);
    List<Modification> mods3 = new ArrayList<Modification>();
    mods3.add(mod3);
    List<Modification> mods3 = newList(mod3);

    List<Modification> mods4 = new ArrayList<Modification>();
    for (int i = 0; i < 10; i++)
@@ -113,16 +109,9 @@

    Attribute attr5 = Attributes.create("namingcontexts", TEST_ROOT_DN_STRING);
    Modification mod5 = new Modification(ModificationType.REPLACE, attr5);
    List<Modification> mods5 = new ArrayList<Modification>();
    mods5.add(mod5);
    List<Modification> mods5 = newList(mod5);

    // Entry attributes
    Attribute eattr1 = Attributes.create("description", "eav description");
    Attribute eattr2 = Attributes.create("namingcontexts", "eav naming contexts");
    List<Attribute> eclIncludes = new ArrayList<Attribute>();
    eclIncludes.add(eattr1);
    eclIncludes.add(eattr2);

    List<Attribute> eclIncludes = getEntryAttributes();
    return new Object[][] {
        { cn1, "dc=test", mods1, false, AssuredMode.SAFE_DATA_MODE, (byte)0, null},
        { cn2, "dc=cn2", mods1, true, AssuredMode.SAFE_READ_MODE, (byte)1, eclIncludes},
@@ -178,21 +167,7 @@
    assertEquals(msg.getChangeNumber(), generatedMsg.getChangeNumber());

    // Get ECL entry attributes
    ArrayList<RawAttribute> genAttrList = generatedMsg.getEclIncludes();
    if (entryAttrList==null)
      assertTrue(genAttrList.size()==0);
    else
    {
      assertTrue(genAttrList.size()==entryAttrList.size());
      int i=0;
      for (Attribute attr : entryAttrList)
      {
        assertTrue(attr.getName().equalsIgnoreCase(genAttrList.get(i).toAttribute().getName()));
        assertTrue(attr.toString().equalsIgnoreCase(genAttrList.get(i).toAttribute().toString()),
            "Comparing: " + attr.toString() + " and " + genAttrList.get(i).toAttribute().toString());
        i++;
      }
    }
    assertAttributesEqual(entryAttrList, generatedMsg.getEclIncludes());

    Operation op = msg.createOperation(connection);
    Operation generatedOperation = generatedMsg.createOperation(connection);
@@ -200,8 +175,8 @@
    assertEquals(op.getClass(), ModifyOperationBasis.class);
    assertEquals(generatedOperation.getClass(), ModifyOperationBasis.class);

    ModifyOperationBasis mod1 = (ModifyOperationBasis) op;
    ModifyOperationBasis mod2 = (ModifyOperationBasis) generatedOperation;
    ModifyOperation mod1 = (ModifyOperation) op;
    ModifyOperation mod2 = (ModifyOperation) generatedOperation;

    assertEquals(mod1.getRawEntryDN(), mod2.getRawEntryDN());
    assertEquals( mod1.getAttachment(SYNCHROCONTEXT),
@@ -237,9 +212,9 @@
    assertEquals(msg.getAssuredMode(), assuredMode);

    // Check safe data level
    assertTrue(msg.getSafeDataLevel() == 1);
    assertEquals(msg.getSafeDataLevel(), 1);
    msg.setSafeDataLevel(safeDataLevel);
    assertTrue(msg.getSafeDataLevel() == safeDataLevel);
    assertEquals(msg.getSafeDataLevel(), safeDataLevel);

    // Check equals
    ModifyMsg generatedMsg = (ModifyMsg) ReplicationMsg.generateMsg(
@@ -282,20 +257,20 @@
  @DataProvider(name = "createDeleteData")
  public Object[][] createDeleteData()
  {

    // Entry attributes
    Attribute eattr1 = Attributes.create("description", "eav description");
    Attribute eattr2 = Attributes.create("namingcontexts", "eav naming contexts");
    List<Attribute> entryAttrList = new ArrayList<Attribute>();
    entryAttrList.add(eattr1);
    entryAttrList.add(eattr2);

    List<Attribute> entryAttrList = getEntryAttributes();
    return new Object[][] {
        {"dc=com", entryAttrList, false},
        {"dc=delete,dc=an,dc=entry,dc=with,dc=a,dc=long dn", null, true},
        };
  }

  private List<Attribute> getEntryAttributes()
  {
    return newList(
        Attributes.create("description", "eav description"),
        Attributes.create("namingcontexts", "eav naming contexts"));
  }

  /**
   * Create a Delete from the data provided above.
   * The call getBytes() to test the encoding of the Msg and
@@ -309,17 +284,17 @@
  {
    InternalClientConnection connection =
        InternalClientConnection.getRootConnection();
    DeleteOperationBasis opBasis =
    DeleteOperation deleteOp =
      new DeleteOperationBasis(connection, 1, 1,null, DN.decode(rawDN));
    if (subtree)
    {
      opBasis.addRequestControl(new SubtreeDeleteControl(false));
      deleteOp.addRequestControl(new SubtreeDeleteControl(false));
    }
    LocalBackendDeleteOperation op = new LocalBackendDeleteOperation(opBasis);
    LocalBackendDeleteOperation op = new LocalBackendDeleteOperation(deleteOp);
    ChangeNumber cn = new ChangeNumber(TimeThread.getTime(),123,  45);
    op.setAttachment(SYNCHROCONTEXT, new DeleteContext(cn, "uniqueid"));
    DeleteMsg msg = new DeleteMsg(op);
    assertTrue((msg.isSubtreeDelete()==subtree));
    assertEquals(msg.isSubtreeDelete(), subtree);
    // Set ECL entry attributes
    if (entryAttrList != null)
    {
@@ -335,21 +310,7 @@
    assertEquals(generatedMsg.isSubtreeDelete(), subtree);

    // Get ECL entry attributes
    ArrayList<RawAttribute> genAttrList = generatedMsg.getEclIncludes();
    if (entryAttrList==null)
      assertTrue(genAttrList.size()==0);
    else
    {
      assertTrue(genAttrList.size()==entryAttrList.size());
      int i=0;
      for (Attribute attr : entryAttrList)
      {
        assertTrue(attr.getName().equalsIgnoreCase(genAttrList.get(i).toAttribute().getName()));
        assertTrue(attr.toString().equalsIgnoreCase(genAttrList.get(i).toAttribute().toString()),
            "Comparing: " + attr.toString() + " and " + genAttrList.get(i).toAttribute().toString());
        i++;
      }
    }
    assertAttributesEqual(entryAttrList, generatedMsg.getEclIncludes());

    Operation generatedOperation = generatedMsg.createOperation(connection);

@@ -358,8 +319,7 @@
        (subtree?(generatedOperation.getRequestControl(SubtreeDeleteControl.DECODER)!=null):
          (generatedOperation.getRequestControl(SubtreeDeleteControl.DECODER)==null)));

    DeleteOperationBasis mod2 = (DeleteOperationBasis) generatedOperation;

    DeleteOperation mod2 = (DeleteOperationBasis) generatedOperation;
    assertEquals(op.getRawEntryDN(), mod2.getRawEntryDN());

    // Create an update message from this op
@@ -375,23 +335,19 @@

    Attribute attr1 = Attributes.create("description", "new value");
    Modification mod1 = new Modification(ModificationType.REPLACE, attr1);
    List<Modification> mods1 = new ArrayList<Modification>();
    mods1.add(mod1);
    List<Modification> mods1 = newList(mod1);

    Attribute attr2 = Attributes.empty("description");
    Modification mod2 = new Modification(ModificationType.DELETE, attr2);
    List<Modification> mods2 = new ArrayList<Modification>();
    mods2.add(mod1);
    mods2.add(mod2);
    List<Modification> mods2 = newList(mod1, mod2);

    AttributeBuilder builder = new AttributeBuilder(type);
    List<Modification> mods3 = new ArrayList<Modification>();
    builder.add("string");
    builder.add("value");
    builder.add("again");
    Attribute attr3 = builder.toAttribute();
    Modification mod3 = new Modification(ModificationType.ADD, attr3);
    mods3.add(mod3);
    List<Modification> mods3 = newList(mod3);

    List<Modification> mods4 = new ArrayList<Modification>();
    for (int i = 0; i < 10; i++)
@@ -402,19 +358,11 @@
      mods4.add(mod);
    }

    // Entry attributes
    Attribute eattr1 = Attributes.create("description", "eav description");
    Attribute eattr2 = Attributes.create("namingcontexts", "eav naming contexts");
    List<Attribute> entryAttrList = new ArrayList<Attribute>();
    entryAttrList.add(eattr1);
    entryAttrList.add(eattr2);


    List<Attribute> entryAttrList = getEntryAttributes();
    return new Object[][] {
        {"dc=test,dc=com", "dc=new", false, "dc=change", mods1, false, AssuredMode.SAFE_DATA_MODE, (byte)0, entryAttrList},
        {"dc=test,dc=com", "dc=new", true, "dc=change", mods2, true, AssuredMode.SAFE_READ_MODE, (byte)1, null},
        // testNG does not like null argument so use "" for the newSuperior
        // instead of null
        // testNG does not like null argument so use "" for the newSuperior instead of null
        {"dc=test,dc=com", "dc=new", false, "", mods3, true, AssuredMode.SAFE_READ_MODE, (byte)3, entryAttrList},
        {"dc=delete,dc=an,dc=entry,dc=with,dc=a,dc=long dn",
                   "dc=new", true, "", mods4, true, AssuredMode.SAFE_DATA_MODE, (byte)99, null},
@@ -431,7 +379,7 @@
  {
    InternalClientConnection connection =
      InternalClientConnection.getRootConnection();
    ModifyDNOperationBasis op =
    ModifyDNOperation op =
      new ModifyDNOperationBasis(connection, 1, 1, null,
                  DN.decode(rawDN), RDN.decode(newRdn), deleteOldRdn,
                  (newSuperior.length() != 0 ? DN.decode(newSuperior) : null));
@@ -464,21 +412,7 @@
    assertEquals(generatedMsg.getSafeDataLevel(), safeDataLevel);

    // Get ECL entry attributes
    ArrayList<RawAttribute> genAttrList = generatedMsg.getEclIncludes();
    if (entryAttrList==null)
      assertTrue(genAttrList.size()==0);
    else
    {
      assertTrue(genAttrList.size()==entryAttrList.size());
      int i=0;
      for (Attribute attr : entryAttrList)
      {
        assertTrue(attr.getName().equalsIgnoreCase(genAttrList.get(i).toAttribute().getName()));
        assertTrue(attr.toString().equalsIgnoreCase(genAttrList.get(i).toAttribute().toString()),
            "Comparing: " + attr.toString() + " and " + genAttrList.get(i).toAttribute().toString());
        i++;
      }
    }
    assertAttributesEqual(entryAttrList, generatedMsg.getEclIncludes());

    Operation oriOp = msg.createOperation(connection);
    Operation generatedOperation = generatedMsg.createOperation(connection);
@@ -486,8 +420,8 @@
    assertEquals(oriOp.getClass(), ModifyDNOperationBasis.class);
    assertEquals(generatedOperation.getClass(), ModifyDNOperationBasis.class);

    ModifyDNOperationBasis moddn1 = (ModifyDNOperationBasis) oriOp;
    ModifyDNOperationBasis moddn2 = (ModifyDNOperationBasis) generatedOperation;
    ModifyDNOperation moddn1 = (ModifyDNOperation) oriOp;
    ModifyDNOperation moddn2 = (ModifyDNOperation) generatedOperation;

    assertEquals(msg.getChangeNumber(), generatedMsg.getChangeNumber());
    assertEquals(moddn1.getRawEntryDN(), moddn2.getRawEntryDN());
@@ -504,13 +438,7 @@
  @DataProvider(name = "createAddData")
  public Object[][] createAddData()
  {

    // Entry attributes
    Attribute eattr1 = Attributes.create("description", "eav description");
    Attribute eattr2 = Attributes.create("namingcontexts", "eav naming contexts");
    List<Attribute> entryAttrList = new ArrayList<Attribute>();
    entryAttrList.add(eattr1);
    entryAttrList.add(eattr2);
    List<Attribute> entryAttrList = getEntryAttributes();
    return new Object[][] {
        {"dc=example,dc=com", false, AssuredMode.SAFE_DATA_MODE, (byte)0, entryAttrList},
        {"o=test", true, AssuredMode.SAFE_READ_MODE, (byte)1, null},
@@ -524,22 +452,22 @@
  {
    Attribute objectClass = Attributes.create(DirectoryServer
        .getObjectClassAttributeType(), "organization");
    HashMap<ObjectClass, String> objectClassList = new HashMap<ObjectClass, String>();
    Map<ObjectClass, String> objectClassList =
        new HashMap<ObjectClass, String>();
    objectClassList.put(DirectoryServer.getObjectClass("organization"),
        "organization");

    ArrayList<Attribute> userAttributes = new ArrayList<Attribute>(1);
    Attribute attr = Attributes.create("o", "com");
    userAttributes.add(attr);
    HashMap<AttributeType, List<Attribute>> userAttList = new HashMap<AttributeType, List<Attribute>>();
    List<Attribute> userAttributes = newList(attr);
    Map<AttributeType, List<Attribute>> userAttList =
        new HashMap<AttributeType, List<Attribute>>();
    userAttList.put(attr.getAttributeType(), userAttributes);


    ArrayList<Attribute> operationalAttributes = new ArrayList<Attribute>(1);
    attr = Attributes.create("creatorsname", "dc=creator");
    operationalAttributes.add(attr);
    HashMap<AttributeType,List<Attribute>> opList=
      new HashMap<AttributeType,List<Attribute>>();
    List<Attribute> operationalAttributes = newList(attr);
    Map<AttributeType, List<Attribute>> opList =
        new HashMap<AttributeType, List<Attribute>>();
    opList.put(attr.getAttributeType(), operationalAttributes);

    ChangeNumber cn = new ChangeNumber(TimeThread.getTime(), 123,  45);
@@ -570,22 +498,7 @@
    assertEquals(generatedMsg.getSafeDataLevel(), safeDataLevel);

    // Get ECL entry attributes
    ArrayList<RawAttribute> genAttrList = generatedMsg.getEclIncludes();
    if (entryAttrList==null)
      assertTrue(genAttrList.size()==0);
    else
    {
      assertTrue(genAttrList.size()==entryAttrList.size());
      int i=0;
      for (Attribute eattr : entryAttrList)
      {
        assertTrue(eattr.getName().equalsIgnoreCase(genAttrList.get(i).toAttribute().getName()));
        assertTrue(eattr.toString().equalsIgnoreCase(genAttrList.get(i).toAttribute().toString()),
            "Comparing: " + eattr.toString() + " and " + genAttrList.get(i).toAttribute().toString());
        i++;
      }
    }

    assertAttributesEqual(entryAttrList, generatedMsg.getEclIncludes());

    // Create an new Add Operation from the current addMsg
    InternalClientConnection connection =
@@ -596,15 +509,14 @@
    assertEquals(op.getClass(), AddOperationBasis.class);
    assertEquals(generatedOperation.getClass(), AddOperationBasis.class);

    AddOperationBasis addOpBasis = (AddOperationBasis) op;
    AddOperationBasis genAddOpBasis = (AddOperationBasis) generatedOperation;
    AddOperation addOp = (AddOperation) op;
    AddOperation genAddOp = (AddOperation) generatedOperation;

    assertEquals(addOpBasis.getRawEntryDN(), genAddOpBasis.getRawEntryDN());
    assertEquals( addOpBasis.getAttachment(SYNCHROCONTEXT),
                  genAddOpBasis.getAttachment(SYNCHROCONTEXT));
    assertEquals(addOpBasis.getObjectClasses(), genAddOpBasis.getObjectClasses());
    assertEquals(addOpBasis.getOperationalAttributes(), genAddOpBasis.getOperationalAttributes());
    assertEquals(addOpBasis.getUserAttributes(), genAddOpBasis.getUserAttributes());
    assertEquals(addOp.getRawEntryDN(), genAddOp.getRawEntryDN());
    assertEquals(addOp.getAttachment(SYNCHROCONTEXT), genAddOp.getAttachment(SYNCHROCONTEXT));
    assertEquals(addOp.getObjectClasses(), genAddOp.getObjectClasses());
    assertEquals(addOp.getOperationalAttributes(), genAddOp.getOperationalAttributes());
    assertEquals(addOp.getUserAttributes(), genAddOp.getUserAttributes());

    assertEquals(msg.getBytes(), generatedMsg.getBytes());
    assertEquals(msg.toString(), generatedMsg.toString());
@@ -614,12 +526,12 @@

    AddOperation addOpB = new AddOperationBasis(connection,
        1, 1, null, dn, objectClassList, userAttList, opList);
    LocalBackendAddOperation addOp = new LocalBackendAddOperation(addOpB);
    LocalBackendAddOperation localAddOp = new LocalBackendAddOperation(addOpB);
    OperationContext opCtx = new AddContext(cn, "thisIsaUniqueID",
        "parentUniqueId");
    addOp.setAttachment(SYNCHROCONTEXT, opCtx);
    localAddOp.setAttachment(SYNCHROCONTEXT, opCtx);

    generatedMsg = new AddMsg(addOp);
    generatedMsg = new AddMsg(localAddOp);

    generatedMsg.setAssured(isAssured);
    generatedMsg.setAssuredMode(assuredMode);
@@ -636,10 +548,31 @@


    // Create an update message from this op
    AddMsg updateMsg = (AddMsg) LDAPUpdateMsg.generateMsg(addOp);
    AddMsg updateMsg = (AddMsg) LDAPUpdateMsg.generateMsg(localAddOp);
    assertEquals(msg.getChangeNumber(), updateMsg.getChangeNumber());
  }

  private void assertAttributesEqual(List<Attribute> entryAttrList,
      List<RawAttribute> genAttrList) throws LDAPException
  {
    if (entryAttrList == null)
    {
      assertEquals(genAttrList.size(), 0);
      return;
    }

    assertEquals(genAttrList.size(), entryAttrList.size());
    int i = 0;
    for (Attribute eattr : entryAttrList)
    {
      final Attribute genAttr = genAttrList.get(i).toAttribute();
      assertTrue(eattr.getName().equalsIgnoreCase(genAttr.getName()));
      assertTrue(eattr.toString().equalsIgnoreCase(genAttr.toString()),
          "Comparing: " + eattr + " and " + genAttr);
      i++;
    }
  }

  /**
   * Build some data for the AckMsg test below.
   */
@@ -649,22 +582,10 @@
    ChangeNumber cn2 = new ChangeNumber(TimeThread.getTime(), 123, 45);
    ChangeNumber cn3 = new ChangeNumber(TimeThread.getTime(), 1234567, 45678);

    ArrayList<Integer> fservers1 = new ArrayList<Integer>();
    fservers1.add(12345);
    fservers1.add(-12345);
    fservers1.add(31657);
    fservers1.add(-28456);
    fservers1.add(0);
    ArrayList<Integer> fservers2 = new ArrayList<Integer>();
    ArrayList<Integer> fservers3 = new ArrayList<Integer>();
    fservers3.add(0);
    ArrayList<Integer> fservers4 = new ArrayList<Integer>();
    fservers4.add(100);
    fservers4.add(2000);
    fservers4.add(30000);
    fservers4.add(-100);
    fservers4.add(-2000);
    fservers4.add(-30000);
    List<Integer> fservers1 = newList(12345, -12345, 31657, -28456, 0);
    List<Integer> fservers2 = newList();
    List<Integer> fservers3 = newList(0);
    List<Integer> fservers4 = newList(100, 2000, 30000, -100, -2000, -30000);

    return new Object[][] {
        {cn1, true, false, false, fservers1},
@@ -695,22 +616,22 @@
    assertFalse(msg1.hasTimeout());
    assertFalse(msg1.hasWrongStatus());
    assertFalse(msg1.hasReplayError());
    assertTrue(msg1.getFailedServers().size() == 0);
    assertEquals(msg1.getFailedServers().size(), 0);

    // Check constructor with error info
    msg1 = new  AckMsg(cn, hasTimeout, hasWrongStatus, hasReplayError, failedServers);
    assertEquals(msg1.getChangeNumber().compareTo(cn), 0);
    assertTrue(msg1.hasTimeout() == hasTimeout);
    assertTrue(msg1.hasWrongStatus() == hasWrongStatus);
    assertTrue(msg1.hasReplayError() == hasReplayError);
    assertEquals(msg1.hasTimeout(), hasTimeout);
    assertEquals(msg1.hasWrongStatus(), hasWrongStatus);
    assertEquals(msg1.hasReplayError(), hasReplayError);
    assertEquals(msg1.getFailedServers(), failedServers);

    // Constructor test (with byte[])
    msg2 = new  AckMsg(msg1.getBytes(getCurrentVersion()));
    assertEquals(msg2.getChangeNumber().compareTo(cn), 0);
    assertTrue(msg1.hasTimeout() == msg2.hasTimeout());
    assertTrue(msg1.hasWrongStatus() == msg2.hasWrongStatus());
    assertTrue(msg1.hasReplayError() == msg2.hasReplayError());
    assertEquals(msg1.hasTimeout(), msg2.hasTimeout());
    assertEquals(msg1.hasWrongStatus(), msg2.hasWrongStatus());
    assertEquals(msg1.hasReplayError(), msg2.hasReplayError());
    assertEquals(msg1.getFailedServers(), msg2.getFailedServers());

    // Check invalid bytes for constructor
@@ -739,9 +660,9 @@
    // create a msg to put in the eclupdatemsg
    InternalClientConnection connection =
      InternalClientConnection.getRootConnection();
    DeleteOperationBasis opBasis =
    DeleteOperation deleteOp =
      new DeleteOperationBasis(connection, 1, 1,null, DN.decode("cn=t1"));
    LocalBackendDeleteOperation op = new LocalBackendDeleteOperation(opBasis);
    LocalBackendDeleteOperation op = new LocalBackendDeleteOperation(deleteOp);
    ChangeNumber cn = new ChangeNumber(TimeThread.getTime(), 123,  45);
    op.setAttachment(SYNCHROCONTEXT, new DeleteContext(cn, "uniqueid"));
    DeleteMsg delmsg = new DeleteMsg(op);
@@ -819,7 +740,7 @@
        newMsg.getServerState().getChangeNumber(1));
    assertEquals(newMsg.getVersion(), getCurrentVersion());
    assertEquals(msg.getGenerationId(), newMsg.getGenerationId());
    assertTrue(msg.getGroupId() == newMsg.getGroupId());
    assertEquals(msg.getGroupId(), newMsg.getGroupId());
  }

  @DataProvider(name="createReplServerStartData")
@@ -862,9 +783,9 @@
    assertEquals(newMsg.getVersion(), getCurrentVersion());
    assertEquals(msg.getGenerationId(), newMsg.getGenerationId());
    assertEquals(msg.getSSLEncryption(), newMsg.getSSLEncryption());
    assertTrue(msg.getGroupId() == newMsg.getGroupId());
    assertTrue(msg.getDegradedStatusThreshold() ==
      newMsg.getDegradedStatusThreshold());
    assertEquals(msg.getGroupId(), newMsg.getGroupId());
    assertEquals(msg.getDegradedStatusThreshold(),
                 newMsg.getDegradedStatusThreshold());
  }

  @DataProvider(name="createReplServerStartDSData")
@@ -908,9 +829,9 @@
    assertEquals(newMsg.getVersion(), getCurrentVersion());
    assertEquals(msg.getGenerationId(), newMsg.getGenerationId());
    assertEquals(msg.getSSLEncryption(), newMsg.getSSLEncryption());
    assertTrue(msg.getGroupId() == newMsg.getGroupId());
    assertTrue(msg.getDegradedStatusThreshold() ==
      newMsg.getDegradedStatusThreshold());
    assertEquals(msg.getGroupId(), newMsg.getGroupId());
    assertEquals(msg.getDegradedStatusThreshold(),
                 newMsg.getDegradedStatusThreshold());
    assertEquals(msg.getWeight(), newMsg.getWeight());
    assertEquals(msg.getConnectedDSNumber(), newMsg.getConnectedDSNumber());
  }
@@ -968,13 +889,10 @@
    urls4.add("ldaps://host:port/dc=foobar2??sub?(sn=Another Entry 2)");


    Set<String> a1 = new HashSet<String>();
    Set<String> a2 = new HashSet<String>();
    a2.add("dc");
    Set<String> a3 = new HashSet<String>();
    a3.add("dc");
    a3.add("uid");
    Set<String> a4 = new HashSet<String>();
    Set<String> a1 = newSet();
    Set<String> a2 = newSet("dc");
    Set<String> a3 = newSet("dc", "uid");
    Set<String> a4 = newSet();

    DSInfo dsInfo1 = new DSInfo(13, "dsHost1:111", 26, 154631, ServerStatus.FULL_UPDATE_STATUS,
      false, AssuredMode.SAFE_DATA_MODE, (byte)12, (byte)132, urls1, a1, a1, (short)1);
@@ -990,37 +908,18 @@
    DSInfo dsInfo5 = new DSInfo(452436, "dsHost5:555", 45591, 0, ServerStatus.NORMAL_STATUS,
        false, AssuredMode.SAFE_READ_MODE, (byte)17, (byte)0, urls3, a1, a1, (short)5);

    List<DSInfo> dsList1 = new ArrayList<DSInfo>();
    dsList1.add(dsInfo1);

    List<DSInfo> dsList2 = new ArrayList<DSInfo>();

    List<DSInfo> dsList3 = new ArrayList<DSInfo>();
    dsList3.add(dsInfo2);

    List<DSInfo> dsList4 = new ArrayList<DSInfo>();
    dsList4.add(dsInfo5);
    dsList4.add(dsInfo4);
    dsList4.add(dsInfo3);
    dsList4.add(dsInfo2);
    dsList4.add(dsInfo1);
    List<DSInfo> dsList1 = newList(dsInfo1);
    List<DSInfo> dsList2 = newList();
    List<DSInfo> dsList3 = newList(dsInfo2);
    List<DSInfo> dsList4 = newList(dsInfo5, dsInfo4, dsInfo3, dsInfo2, dsInfo1);

    RSInfo rsInfo1 = new RSInfo(4527, "rsHost1:123", 45316, (byte)103, 1);

    RSInfo rsInfo2 = new RSInfo(4527, "rsHost2:456", 0, (byte)0, 1);

    RSInfo rsInfo3 = new RSInfo(0, "rsHost3:789", -21113, (byte)98, 1);

    RSInfo rsInfo4 = new RSInfo(45678, "rsHost4:1011", -21113, (byte)98, 1);

    List<RSInfo> rsList1 = new ArrayList<RSInfo>();
    rsList1.add(rsInfo1);

    List<RSInfo> rsList2 = new ArrayList<RSInfo>();
    rsList2.add(rsInfo1);
    rsList2.add(rsInfo2);
    rsList2.add(rsInfo3);
    rsList2.add(rsInfo4);
    List<RSInfo> rsList1 = newList(rsInfo1);
    List<RSInfo> rsList2 = newList(rsInfo1, rsInfo2, rsInfo3, rsInfo4);

    return new Object [][] {
      {dsList1, rsList1, a1},
@@ -1033,6 +932,16 @@
    };
  }

  private <T> Set<T> newSet(T... elems)
  {
    return new HashSet<T>(Arrays.asList(elems));
  }

  private <T> List<T> newList(T... elems)
  {
    return Arrays.asList(elems);
  }

  /**
   * Test TopologyMsg encoding and decoding.
   */
@@ -1080,12 +989,9 @@
    urls6.add("ldaps://host:port/dc=foo??sub?(sn=Fourth Entry)");
    urls6.add("ldaps://host:port/dc=foo??sub?(sn=Fifth Entry)");

    Set<String> a1 = new HashSet<String>();
    Set<String> a2 = new HashSet<String>();
    a2.add("dc");
    Set<String> a3 = new HashSet<String>();
    a3.add("dc");
    a3.add("uid");
    Set<String> a1 = newSet();
    Set<String> a2 = newSet("dc");
    Set<String> a3 = newSet("dc", "uid");

    return new Object[][]{
      {ServerStatus.NORMAL_STATUS, urls1, true, AssuredMode.SAFE_DATA_MODE, (byte)1, a1},
@@ -1112,9 +1018,9 @@
    StartSessionMsg newMsg =
      new StartSessionMsg(msg.getBytes(getCurrentVersion()),getCurrentVersion());
    assertEquals(msg.getStatus(), newMsg.getStatus());
    assertTrue(msg.isAssured() == newMsg.isAssured());
    assertEquals(msg.isAssured(), newMsg.isAssured());
    assertEquals(msg.getAssuredMode(), newMsg.getAssuredMode());
    assertTrue(msg.getSafeDataLevel() == newMsg.getSafeDataLevel());
    assertEquals(msg.getSafeDataLevel(), newMsg.getSafeDataLevel());
    assertEquals(msg.getReferralsURLs(), newMsg.getReferralsURLs());
    assertTrue(attrs.equals(newMsg.getEclIncludes()));
    assertTrue(attrs.equals(newMsg.getEclIncludesForDeletes()));
@@ -1237,12 +1143,12 @@
      if (sid == sid1)
      {
        assertEquals(s.toString(), s1.toString(), "");
        assertEquals((Long)(now+1), newMsg.getLDAPApproxFirstMissingDate(sid), "");
        assertEquals(now + 1, newMsg.getLDAPApproxFirstMissingDate(sid), "");
      }
      else if (sid == sid2)
      {
        assertEquals(s.toString(), s2.toString());
        assertEquals((Long)(now+2), newMsg.getLDAPApproxFirstMissingDate(sid), "");
        assertEquals(now + 2, newMsg.getLDAPApproxFirstMissingDate(sid), "");
      }
      else
      {
@@ -1258,7 +1164,7 @@
      if (sid == sid3)
      {
        assertEquals(s.toString(), s3.toString(), "");
        assertEquals((Long)(now+3), newMsg.getRSApproxFirstMissingDate(sid), "");
        assertEquals(now + 3, newMsg.getRSApproxFirstMissingDate(sid), "");
      }
      else
      {
@@ -1374,10 +1280,8 @@
  public void UpdateMsgTest() throws Exception
  {
    final String test = "string used for test";
    UpdateMsg msg =
      new UpdateMsg(
          new ChangeNumber(1, 2 , 39123),
          test.getBytes());
    ChangeNumber cn = new ChangeNumber(1, 2 , 39123);
    UpdateMsg msg = new UpdateMsg(cn, test.getBytes());
    UpdateMsg newMsg = new UpdateMsg(msg.getBytes());
    assertEquals(test.getBytes(), newMsg.getPayload());
  }
@@ -1406,7 +1310,7 @@
        newMsg.getServerState().getChangeNumber(1));
    assertEquals(newMsg.getVersion(), getCurrentVersion());
    assertEquals(msg.getGenerationId(), newMsg.getGenerationId());
    assertTrue(msg.getGroupId() == newMsg.getGroupId());
    assertEquals(msg.getGroupId(), newMsg.getGroupId());
  }

  /**
@@ -1418,41 +1322,32 @@
  {
    // data
    ChangeNumber changeNumber = new ChangeNumber(TimeThread.getTime(), 123,  45);
    String generalizedState = "fakegenstate";
    ServerState state = new ServerState();
    assertTrue(state.update(new ChangeNumber(75, 5,263)));
    short mode = 3;
    int firstDraftChangeNumber = 13;
    int lastDraftChangeNumber  = 14;
    String myopid = "fakeopid";

    // create original
    StartECLSessionMsg msg = new StartECLSessionMsg();
    msg.setChangeNumber(changeNumber);
    msg.setCrossDomainServerState(generalizedState);
    msg.setCrossDomainServerState("fakegenstate");
    msg.setPersistent(StartECLSessionMsg.PERSISTENT);
    msg.setFirstDraftChangeNumber(firstDraftChangeNumber);
    msg.setLastDraftChangeNumber(lastDraftChangeNumber);
    msg.setECLRequestType(mode);
    msg.setOperationId(myopid);
    msg.setFirstDraftChangeNumber(13);
    msg.setLastDraftChangeNumber(14);
    msg.setECLRequestType((short) 3);
    msg.setOperationId("fakeopid");
    String dn1 = "cn=admin data";
    String dn2 = "cn=config";
    Set<String> dns = new HashSet<String>();
    dns.add(dn1);
    dns.add(dn2);
    msg.setExcludedDNs(dns);
    msg.setExcludedDNs(newSet(dn1, dn2));

    // create copy
    StartECLSessionMsg newMsg = new StartECLSessionMsg(msg.getBytes(getCurrentVersion()));
    // test equality between the two copies
    assertEquals(msg.getChangeNumber(), newMsg.getChangeNumber());
    assertEquals(msg.isPersistent(), newMsg.isPersistent());
    assertEquals(msg.getFirstDraftChangeNumber(), newMsg
        .getFirstDraftChangeNumber());
    assertEquals(msg.getFirstDraftChangeNumber(), newMsg.getFirstDraftChangeNumber());
    assertEquals(msg.getECLRequestType(), newMsg.getECLRequestType());
    assertEquals(msg.getLastDraftChangeNumber(), newMsg.getLastDraftChangeNumber());
    assertTrue(
        msg.getCrossDomainServerState().equalsIgnoreCase(newMsg.getCrossDomainServerState()));
    assertTrue(
        msg.getOperationId().equalsIgnoreCase(newMsg.getOperationId()));
    assertTrue(msg.getCrossDomainServerState().equalsIgnoreCase(newMsg.getCrossDomainServerState()));
    assertTrue(msg.getOperationId().equalsIgnoreCase(newMsg.getOperationId()));
    Set<String> dns2 = newMsg.getExcludedBaseDNs();
    assertEquals(dns2.size(), 2);
    boolean dn1found=false,dn2found=false;
@@ -1481,21 +1376,21 @@
    long buildnew = 0;
    long t1,t2,t3,t31,t4,t5,t6 = 0;

    HashMap<ObjectClass, String> objectClassList = new HashMap<ObjectClass, String>();
    Map<ObjectClass, String> objectClassList =
        new HashMap<ObjectClass, String>();
    objectClassList.put(DirectoryServer.getObjectClass("organization"),
        "organization");

    ArrayList<Attribute> userAttributes = new ArrayList<Attribute>(1);
    Attribute attr = Attributes.create("o", "com");
    userAttributes.add(attr);
    HashMap<AttributeType, List<Attribute>> userAttList = new HashMap<AttributeType, List<Attribute>>();
    List<Attribute> userAttributes = newList(attr);
    Map<AttributeType, List<Attribute>> userAttList =
        new HashMap<AttributeType, List<Attribute>>();
    userAttList.put(attr.getAttributeType(), userAttributes);


    ArrayList<Attribute> operationalAttributes = new ArrayList<Attribute>(1);
    attr = Attributes.create("creatorsname", "dc=creator");
    operationalAttributes.add(attr);
    HashMap<AttributeType,List<Attribute>> opList=
    List<Attribute> operationalAttributes = newList(attr);
    Map<AttributeType, List<Attribute>> opList =
      new HashMap<AttributeType,List<Attribute>>();
    opList.put(attr.getAttributeType(), operationalAttributes);

@@ -1658,9 +1553,10 @@
      t1 = System.nanoTime();

      // create op
      DeleteOperationBasis opBasis =
      DeleteOperation deleteOp =
        new DeleteOperationBasis(connection, 1, 1,null, DN.decode(rawDN));
      LocalBackendDeleteOperation op = new LocalBackendDeleteOperation(opBasis);
      LocalBackendDeleteOperation op =
          new LocalBackendDeleteOperation(deleteOp);
      ChangeNumber cn = new ChangeNumber(TimeThread.getTime(), 123, 45);
      op.setAttachment(SYNCHROCONTEXT, new DeleteContext(cn, "uniqueid"));
      t2 = System.nanoTime();