Fix unit test regression introduced in previous fix for OPENDJ-866 (Local R...

Matthew Swift

04.06.2013 d74313cd4989848ab8cb3106a22cc1c8378cde58

Fix unit test regression introduced in previous fix for OPENDJ-866 (Local RS is named differently).

This commit addresses the InternalSearchMonitorTestCase unit test regression introduced by my previous fix for OPENDJ-866. 

Monitor provider instance names are required to be stable so that deregistration attempts succeed. This is not the case for many of our replication monitors which include mutable components such as host addresses or connected RS identifiers (as a parent RDN). Before my fix for OPENDJ-866 these "mutating" monitors took care to re-register themselves whenever they detected that the name was about to change. Unfortunately my fix for OPENDJ-866 made the names of local replica monitors mutable as well by including the local address. The side effect was that the monitor providers would sometimes not be deregistered which was causing the org.opends.server.monitors.InternalSearchMonitorTestCase unit test to fail. 

This fix modifies the replication domain monitor provider so that it re-registers itself whenever it senses a potential name change (i.e. connection establishment). The change was not as simple as I hoped because many unit tests create replication brokers with null replication domains. 

I've refactored the code a little and pushed ownership of the monitor provider into the broker in order to make its lifecycle clearer.

 opends/src/server/org/opends/server/replication/plugin/LDAPReplicationDomain.java

@@ -601,9 +601,6 @@
    pendingChanges =
      new PendingChanges(generator, this);

    startPublishService(replicationServers, window, heartbeatInterval,
        configuration.getChangetimeHeartbeatInterval());

    remotePendingChanges = new RemotePendingChanges(getServerState());

    // listen for changes on the configuration
@@ -611,6 +608,9 @@

    // register as an AlertGenerator
    DirectoryServer.registerAlertGenerator(this);

    startPublishService(replicationServers, window, heartbeatInterval,
        configuration.getChangetimeHeartbeatInterval());
  }

  /**

 opends/src/server/org/opends/server/replication/service/ReplicationBroker.java

@@ -59,6 +59,7 @@

import org.opends.messages.Message;
import org.opends.messages.MessageBuilder;
import org.opends.server.core.DirectoryServer;
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.replication.common.ChangeNumber;
import org.opends.server.replication.common.DSInfo;
@@ -194,6 +195,15 @@
  private int mustRunBestServerCheckingAlgorithm = 0;

  /**
   * The monitor provider for this replication domain. The name of the monitor
   * includes the local address and must therefore be re-registered every time
   * the session is re-established or destroyed. The monitor provider can only
   * be created (i.e. non-null) if there is a replication domain, which is not
   * the case in unit tests.
   */
  private final ReplicationMonitor monitor;

  /**
   * Creates a new ReplicationServer Broker for a particular ReplicationDomain.
   *
   * @param replicationDomain The replication domain that is creating us.
@@ -233,6 +243,14 @@
    this.maxRcvWindow = window;
    this.halfRcvWindow = window / 2;
    this.changeTimeHeartbeatSendInterval = changeTimeHeartbeatInterval;

    /*
     * Only create a monitor if there is a replication domain (this is not the
     * case in some unit tests).
     */
    this.monitor = replicationDomain != null ? new ReplicationMonitor(
        replicationDomain) : null;
    registerReplicationMonitor();
  }

  /**
@@ -1107,12 +1125,7 @@
    {
      if (!connected)
      {
        ProtocolSession localSession = session;
        if (localSession != null)
        {
          localSession.close();
          session = null;
        }
        setSession(null);
      }
    }
  }
@@ -1305,7 +1318,7 @@
      // updates, store it.
      if (keepConnection)
      {
        session = localSession;
        setSession(localSession);
      }

      return replServerInfo;
@@ -1414,11 +1427,8 @@
          server, baseDn, stackTraceToSingleLineString(e));
      logError(message);

      if (session != null)
      {
        session.close();
        session = null;
      }
      setSession(null);

      // Be sure to return null.
      topologyMsg = null;
    }
@@ -1489,11 +1499,8 @@
          server, baseDn, stackTraceToSingleLineString(e));
      logError(message);

      if (session != null)
      {
        session.close();
        session = null;
      }
      setSession(null);

      // Be sure to return null.
      topologyMsg = null;
    }
@@ -2188,7 +2195,7 @@
      rsGroupId = -1;
      rsServerId = -1;
      rsServerUrl = null;
      session = null;
      setSession(null);
    }

    while (true)
@@ -2713,10 +2720,8 @@
      rsGroupId = -1;
      rsServerId = -1;
      rsServerUrl = null;
      if (session != null)
      {
        session.close();
      }
      setSession(null);
      deregisterReplicationMonitor();
    }
  }

@@ -2875,12 +2880,8 @@
   */
  public boolean isSessionEncrypted()
  {
    boolean isEncrypted = false;
    if (session != null)
    {
      return session.isEncrypted();
    }
    return isEncrypted;
    final ProtocolSession tmp = session;
    return tmp != null ? tmp.isEncrypted() : false;
  }

  /**
@@ -3131,4 +3132,54 @@
    return tmp != null ? tmp.getLocalUrl() : "";
  }

  /**
   * Returns the replication monitor associated with this broker.
   *
   * @return The replication monitor.
   */
  ReplicationMonitor getReplicationMonitor()
  {
    // Only invoked by replication domain so always non-null.
    return monitor;
  }

  private void setSession(final ProtocolSession newSession)
  {
    // De-register the monitor with the old name.
    deregisterReplicationMonitor();

    final ProtocolSession oldSession = session;
    if (oldSession != null)
    {
      oldSession.close();
    }
    session = newSession;

    // Re-register the monitor with the new name.
    registerReplicationMonitor();
  }

  private void registerReplicationMonitor()
  {
    /*
     * The monitor should not be registered if this is a unit test because the
     * replication domain is null.
     */
    if (monitor != null)
    {
      DirectoryServer.registerMonitorProvider(monitor);
    }
  }

  private void deregisterReplicationMonitor()
  {
    /*
     * The monitor should not be deregistered if this is a unit test because the
     * replication domain is null.
     */
    if (monitor != null)
    {
      DirectoryServer.deregisterMonitorProvider(monitor);
    }
  }
}

 opends/src/server/org/opends/server/replication/service/ReplicationDomain.java

