OPENDJ-1430 - Some changes are missing from the external changelog

Jean-Noel Rouvignac

14.57.2014 8339e4504244448349666eda0af74e2a080913b4

OPENDJ-1430 - Some changes are missing from the external changelog

Code cleanup in ECLServerHandler:
The eligibleCSN was used when the ECLServerHandler was responsible for inserting replicaDB changes into the changeNumber index DB.
To this end, the eligibleCSN was computed to provide a kind of "medium consistency point" ("kind of" because it was badly computed).
After OPENDJ-1174, it is no longer ECLServerHandler's responsibility to inserting replicaDB changes into the changeNumber index DB, so the eligibleCSN is now useless and this commit removes it.

In addition this commit enhances the toString() methods for a better readability.



ECLServerHandler.java:
Removed eligibleCSN field + refreshEligibleCSN().
Renamed computeNextEligibleMessageForDomain() to computeNextAvailableMessage() + simplified code.
In DomainContext, removed nextNonEligibleMsg field + removed isEligible(), debugInfo() and toString(CSN).
In dumpState(), made the debug string more readable.
In buildDomainContexts(), used DomainContext new ctor + extracted method newDomainContext() to build full DomainContext objects in one go + fixed a possible ConcurrentModificationException.
In DomainContext, made some fields final + added a ctor + made toString() more readable.

ReplicationServerDomain.java:
Removed getEligibleCSN() and isServerConnected(), now unused.

ReplicationServer.java:
Removed getEligibleCSN(), now unused.

ReplicationDomainDB.java, JEChangelogDB.java, ChangeNumberIndexer.java:
Removed getDomainLastAliveCSNs(), now unused.

ExternalChangeLogTest.java
Consequence of the changes above.

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

@@ -116,12 +116,6 @@
  private MultiDomainServerState previousCookie = new MultiDomainServerState();

  /**
   * Eligible CSN - only changes older or equal to eligibleCSN are published in
   * the ECL.
   */
  private CSN eligibleCSN;

  /**
   * The global list of contexts by domain for the search currently processed.
   */
  private Set<DomainContext> domainCtxts = Collections.emptySet();