@@ -49,7 +49,6 @@
import org.opends.server.api.DirectoryThread;
import org.opends.server.backends.task.Task;
import org.opends.server.config.ConfigException;
import org.opends.server.core.DirectoryServer;
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.replication.common.AssuredMode;
import org.opends.server.replication.common.ChangeNumber;
@@ -189,11 +188,6 @@
  private static Map<String, ReplicationDomain> domains =
    new HashMap<String, ReplicationDomain>();

  /**
   * The Monitor in charge of replication monitoring.
   */
  private ReplicationMonitor monitor;

  /*
   * Assured mode properties
   */
@@ -1480,10 +1474,8 @@
      }

      if (debugEnabled())
        TRACER.debugInfo(
           "[IE] In " + this.monitor.getMonitorInstanceName()
           + " export ends with "
           + " connected=" + broker.isConnected()
        TRACER.debugInfo("[IE] In " + getReplicationMonitorInstanceName()
            + " export ends with " + " connected=" + broker.isConnected()
           + " exportRootException=" + exportRootException);

      if (exportRootException != null)
@@ -1591,6 +1583,11 @@

  }

  private String getReplicationMonitorInstanceName()
  {
    return broker.getReplicationMonitor().getMonitorInstanceName();
  }

  /*
   * For all remote servers in tht start list,
   * - wait it has finished the import and present the expected generationID
@@ -1839,9 +1836,8 @@
        msg = broker.receive(false, false, true);

        if (debugEnabled())
          TRACER.debugInfo(
              "[IE] In " + this.monitor.getMonitorInstanceName() +
            ", receiveEntryBytes " + msg);
          TRACER.debugInfo("[IE] In " + getReplicationMonitorInstanceName()
              + ", receiveEntryBytes " + msg);

        if (msg == null)
        {
@@ -1892,9 +1888,9 @@
                  ieContext.msgCnt);
              broker.publish(amsg, false);
              if (debugEnabled())
                TRACER.debugInfo(
                    "[IE] In " + this.monitor.getMonitorInstanceName() +
                    ", publish InitializeRcvAckMsg" + amsg);
                TRACER.debugInfo("[IE] In "
                    + getReplicationMonitorInstanceName()
                    + ", publish InitializeRcvAckMsg" + amsg);
            }
          }
          return entryBytes;
@@ -3006,15 +3002,7 @@
            changetimeHeartbeatInterval);

        broker.start(replicationServers);

        /*
         * Create a replication monitor object responsible for publishing
         * monitoring information below cn=monitor.
         */
        monitor = new ReplicationMonitor(this);
      }

      DirectoryServer.registerMonitorProvider(monitor);
    }
  }

@@ -3101,7 +3089,6 @@
   */
  public void stopDomain()
  {
    DirectoryServer.deregisterMonitorProvider(monitor);
    disableService();
    domains.remove(serviceID);
  }