@@ -132,16 +126,15 @@
   */
  private String dumpState()
  {
    return getClass().getCanonicalName() +
           "[" +
    return getClass().getSimpleName() +
           "[draftCompat=" + draftCompat +
           "] [persistent=" + startECLSessionMsg.getPersistent() +
           "] [startChangeNumber=" + startECLSessionMsg.getLastChangeNumber() +
           "] [isEndOfCNIndexDBReached=" + isEndOfCNIndexDBReached +
           "] [searchPhase=" + searchPhase +
           "] [startCookie=" + startCookie +
           "] [previousCookie=" + previousCookie +
           "]]";
           ", persistent=" + startECLSessionMsg.getPersistent() +
           ", startChangeNumber=" + startECLSessionMsg.getLastChangeNumber() +
           ", endOfCNIndexDBReached=" + isEndOfCNIndexDBReached +
           ", searchPhase=" + searchPhase +
           ", startCookie=" + startCookie +
           ", previousCookie=" + previousCookie +
           "]";
  }

  /**
@@ -154,27 +147,35 @@
   */
  private class DomainContext
  {
    private ReplicationServerDomain rsDomain;
    private final ReplicationServerDomain rsDomain;

    /**
     * active when there are still changes supposed eligible for the ECL.
     * Active when there are still changes supposed eligible for the ECL. It is
     * active by default.
     */
    private boolean active;
    private boolean active = true;
    private UpdateMsg nextMsg;

    /**
     * the message handler from which are reading the changes for this domain.
     */
    private MessageHandler mh;
    private UpdateMsg nextMsg;
    private UpdateMsg nextNonEligibleMsg;
    private ServerState startState;
    private ServerState currentState;
    private ServerState stopState;
    private long domainLatestTrimDate;
    private final MessageHandler mh;
    private final ServerState startState;
    private final ServerState currentState = new ServerState();
    private final ServerState stopState;
    private final long domainLatestTrimDate;

    /**
     * {@inheritDoc}
     */
    public DomainContext(ReplicationServerDomain domain,
        ServerState startState, ServerState stopState, MessageHandler mh)
    {
      this.rsDomain = domain;
      this.startState = startState;
      this.stopState = stopState;
      this.mh = mh;
      this.domainLatestTrimDate = domain.getLatestDomainTrimDate();
    }

    /** {@inheritDoc} */
    @Override
    public String toString()
    {
@@ -182,96 +183,38 @@
      toString(buffer);
      return buffer.toString();
    }
    /**
     * Provide a string representation of this object for debug purpose..
     * @param buffer Append to this buffer.
     */
    public void toString(StringBuilder buffer)

    private StringBuilder toString(StringBuilder buffer)
    {
      buffer.append("[ [active=").append(active)
          .append("] [rsDomain=").append(rsDomain)
          .append("] [nextMsg=").append(nextMsg).append("(")
          .append(nextMsg != null ? asDate(nextMsg.getCSN()).toString() : "")
          .append(")")
          .append("] [nextNonEligibleMsg=").append(nextNonEligibleMsg)
          .append("] [startState=").append(startState)
          .append("] [currentState=").append(currentState)
          .append("] [stopState=").append(stopState)
          .append("]]");
      buffer.append(getClass().getSimpleName());
      buffer.append(" [");
      buffer.append(active ? "active" : "inactive");
      buffer.append(", baseDN=\"").append(rsDomain.getBaseDN()).append("\"");
      if (nextMsg != null)
      {
        buffer.append(", csn=").append(nextMsg.getCSN().toStringUI());
      }
      buffer.append(", nextMsg=[").append(nextMsg);
      buffer.append("]")
          .append(", startState=").append(startState)
          .append(", currentState=").append(currentState)
          .append(", stopState=").append(stopState)
          .append("]");
      return buffer;
    }

    /**
     * Computes the next message eligible regarding the crossDomain eligible
     * CSN.
     *
     * @param opId The operation id.
     * Computes the next available message for this domain context.
     */
    private void computeNextEligibleMessageForDomain(String opId)
    private void computeNextAvailableMessage()
    {
      nextMsg = getNextMessage();
      if (debugEnabled())
        debugInfo(opId, "ctxt=" + this);

      assert(nextMsg == null);
      try
      {
        // Before get a new message from the domain, evaluate in priority
        // a message that has not been published to the ECL because it was
        // not eligible
        if (nextNonEligibleMsg != null)
        {
          final boolean hasBecomeEligible = isEligible(nextNonEligibleMsg);

          if (debugEnabled())
            debugInfo(opId, "stored nonEligibleMsg " + nextNonEligibleMsg
                + " has now become eligible regarding the eligibleCSN ("
                + eligibleCSN + " ): " + hasBecomeEligible);

          if (hasBecomeEligible)
          {
            nextMsg = nextNonEligibleMsg;
            nextNonEligibleMsg = null;
        TRACER.debugInfo("In ECLServerHandler, for baseDN="
            + mh.getBaseDNString() + " computeNextAvailableMessage("
            + getOperationId() + ") : newMsg=[" + nextMsg + "] " + dumpState());
          }
          // else the oldest is still not eligible - let's wait next
        }
        else
        {
          // Here comes a new message !!!
          final UpdateMsg newMsg = getNextMessage();
          if (newMsg == null)
          {
            return;
          }

          if (debugEnabled())
            debugInfo(opId, "got new message : [newMsg=" + newMsg + "] "
                + dumpState());

          final boolean isEligible = isEligible(newMsg);

          if (debugEnabled())
            debugInfo(opId, "newMsg isEligible=" + isEligible + " since "
                + "newMsg=[" + toString(newMsg.getCSN()) + "] eligibleCSN=["
                + toString(eligibleCSN) + "] " + dumpState());

          if (isEligible)
          {
            nextMsg = newMsg;
          }
          else
          {
            nextNonEligibleMsg = newMsg;
          }
        }
      }
      catch(Exception e)
      {
        TRACER.debugCaught(DebugLogLevel.ERROR, e);
      }
    }

    private boolean isEligible(UpdateMsg msg)
    {
      return msg.getCSN().getTime() <= eligibleCSN.getTime();
    }

    private UpdateMsg getNextMessage()
@@ -295,18 +238,6 @@
      }
    }

    private String toString(CSN csn)
    {
      return csn + " " + asDate(csn);
    }

    private void debugInfo(String opId, String message)
    {
      TRACER.debugInfo("In ECLServerHandler, for baseDN="
          + mh.getBaseDNString() + " getNextEligibleMessageForDomain(" + opId
          + ") " + message);
    }

    /**
     * Unregister the handler from the DomainContext ReplicationDomain.
     * @return Whether the handler has been unregistered with success.
@@ -327,11 +258,10 @@

  private String domaimCtxtsToString(String msg)
  {
    StringBuilder buffer = new StringBuilder();
    final StringBuilder buffer = new StringBuilder();
    buffer.append(msg).append("\n");
    for (DomainContext domainCtxt : domainCtxts) {
      domainCtxt.toString(buffer);
      buffer.append("\n");
      domainCtxt.toString(buffer).append("\n");
    }
    return buffer.toString();
  }
@@ -702,12 +632,14 @@
      // Initializes each and every domain with the next(first) eligible message
      // from the domain.
      for (DomainContext domainCtxt : domainCtxts) {
        domainCtxt.computeNextEligibleMessageForDomain(getOperationId());
        domainCtxt.computeNextAvailableMessage();

        if (domainCtxt.nextMsg == null)
        {
          domainCtxt.active = false;
      }
    }
    }
    catch(DirectoryException de)
    {
      throw de;
@@ -767,64 +699,48 @@
        continue;

      // Creates the new domain context
      final DomainContext newDomainCtxt = new DomainContext();
      newDomainCtxt.active = true;
      newDomainCtxt.rsDomain = domain;
      newDomainCtxt.domainLatestTrimDate = domain.getLatestDomainTrimDate();

      // Assign the start state for the domain
      final DomainContext newDomainCtxt;
      final ServerState domainStartState =
          startStatesFromProvidedCookie.remove(domain.getBaseDN());
      if (startECLSessionMsg.getPersistent() == PERSISTENT_CHANGES_ONLY)
      {
        newDomainCtxt.startState = latestState;
        startStatesFromProvidedCookie.remove(domain.getBaseDN());
        newDomainCtxt = newDomainContext(domain, null, latestState);
      }
      else
      {
        // let's take the start state for this domain from the provided cookie
        newDomainCtxt.startState =
            startStatesFromProvidedCookie.remove(domain.getBaseDN());

        ServerState startState = domainStartState;
        if (providedCookie == null || providedCookie.length() == 0
            || allowUnknownDomains)
        {
          // when there is no cookie provided in the request,
          // let's start traversing this domain from the beginning of
          // what we have in the replication changelog
          if (newDomainCtxt.startState == null)
          if (startState == null)
          {
            newDomainCtxt.startState =
            startState =
                domain.getOldestState().duplicateOnlyOlderThan(
                    newDomainCtxt.domainLatestTrimDate);
                    domain.getLatestDomainTrimDate());
          }
        }
        else
        {
          // when there is a cookie provided in the request,
          if (newDomainCtxt.startState == null)
          if (startState == null)
          {
            missingDomains.append(domain.getBaseDN()).append(":;");
            continue;
          }
          else if (!newDomainCtxt.startState.isEmpty()
              && hasCookieBeenTrimmedFromDB(domain, newDomainCtxt.startState))
          else if (!startState.isEmpty()
              && hasCookieBeenTrimmedFromDB(domain, startState))
          {
            throw new DirectoryException(ResultCode.UNWILLING_TO_PERFORM,
                ERR_RESYNC_REQUIRED_TOO_OLD_DOMAIN_IN_PROVIDED_COOKIE.get(
                    newDomainCtxt.rsDomain.getBaseDN().toNormalizedString()));
                    domain.getBaseDN().toNormalizedString()));
          }
        }

        newDomainCtxt.stopState = latestState;
        newDomainCtxt = newDomainContext(domain, startState, latestState);
      }
      newDomainCtxt.currentState = new ServerState();

      // Creates an unconnected SH for the domain
      MessageHandler mh = new MessageHandler(maxQueueSize, replicationServer);
      mh.setInitialServerState(newDomainCtxt.startState);
      mh.setBaseDNAndDomain(domain.getBaseDN(), false);
      // register the unconnected into the domain
      domain.registerHandler(mh);
      newDomainCtxt.mh = mh;

      previousCookie.replace(newDomainCtxt.rsDomain.getBaseDN(),
                             newDomainCtxt.startState.duplicate());
@@ -851,11 +767,16 @@
    */
    if (!startStatesFromProvidedCookie.isEmpty() && allowUnknownDomains)
    {
      // JNR: Will the following code trigger a ConcurrentModificationException?
      for (DN providedDomain : startStatesFromProvidedCookie.keySet())
      final Set<DN> providedDomains = startStatesFromProvidedCookie.keySet();
      for (Iterator<DN> iter = providedDomains.iterator(); iter.hasNext();)
      {
        DN providedDomain = iter.next();
        if (rs.getReplicationServerDomain(providedDomain) == null)
        {
          // the domain provided in the cookie is not replicated
          startStatesFromProvidedCookie.remove(providedDomain);
          iter.remove();
        }
      }
    }

    // Now do the final checking
@@ -880,6 +801,19 @@
    return results;
  }

  private DomainContext newDomainContext(ReplicationServerDomain domain,
      ServerState startState, ServerState stopState) throws DirectoryException
  {
    // Create an unconnected MessageHandler for the domain
    MessageHandler mh = new MessageHandler(maxQueueSize, replicationServer);
    mh.setInitialServerState(startState);
    mh.setBaseDNAndDomain(domain.getBaseDN(), false);
    // register the unconnected into the domain
    domain.registerHandler(mh);

    return new DomainContext(domain, startState, stopState, mh);
  }

  private boolean hasCookieBeenTrimmedFromDB(ReplicationServerDomain rsDomain,
      ServerState cookie)
  {
@@ -1072,15 +1006,12 @@

    if (debugEnabled())
      TRACER.debugInfo(getClass().getCanonicalName() + " " + getOperationId()
          + " initialized: " + " " + dumpState() + " " + " "
          + domaimCtxtsToString(""));
          + " initialized: " + " " + dumpState() + domaimCtxtsToString(""));
  }

  private void initializeChangelogSearch(StartECLSessionMsg msg)
      throws DirectoryException
  {
    refreshEligibleCSN();

    if (msg.getECLRequestType() == REQUEST_TYPE_FROM_COOKIE)
    {
      initializeCLSearchFromCookie(msg.getCrossDomainServerState());
@@ -1101,7 +1032,6 @@
   */
  public ECLUpdateMsg takeECLUpdate() throws DirectoryException
  {
    refreshEligibleCSN();
    ECLUpdateMsg msg = getNextECLUpdate();

    // TODO:ECL We should refactor so that a SH always have a session
@@ -1223,7 +1153,7 @@
        }
        if (oldestContext.active)
        {
          oldestContext.computeNextEligibleMessageForDomain(getOperationId());
          oldestContext.computeNextAvailableMessage();
        }
        oldestChange = change;
      }
@@ -1235,8 +1165,9 @@
              "In getNextECLUpdate (persistent): "
                  + "looking for the generalized oldest change"));

        for (DomainContext domainCtxt : domainCtxts) {
          domainCtxt.computeNextEligibleMessageForDomain(getOperationId());
        for (DomainContext domainCtxt : domainCtxts)
        {
          domainCtxt.computeNextAvailableMessage();
        }

        final DomainContext oldestContext = findDomainCtxtWithOldestChange();
@@ -1499,13 +1430,4 @@
    return this.searchPhase != INIT_PHASE;
  }

  /**
   * Refresh the eligibleCSN by requesting the replication server.
   */
  private void refreshEligibleCSN()
  {
    Set<String> excludedBaseDNs = startECLSessionMsg.getExcludedBaseDNs();
    eligibleCSN = replicationServer.getEligibleCSN(excludedBaseDNs);
  }

}

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

@@ -55,7 +55,6 @@
import org.opends.server.types.*;
import org.opends.server.util.LDIFReader;
import org.opends.server.util.ServerConstants;
import org.opends.server.util.TimeThread;
import org.opends.server.workflowelement.externalchangelog.ECLWorkflowElement;

import static org.opends.messages.ReplicationMessages.*;
@@ -1290,62 +1289,6 @@
  }

  /**
   * Returns the eligible CSN cross domains - relies on the eligible CSN from
   * each domain.
   *
   * @param excludedBaseDNs
   *          the list of baseDNs excluded from the computation of eligibleCSN
   * @return the cross domain eligible CSN.
   */
  public CSN getEligibleCSN(Set<String> excludedBaseDNs)
  {
    String debugLog = "";

    // traverse the domains and get the eligible CSN from each domain
    // store the oldest one as the cross domain eligible CSN
    CSN eligibleCSN = null;
    for (ReplicationServerDomain domain : getReplicationServerDomains())
    {
      if (contains(excludedBaseDNs, domain.getBaseDN().toNormalizedString()))
        continue;

      final CSN domainEligibleCSN = domain.getEligibleCSN();
      if (eligibleCSN == null ||
          (domainEligibleCSN != null
           && domainEligibleCSN.isOlderThan(eligibleCSN)))
      {
        eligibleCSN = domainEligibleCSN;
      }

      if (debugEnabled())
      {
        final String dates = domainEligibleCSN == null ?
            "" : new Date(domainEligibleCSN.getTime()).toString();
        debugLog += "[baseDN=" + domain.getBaseDN()
            + "] [eligibleCSN=" + domainEligibleCSN + ", " + dates + "]";
      }
    }

    if (eligibleCSN==null )
    {
      eligibleCSN = new CSN(TimeThread.getTime(), 0, 0);
    }

    if (debugEnabled()) {
      TRACER.debugInfo("In " + this + " getEligibleCSN() ends with " +
        " the following domainEligibleCSN for each domain :" + debugLog +
        " thus CrossDomainEligibleCSN=" + eligibleCSN +
        "  ts=" + new Date(eligibleCSN.getTime()));
    }
    return eligibleCSN;
  }

  private boolean contains(Set<String> col, String elem)
  {
    return col != null && col.contains(elem);
  }

  /**
   * Get (or create) a handler on the {@link ChangeNumberIndexDB} for external
   * changelog.
   *
@@ -1409,16 +1352,23 @@
    final MultiDomainServerState result = new MultiDomainServerState();
    for (ReplicationServerDomain rsDomain : getReplicationServerDomains())
    {
      if (contains(excludedBaseDNs, rsDomain.getBaseDN().toNormalizedString()))
        continue;
      if (!contains(excludedBaseDNs, rsDomain.getBaseDN().toNormalizedString()))
      {
      final ServerState latestDBServerState = rsDomain.getLatestServerState();
      if (latestDBServerState.isEmpty())
        continue;
        if (!latestDBServerState.isEmpty())
        {
      result.replace(rsDomain.getBaseDN(), latestDBServerState);
    }
      }
    }
    return result;
  }

  private boolean contains(Set<String> col, String elem)
  {
    return col != null && col.contains(elem);
  }

  /**
   * Gets the weight affected to the replication server.
   * <p>

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

@@ -2496,65 +2496,6 @@
    return domainDB.getDomainOldestCSNs(baseDN);
  }

  /**
   * Returns the eligible CSN for that domain - relies on the
   * ChangeTimeHeartbeat state.
   * <p>
   * For each DS, take the oldest CSN from the changetime heartbeat state and
   * from the changelog db last CSN. Can be null.
   *
   * @return the eligible CSN.
   */
  CSN getEligibleCSN()
  {
    CSN eligibleCSN = null;
    for (final CSN lastAliveCSN : domainDB.getDomainLastAliveCSNs(baseDN))
    {
      // Should it be considered for eligibility ?
      final int serverId = lastAliveCSN.getServerId();
      if (!isServerConnected(serverId))
      {
        if (debugEnabled())
        {
          debug("serverId=" + serverId
              + " is not considered for eligibility ... potentially down");
        }
        continue;
      }

      if (eligibleCSN == null || lastAliveCSN.isNewerThan(eligibleCSN))
      {
        eligibleCSN = lastAliveCSN;
      }
    }

    if (debugEnabled())
    {
      debug("getEligibleCSN() returns result =" + eligibleCSN);
    }
    return eligibleCSN;
  }

  private boolean isServerConnected(int serverId)
  {
    if (connectedDSs.containsKey(serverId))
    {
      return true;
    }

    // not directly connected
    for (ReplicationServerHandler rsHandler : connectedRSs.values())
    {
      if (rsHandler.isRemoteLDAPServer(serverId))
      {
        return true;
      }
    }
    return false;
  }



  private void sendTopologyMsg(String type, ServerHandler handler,
      TopologyMsg msg)
  {

 opends/src/server/org/opends/server/replication/server/changelog/api/ReplicationDomainDB.java

@@ -21,7 +21,7 @@
 * CDDL HEADER END
 *
 *
 *      Copyright 2013 ForgeRock AS
 *      Copyright 2013-2014 ForgeRock AS
 */
package org.opends.server.replication.server.changelog.api;

@@ -71,16 +71,6 @@
  ServerState getDomainNewestCSNs(DN baseDN);

  /**
   * Returns the last time each serverId was seen alive for the specified
   * replication domain.
   *
   * @param baseDN
   *          the replication domain baseDN
   * @return a non null new ServerState object holding the {serverId => CSN} Map
   */
  ServerState getDomainLastAliveCSNs(DN baseDN);

  /**
   * Retrieves the latest trim date for the specified replication domain.
   * <p>
   * FIXME will be removed when ECLServerHandler will not be responsible anymore

 opends/src/server/org/opends/server/replication/server/changelog/je/ChangeNumberIndexer.java

@@ -228,20 +228,6 @@
  }

  /**
   * Returns the last time each serverId was seen alive for the specified
   * replication domain.
   *
   * @param baseDN
   *          the replication domain baseDN
   * @return a new ServerState object holding the {serverId => CSN} Map. Can be
   *         null if domain is not replicated.
   */
  public ServerState getDomainLastAliveCSNs(DN baseDN)
  {
    return lastAliveCSNs.getServerState(baseDN);
  }

  /**
   * Signals a replica went offline.
   *
   * @param baseDN

 opends/src/server/org/opends/server/replication/server/changelog/je/JEChangelogDB.java

@@ -540,23 +540,6 @@

  /** {@inheritDoc} */
  @Override
  public ServerState getDomainLastAliveCSNs(DN baseDN)
  {
    final ChangeNumberIndexer indexer = this.cnIndexer.get();
    if (indexer != null)
    {
      final ServerState results = indexer.getDomainLastAliveCSNs(baseDN);
      if (results != null)
      {
        // return a copy to protect against concurrent modifications
        return results.duplicate();
      }
    }
    return new ServerState();
  }

  /** {@inheritDoc} */
  @Override
  public void removeDomain(DN baseDN) throws ChangelogException
  {
    // Remember the first exception because :

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

@@ -2087,17 +2087,11 @@
      Thread.sleep(500);

      ReplicationServerDomain rsd1 = replicationServer.getReplicationServerDomain(TEST_ROOT_DN);
      debugInfo(tn, rsd1.getBaseDN()
          + " LatestServerState=" + rsd1.getLatestServerState()
          + " eligibleCSN=" + rsd1.getEligibleCSN()
          + " rs eligibleCSN=" + replicationServer.getEligibleCSN(null));
            debugInfo(tn, rsd1.getBaseDN() + " LatestServerState=" + rsd1.getLatestServerState());
      // FIXME:ECL Enable this test by adding an assert on the right value

      ReplicationServerDomain rsd2 = replicationServer.getReplicationServerDomain(TEST_ROOT_DN2);
      debugInfo(tn, rsd2.getBaseDN()
          + " LatestServerState=" + rsd2.getLatestServerState()
          + " eligibleCSN=" + rsd2.getEligibleCSN()
          + " rs eligibleCSN=" + replicationServer.getEligibleCSN(null));
            debugInfo(tn, rsd2.getBaseDN() + " LatestServerState=" + rsd2.getLatestServerState());
      // FIXME:ECL Enable this test by adding an assert on the right value
    }
    finally