OPENDJ-1206 (CR-4393) Create a new ReplicationBackend/ChangelogBackend to s...

Jean-Noel Rouvignac

01.51.2014 c11b3a5611e1a431e62c6cfa23b881a5fdbb62b9

 opendj-sdk/opends/src/guitools/org/opends/guitools/controlpanel/util/ReadOnlyConfigFileHandler.java

@@ -26,9 +26,6 @@
 */
package org.opends.guitools.controlpanel.util;

import static org.opends.messages.ConfigMessages.*;
import static org.opends.server.util.StaticUtils.*;

import java.io.File;
import java.util.Collections;
import java.util.HashMap;
@@ -63,6 +60,9 @@
import org.opends.server.util.LDIFException;
import org.opends.server.util.LDIFReader;

import static org.opends.messages.ConfigMessages.*;
import static org.opends.server.util.StaticUtils.*;

/**
 * A class used to read the configuration from a file.  This config file
 * handler does not allow to modify the configuration, only to read it.
@@ -90,6 +90,7 @@
  @Override
  public void finalizeConfigHandler()
  {
    finalizeBackend();
  }

  /** {@inheritDoc} */
@@ -299,12 +300,6 @@

  /** {@inheritDoc} */
  @Override
  public void finalizeBackend()
  {
  }

  /** {@inheritDoc} */
  @Override
  public DN[] getBaseDNs()
  {
    return baseDNs;

 opendj-sdk/opends/src/server/org/opends/server/api/Backend.java

@@ -29,12 +29,15 @@
import java.util.ArrayList;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.opends.messages.Message;
import org.opends.server.admin.Configuration;
import org.opends.server.config.ConfigException;
import org.opends.server.core.*;
import org.opends.server.core.PersistentSearch.CancellationCallback;
import org.opends.server.monitors.BackendMonitor;
import org.opends.server.types.*;

@@ -79,6 +82,10 @@
  /** The writability mode for this backend. */
  private WritabilityMode writabilityMode = WritabilityMode.ENABLED;

  /** The set of persistent searches registered with this backend. */
  private final ConcurrentLinkedQueue<PersistentSearch> persistentSearches =
      new ConcurrentLinkedQueue<PersistentSearch>();

  /**
   * Configure this backend based on the information in the provided
   * configuration.
@@ -146,16 +153,26 @@
  /**
   * Performs any necessary work to finalize this backend, including
   * closing any underlying databases or connections and deregistering
   * any suffixes that it manages with the Directory Server.  This may
   * any suffixes that it manages with the Directory Server. This may
   * be called during the Directory Server shutdown process or if a
   * backend is disabled with the server online.  It must not return
   * until the backend is closed.
   * <BR><BR>
   * This method may not throw any exceptions.  If any problems are
   * encountered, then they may be logged but the closure should
   * progress as completely as possible.
   * backend is disabled with the server online.
   * It must not return until the backend is closed.
   * <p>
   * This method may not throw any exceptions. If any problems are encountered,
   * then they may be logged but the closure should progress as completely as
   * possible.
   * <p>
   * This method must be called by all overriding methods with
   * <code>super.finalizeBackend()</code>.
   */
  public abstract void finalizeBackend();
  public void finalizeBackend()
  {
    for (PersistentSearch psearch : persistentSearches)
    {
      psearch.cancel();
    }
    persistentSearches.clear();
  }



@@ -867,7 +884,39 @@
    return backendMonitor;
  }

  /**
   * Registers the provided persistent search operation with this backend so
   * that it will be notified of any add, delete, modify, or modify DN
   * operations that are performed.
   *
   * @param persistentSearch
   *          The persistent search operation to register with this backend
   */
  public void registerPersistentSearch(PersistentSearch persistentSearch)
  {
    persistentSearches.add(persistentSearch);

    persistentSearch.registerCancellationCallback(new CancellationCallback()
    {
      @Override
      public void persistentSearchCancelled(PersistentSearch psearch)
      {
        persistentSearches.remove(psearch);
      }
    });
  }

  /**
   * Returns the persistent searches currently active against this local
   * backend.
   *
   * @return the list of persistent searches currently active against this local
   *         backend
   */
  public Queue<PersistentSearch> getPersistentSearches()
  {
    return persistentSearches;
  }

  /**
   * Sets the backend monitor for this backend.

 opendj-sdk/opends/src/server/org/opends/server/backends/BackupBackend.java

@@ -41,9 +41,9 @@
import org.opends.server.core.ModifyDNOperation;
import org.opends.server.core.SearchOperation;
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.types.*;
import org.opends.server.schema.BooleanSyntax;
import org.opends.server.schema.GeneralizedTimeSyntax;
import org.opends.server.types.*;

import static org.opends.messages.BackendMessages.*;
import static org.opends.server.config.ConfigConstants.*;
@@ -207,6 +207,7 @@
  @Override
  public void finalizeBackend()
  {
    super.finalizeBackend();
    currentConfig.removeBackupChangeListener(this);

    try

 opendj-sdk/opends/src/server/org/opends/server/backends/ChangelogBackend.java

@@ -29,8 +29,7 @@
import static org.opends.messages.ReplicationMessages.*;
import static org.opends.server.config.ConfigConstants.*;
import static org.opends.server.loggers.ErrorLogger.*;
import static org.opends.server.loggers.debug.DebugLogger.debugEnabled;
import static org.opends.server.loggers.debug.DebugLogger.getTracer;
import static org.opends.server.loggers.debug.DebugLogger.*;
import static org.opends.server.replication.protocol.StartECLSessionMsg.ECLRequestType.*;
import static org.opends.server.replication.server.changelog.api.DBCursor.PositionStrategy.*;
import static org.opends.server.util.LDIFWriter.*;
@@ -39,6 +38,7 @@

import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;

import org.opends.messages.Category;
import org.opends.messages.Message;
@@ -53,6 +53,7 @@
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.MultiDomainServerState;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.plugin.MultimasterReplication;
import org.opends.server.replication.protocol.AddMsg;
import org.opends.server.replication.protocol.DeleteMsg;
@@ -62,6 +63,7 @@
import org.opends.server.replication.protocol.StartECLSessionMsg.ECLRequestType;
import org.opends.server.replication.protocol.UpdateMsg;
import org.opends.server.replication.server.ReplicationServer;
import org.opends.server.replication.server.ReplicationServerDomain;
import org.opends.server.replication.server.changelog.api.ChangeNumberIndexDB;
import org.opends.server.replication.server.changelog.api.ChangeNumberIndexRecord;
import org.opends.server.replication.server.changelog.api.ChangelogDB;
@@ -72,10 +74,11 @@
import org.opends.server.replication.server.changelog.je.ECLMultiDomainDBCursor;
import org.opends.server.replication.server.changelog.je.MultiDomainDBCursor;
import org.opends.server.types.*;
import org.opends.server.util.ServerConstants;
import org.opends.server.util.StaticUtils;

/**
 * A backend that provides access to the changelog, ie the "cn=changelog"
 * A backend that provides access to the changelog, i.e. the "cn=changelog"
 * suffix. It is a read-only backend that is created by a
 * {@code ReplicationServer} and is not configurable.
 * <p>
@@ -85,8 +88,8 @@
 * request. The cookie provided in the control is used to retrieve entries from
 * the ReplicaDBs. The <code>changeNumber</code> attribute is not returned with
 * the entries.</li>
 * <li>Draft compat mode: when no "ECL Cookie Exchange Control" is provided with
 * the request. The entries are retrieved using the ChangeNumberIndexDB (or
 * <li>Draft compatibility mode: when no "ECL Cookie Exchange Control" is provided
 * with the request. The entries are retrieved using the ChangeNumberIndexDB (or
 * DraftDB, hence the name) and their attributes are set with the information
 * from the ReplicasDBs. The <code>changeNumber</code> attribute value is set
 * from the content of ChangeNumberIndexDB.</li>
@@ -134,8 +137,20 @@
  private static final AttributeType MODIFIERS_NAME_TYPE =
      DirectoryConfig.getAttributeType(OP_ATTR_MODIFIERS_NAME_LC, true);

  /** The DN for the base changelog entry. */
  private DN baseChangelogDN;
  /** The base DN for the external change log. */
  public static final DN CHANGELOG_BASE_DN;

  static
  {
    try
    {
      CHANGELOG_BASE_DN = DN.decode(DN_EXTERNAL_CHANGELOG_ROOT);
    }
    catch (DirectoryException e)
    {
      throw new RuntimeException(e);
    }
  }

  /** The set of base DNs for this backend. */
  private DN[] baseDNs;
@@ -149,7 +164,7 @@
  private final ECLEnabledDomainPredicate domainPredicate;

  /**
   * Creates a new backend with the provided repication server.
   * Creates a new backend with the provided replication server.
   *
   * @param replicationServer
   *          The replication server on which the changes are read.
@@ -165,6 +180,23 @@
    setPrivateBackend(true);
  }

  private ChangelogDB getChangelogDB()
  {
    return replicationServer.getChangelogDB();
  }

  /**
   * Returns the ChangelogBackend configured for "cn=changelog" in this directory server.
   *
   * @return the ChangelogBackend configured for "cn=changelog" in this directory server
   * @deprecated instead inject the required object where needed
   */
  @Deprecated
  public static ChangelogBackend getInstance()
  {
    return (ChangelogBackend) DirectoryServer.getBackend(CHANGELOG_BASE_DN);
  }

  /** {@inheritDoc} */
  @Override
  public void configureBackend(final Configuration config) throws ConfigException
@@ -176,29 +208,16 @@
  @Override
  public void initializeBackend() throws InitializationException
  {
    try
    {
      baseChangelogDN = DN.decode(DN_EXTERNAL_CHANGELOG_ROOT);
      baseDNs = new DN[] { baseChangelogDN };
    }
    catch (final DirectoryException e)
    {
      if (debugEnabled())
      {
        TRACER.debugCaught(DebugLogLevel.ERROR, e);
      }
      throw new InitializationException(
          ERR_BACKEND_CANNOT_DECODE_BACKEND_ROOT_DN.get(getBackendID(), getExceptionMessage(e)), e);
    }
    baseDNs = new DN[] { CHANGELOG_BASE_DN };

    try
    {
      DirectoryServer.registerBaseDN(baseChangelogDN, this, true);
      DirectoryServer.registerBaseDN(CHANGELOG_BASE_DN, this, true);
    }
    catch (final DirectoryException e)
    {
      throw new InitializationException(
          ERR_BACKEND_CANNOT_REGISTER_BASEDN.get(baseChangelogDN.toString(), getExceptionMessage(e)), e);
          ERR_BACKEND_CANNOT_REGISTER_BASEDN.get(CHANGELOG_BASE_DN.toString(), getExceptionMessage(e)), e);
    }
  }

@@ -206,9 +225,11 @@
  @Override
  public void finalizeBackend()
  {
    super.finalizeBackend();

    try
    {
      DirectoryServer.deregisterBaseDN(baseChangelogDN);
      DirectoryServer.deregisterBaseDN(CHANGELOG_BASE_DN);
    }
    catch (final DirectoryException e)
    {
@@ -299,7 +320,7 @@
    @Override
    public DN getBaseDN()
    {
      return baseChangelogDN;
      return CHANGELOG_BASE_DN;
    }

    @Override
@@ -313,6 +334,13 @@
    {
      return SearchScope.WHOLE_SUBTREE;
    }

    /** {@inheritDoc} */
    @Override
    public Object setAttachment(String name, Object value)
    {
      return null;
    }
  }

  /** {@inheritDoc} */
@@ -320,7 +348,7 @@
  public long numSubordinates(final DN entryDN, final boolean subtree) throws DirectoryException
  {
    // Compute the num subordinates only for the base DN
    if (entryDN == null || !baseChangelogDN.equals(entryDN))
    if (entryDN == null || !CHANGELOG_BASE_DN.equals(entryDN))
    {
      return -1;
    }
@@ -329,11 +357,9 @@
      return 1;
    }
    // Search with cookie mode to count all update messages
    final Set<String> excludedDomains = MultimasterReplication.getECLDisabledDomains();
    excludedDomains.add(DN_EXTERNAL_CHANGELOG_ROOT);
    SearchParams params = new SearchParams("0", excludedDomains);
    final SearchParams params = new SearchParams(getExcludedDomains());
    params.requestType = REQUEST_TYPE_FROM_COOKIE;
    params.multiDomainServerState = new MultiDomainServerState();
    params.cookie = new MultiDomainServerState();
    NumSubordinatesSearchOperation searchOp = new NumSubordinatesSearchOperation();
    try
    {
@@ -342,11 +368,118 @@
    catch (ChangelogException e)
    {
      throw new DirectoryException(ResultCode.UNWILLING_TO_PERFORM, ERR_CHANGELOG_BACKEND_NUM_SUBORDINATES.get(
          baseChangelogDN.toString(), stackTraceToSingleLineString(e)));
          CHANGELOG_BASE_DN.toString(), stackTraceToSingleLineString(e)));
    }
    return searchOp.numSubordinates;
  }

  private Set<String> getExcludedDomains()
  {
    final Set<String> domains = MultimasterReplication.getECLDisabledDomains();
    domains.add(DN_EXTERNAL_CHANGELOG_ROOT);
    return domains;
  }

  /**
   * Notifies persistent searches of this backend that a new entry was added to it.
   * <p>
   * Note: This method is called in a multi-threaded context.
   *
   * @param baseDN
   *          the baseDN of the newly added entry.
   * @param changeNumber
   *          the change number of the newly added entry. It will be greater
   *          than zero for entries added to the change number index and less
   *          than or equal to zero for entries added to any replica DB
   * @param cookieString
   *          a string representing the cookie of the newly added entry.
   *          This is only meaningful for entries added to the change number index
   * @param updateMsg
   *          the update message of the newly added entry
   * @throws ChangelogException
   *           If a problem occurs while notifying of the newly added entry.
   */
  public void notifyEntryAdded(DN baseDN, long changeNumber, String cookieString, UpdateMsg updateMsg)
      throws ChangelogException
  {
    final boolean isCookieEntry = changeNumber <= 0;
    final List<SearchOperation> pSearchOps = getPersistentSearches(isCookieEntry);
    if (pSearchOps.isEmpty() || !(updateMsg instanceof LDAPUpdateMsg))
    {
      return;
    }

    try
    {
      final Entry entry = createEntryFromMsg(baseDN, changeNumber, cookieString, updateMsg);
      for (SearchOperation pSearchOp : pSearchOps)
      {
        final EntrySender entrySender = (EntrySender)
            pSearchOp.getAttachment(OID_ECL_COOKIE_EXCHANGE_CONTROL);

        // when returning changesOnly, the first incoming update must return
        // the base entry before any other changes,
        // so force sending now, when protected by the synchronized block
        if (isCookieEntry)
        { // cookie based search
          final String cookieStr;
          synchronized (entrySender)
          { // forbid concurrent updates to the cookie
            entrySender.cookie.update(baseDN, updateMsg.getCSN());
            cookieStr = entrySender.cookie.toString();

            entrySender.sendBaseChangelogEntry(true);
          }
          final Entry entry2 = createEntryFromMsg(baseDN, changeNumber, cookieStr, updateMsg);
          // FIXME JNR use this instead of previous line:
          // entry.replaceAttribute(Attributes.create("changelogcookie", cookieStr));
          entrySender.sendEntryIfMatches(entry2, cookieStr);
        }
        else
        { // draft changeNumber search
          if (!entrySender.hasReturnedBaseEntry.get())
          {
            synchronized (entrySender)
            {
              entrySender.sendBaseChangelogEntry(true);
            }
          }
          entrySender.sendEntryIfMatches(entry, null);
        }
      }
    }
    catch (DirectoryException e)
    {
      throw new ChangelogException(e.getMessageObject(), e);
    }
  }

  private List<SearchOperation> getPersistentSearches(boolean wantCookieBasedSearch)
  {
    final List<SearchOperation> results = new ArrayList<SearchOperation>();
    for (PersistentSearch pSearch : getPersistentSearches())
    {
      final SearchOperation op = pSearch.getSearchOperation();
      if (wantCookieBasedSearch == isCookieBased(op))
      {
        results.add(op);
      }
    }
    return results;
  }

  private boolean isCookieBased(final SearchOperation searchOp)
  {
    for (Control c : searchOp.getRequestControls())
    {
      if (OID_ECL_COOKIE_EXCHANGE_CONTROL.equals(c.getOID()))
      {
        return true;
      }
    }
    return false;
  }

  /** {@inheritDoc} */
  @Override
  public void addEntry(Entry entry, AddOperation addOperation)
@@ -409,9 +542,7 @@

  private SearchParams buildSearchParameters(final SearchOperation searchOperation) throws DirectoryException
  {
    final Set<String> excludedDomains = MultimasterReplication.getECLDisabledDomains();
    excludedDomains.add(DN_EXTERNAL_CHANGELOG_ROOT);
    final SearchParams params = new SearchParams(searchOperation.toString(), excludedDomains);
    final SearchParams params = new SearchParams(getExcludedDomains());
    final ExternalChangelogRequestControl eclRequestControl =
        searchOperation.getRequestControl(ExternalChangelogRequestControl.DECODER);
    if (eclRequestControl == null)
@@ -421,7 +552,7 @@
    else
    {
      params.requestType = REQUEST_TYPE_FROM_COOKIE;
      params.multiDomainServerState = eclRequestControl.getCookie();
      params.cookie = eclRequestControl.getCookie();
    }
    return params;
  }
@@ -523,7 +654,7 @@
  {
    try
    {
      return numSubordinates(baseChangelogDN, true) + 1;
      return numSubordinates(CHANGELOG_BASE_DN, true) + 1;
    }
    catch (DirectoryException e)
    {
@@ -543,33 +674,28 @@
  static class SearchParams
  {
    private ECLRequestType requestType;
    private final String operationId;
    private final Set<String> excludedBaseDNs;
    private long lowestChangeNumber = -1;
    private long highestChangeNumber = -1;
    private CSN csn = new CSN(0, 0, 0);
    private MultiDomainServerState multiDomainServerState;
    private MultiDomainServerState cookie;

    /**
     * Creates search parameters.
     */
    SearchParams()
    {
      operationId = "";
      excludedBaseDNs = Collections.emptySet();
      this.excludedBaseDNs = Collections.emptySet();
    }

    /**
     * Creates search parameters with provided id and excluded domain DNs.
     *
     * @param operationId
     *          The id of the operation.
     * @param excludedBaseDNs
     *          Set of DNs to exclude from search.
     */
    SearchParams(final String operationId, final Set<String> excludedBaseDNs)
    SearchParams(final Set<String> excludedBaseDNs)
    {
      this.operationId = operationId;
      this.excludedBaseDNs = excludedBaseDNs;
    }

@@ -802,45 +928,40 @@
  private void searchFromCookie(final SearchParams searchParams, final SearchOperation searchOperation)
      throws DirectoryException, ChangelogException
  {
    final ReplicationDomainDB replicationDomainDB = replicationServer.getChangelogDB().getReplicationDomainDB();
    validateProvidedCookie(searchParams);
    final boolean isPersistentSearch = isPersistentSearch(searchOperation);

    boolean hasReturnedBaseEntry = false;
    final EntrySender entrySender = new EntrySender(searchOperation, searchParams.cookie);
    if (isPersistentSearch)
    {
      searchOperation.setAttachment(OID_ECL_COOKIE_EXCHANGE_CONTROL, entrySender);
    }

    ECLMultiDomainDBCursor replicaUpdatesCursor = null;
    try
    {
      final ReplicationDomainDB replicationDomainDB = getChangelogDB().getReplicationDomainDB();
      final MultiDomainDBCursor cursor = replicationDomainDB.getCursorFrom(
          searchParams.multiDomainServerState, AFTER_MATCHING_KEY, searchParams.getExcludedBaseDNs());
          searchParams.cookie, AFTER_MATCHING_KEY, searchParams.getExcludedBaseDNs());
      replicaUpdatesCursor = new ECLMultiDomainDBCursor(domainPredicate, cursor);

      MultiDomainServerState cookie = searchParams.multiDomainServerState;
      boolean continueSearch = true;
      while (continueSearch && replicaUpdatesCursor.next())
      {
        // Handle creation of base changelog entry on first update message found
        if (!hasReturnedBaseEntry)
        {
          if (!returnBaseChangelogEntry(searchOperation, true))
          {
            return;
          }
          hasReturnedBaseEntry = true;
        }
        // Handle the update message
        final UpdateMsg updateMsg = replicaUpdatesCursor.getRecord();
        final DN domainBaseDN = replicaUpdatesCursor.getData();
        cookie.update(domainBaseDN, updateMsg.getCSN());
        final Entry entry = createEntryFromMsg(domainBaseDN, 0L, cookie.toString(), updateMsg);
        if (matchBaseAndScopeAndFilter(entry, searchOperation))
        {
          Control control = new EntryChangelogNotificationControl(true, cookie.toString());
          continueSearch = searchOperation.returnEntry(entry, Arrays.asList(control));
        }
        searchParams.cookie.update(domainBaseDN, updateMsg.getCSN());
        final String cookieString = searchParams.cookie.toString();

        final Entry entry = createEntryFromMsg(domainBaseDN, 0, cookieString, updateMsg);
        continueSearch = entrySender.sendEntryIfMatches(entry, cookieString);
      }
      // Handle creation of base changelog entry when no update message is found
      if (!hasReturnedBaseEntry)

      if (!isPersistentSearch)
      {
        returnBaseChangelogEntry(searchOperation, false);
        // send the base changelog entry if no update message is found
        entrySender.sendBaseChangelogEntry(false);
      }
    }
    finally
@@ -849,6 +970,52 @@
    }
  }

  private boolean isPersistentSearch(SearchOperation op)
  {
    for (PersistentSearch pSearch : getPersistentSearches())
    {
      if (op == pSearch.getSearchOperation())
      {
        return true;
      }
    }
    return false;
  }

  /** {@inheritDoc} */
  @Override
  public void registerPersistentSearch(PersistentSearch pSearch)
  {
    super.registerPersistentSearch(pSearch);

    final SearchOperation searchOp = pSearch.getSearchOperation();
    if (pSearch.isChangesOnly())
    {
      // this persistent search will not go through #search0() down below
      // so we must initialize the cookie here
      searchOp.setAttachment(OID_ECL_COOKIE_EXCHANGE_CONTROL,
          new EntrySender(searchOp, getNewestCookie(searchOp)));
    }
  }

  private MultiDomainServerState getNewestCookie(SearchOperation searchOp)
  {
    if (!isCookieBased(searchOp))
    {
      return null;
    }

    final MultiDomainServerState cookie = new MultiDomainServerState();
    for (final Iterator<ReplicationServerDomain> it =
        replicationServer.getDomainIterator(); it.hasNext();)
    {
      final DN baseDN = it.next().getBaseDN();
      final ServerState state = getChangelogDB().getReplicationDomainDB().getDomainNewestCSNs(baseDN);
      cookie.update(baseDN, state);
    }
    return cookie;
  }

  /**
   * Validates the cookie contained in search parameters by checking its content
   * with the actual replication server state.
@@ -858,7 +1025,7 @@
   */
  private void validateProvidedCookie(final SearchParams searchParams) throws DirectoryException
  {
    final MultiDomainServerState state = searchParams.multiDomainServerState;
    final MultiDomainServerState state = searchParams.cookie;
    if (state != null && !state.isEmpty())
    {
      replicationServer.validateServerState(state, searchParams.getExcludedBaseDNs());
@@ -871,102 +1038,67 @@
  private void searchFromChangeNumber(final SearchParams params, final SearchOperation searchOperation)
      throws ChangelogException, DirectoryException
  {
    boolean hasReturnedBaseEntry = false;
    final ChangelogDB changelogDB = replicationServer.getChangelogDB();
    final EntrySender entrySender = new EntrySender(searchOperation, null);
    final boolean isPersistentSearch = isPersistentSearch(searchOperation);
    if (isPersistentSearch)
    {
      searchOperation.setAttachment(OID_ECL_COOKIE_EXCHANGE_CONTROL, entrySender);
    }

    DBCursor<ChangeNumberIndexRecord> cnIndexDBCursor = null;
    MultiDomainDBCursor replicaUpdatesCursor = null;
    try {
      cnIndexDBCursor = getCNIndexDBCursor(changelogDB, params.lowestChangeNumber);
    try
    {
      cnIndexDBCursor = getCNIndexDBCursor(params.lowestChangeNumber);
      boolean continueSearch = true;
      while (continueSearch && cnIndexDBCursor.next())
      {
        // Handle creation of base changelog entry on cnIndex record found
        if (!hasReturnedBaseEntry)
        {
          if (!returnBaseChangelogEntry(searchOperation, true))
          {
            return;
          }
          hasReturnedBaseEntry = true;
        }
        // Handle the current cnIndex record
        final ChangeNumberIndexRecord cnIndexRecord = cnIndexDBCursor.getRecord();
        if (replicaUpdatesCursor == null)
        {
          replicaUpdatesCursor = initializeReplicaUpdatesCursor(changelogDB, cnIndexRecord);
          replicaUpdatesCursor = initializeReplicaUpdatesCursor(cnIndexRecord);
        }
        continueSearch = params.changeNumberIsInRange(cnIndexRecord.getChangeNumber());
        if (continueSearch)
        {
           UpdateMsg updateMsg = findReplicaUpdateMessage(cnIndexRecord, replicaUpdatesCursor);
           if (updateMsg != null)
           {
             continueSearch = returnEntryForUpdateMessage(searchOperation, cnIndexRecord, updateMsg);
             replicaUpdatesCursor.next();
           }
          UpdateMsg updateMsg = findReplicaUpdateMessage(cnIndexRecord, replicaUpdatesCursor);
          if (updateMsg != null)
          {
            continueSearch = sendEntryForUpdateMessage(entrySender, cnIndexRecord, updateMsg);
            replicaUpdatesCursor.next();
          }
        }
      }
      // Handle creation of base changelog entry when no update message is found
      if (!hasReturnedBaseEntry)

      if (!isPersistentSearch)
      {
        returnBaseChangelogEntry(searchOperation, false);
        // send the base changelog entry if no update message is found
        entrySender.sendBaseChangelogEntry(false);
      }
    }
    finally {
    finally
    {
      StaticUtils.close(cnIndexDBCursor, replicaUpdatesCursor);
    }
  }

  /**
   * Create and returns the base changelog entry to provided search operation.
   *
   * @return {@code true} if search should continue, {@code false} otherwise
   */
  private boolean returnBaseChangelogEntry(final SearchOperation searchOperation, boolean hasSubordinates)
      throws DirectoryException
  private boolean sendEntryForUpdateMessage(EntrySender entrySender,
      ChangeNumberIndexRecord cnIndexRecord, UpdateMsg updateMsg) throws DirectoryException
  {
    final DN baseDN = searchOperation.getBaseDN();
    final SearchFilter filter = searchOperation.getFilter();
    final SearchScope scope = searchOperation.getScope();

    if (baseChangelogDN.matchesBaseAndScope(baseDN, scope))
    {
      final Entry entry = buildBaseChangelogEntry(hasSubordinates);
      if (filter.matchesEntry(entry) && !searchOperation.returnEntry(entry, null))
      {
        // Abandon, size limit reached.
        return false;
      }
    }
    if (baseDN.equals(baseChangelogDN) && scope.equals(SearchScope.BASE_OBJECT))
    {
      // Only the change log root entry was requested
      return false;
    }
    return true;
  }

  /**
   * @return {@code true} if search should continue, {@code false} otherwise
   */
  private boolean returnEntryForUpdateMessage(
      final SearchOperation searchOperation,
      final ChangeNumberIndexRecord cnIndexRecord,
      final UpdateMsg updateMsg)
          throws DirectoryException
  {
    final DN baseDN = cnIndexRecord.getBaseDN();
    final MultiDomainServerState cookie = new MultiDomainServerState(cnIndexRecord.getPreviousCookie());
    final DN changeDN = cnIndexRecord.getBaseDN();
    cookie.update(changeDN, cnIndexRecord.getCSN());
    final Entry entry = createEntryFromMsg(changeDN, cnIndexRecord.getChangeNumber(), cookie.toString(), updateMsg);
    if (matchBaseAndScopeAndFilter(entry, searchOperation))
    {
      return searchOperation.returnEntry(entry, null);
    }
    return true;
    cookie.update(baseDN, cnIndexRecord.getCSN());
    final String cookieString = cookie.toString();

    final Entry entry = createEntryFromMsg(baseDN, cnIndexRecord.getChangeNumber(), cookieString, updateMsg);
    return entrySender.sendEntryIfMatches(entry, null);
  }

  private MultiDomainDBCursor initializeReplicaUpdatesCursor(final ChangelogDB changelogDB,
  private MultiDomainDBCursor initializeReplicaUpdatesCursor(
      final ChangeNumberIndexRecord cnIndexRecord) throws ChangelogException
  {
    final MultiDomainServerState state = new MultiDomainServerState();
@@ -975,7 +1107,7 @@
    // No need for ECLMultiDomainDBCursor in this case
    // as updateMsg will be matched with cnIndexRecord
    final MultiDomainDBCursor replicaUpdatesCursor =
        changelogDB.getReplicationDomainDB().getCursorFrom(state, ON_MATCHING_KEY);
        getChangelogDB().getReplicationDomainDB().getCursorFrom(state, ON_MATCHING_KEY);
    replicaUpdatesCursor.next();
    return replicaUpdatesCursor;
  }
@@ -1023,10 +1155,10 @@
  }

  /** Returns a cursor on CNIndexDB for the provided first change number. */
  private DBCursor<ChangeNumberIndexRecord> getCNIndexDBCursor(final ChangelogDB changelogDB,
  private DBCursor<ChangeNumberIndexRecord> getCNIndexDBCursor(
      final long firstChangeNumber) throws ChangelogException
  {
    final ChangeNumberIndexDB cnIndexDB = changelogDB.getChangeNumberIndexDB();
    final ChangeNumberIndexDB cnIndexDB = getChangelogDB().getChangeNumberIndexDB();
    long changeNumberToUse = firstChangeNumber;
    if (changeNumberToUse <= 1)
    {
@@ -1036,31 +1168,6 @@
    return cnIndexDB.getCursorFrom(changeNumberToUse);
  }

  /** Indicates if the provided entry matches the filter, base and scope. */
  private boolean matchBaseAndScopeAndFilter(Entry entry, SearchOperation searchOp) throws DirectoryException
  {
    return entry.matchesBaseAndScope(searchOp.getBaseDN(), searchOp.getScope())
        && searchOp.getFilter().matchesEntry(entry);
  }

  /**
   * Retrieves the base changelog entry.
   */
  private Entry buildBaseChangelogEntry(boolean hasSubordinates)
  {
    final Map<AttributeType, List<Attribute>> userAttrs = new LinkedHashMap<AttributeType,List<Attribute>>();
    final Map<AttributeType, List<Attribute>> operationalAttrs = new LinkedHashMap<AttributeType,List<Attribute>>();

    addAttributeByUppercaseName(ATTR_COMMON_NAME, ATTR_COMMON_NAME, BACKEND_ID, userAttrs, operationalAttrs);
    addAttributeByUppercaseName(ATTR_SUBSCHEMA_SUBENTRY_LC, ATTR_SUBSCHEMA_SUBENTRY,
        ConfigConstants.DN_DEFAULT_SCHEMA_ROOT, userAttrs, operationalAttrs);
    addAttributeByUppercaseName("hassubordinates", "hasSubordinates", Boolean.toString(hasSubordinates),
        userAttrs, operationalAttrs);
    addAttributeByUppercaseName("entrydn", "entryDN", baseChangelogDN.toString(),
        userAttrs, operationalAttrs);
    return new Entry(baseChangelogDN, CHANGELOG_ROOT_OBJECT_CLASSES, userAttrs, operationalAttrs);
  }

  /**
   * Creates a changelog entry.
   */
@@ -1225,16 +1332,16 @@
  {
    final CSN csn = msg.getCSN();
    String dnString;
    if (changeNumber == 0)
    {
      // Cookie mode
      dnString = "replicationCSN=" + csn + "," + baseDN.toString() + "," + DN_EXTERNAL_CHANGELOG_ROOT;
    }
    else
    if (changeNumber > 0)
    {
      // Draft compat mode
      dnString = "changeNumber=" + changeNumber + "," + DN_EXTERNAL_CHANGELOG_ROOT;
    }
    else
    {
      // Cookie mode
      dnString = "replicationCSN=" + csn + "," + baseDN + "," + DN_EXTERNAL_CHANGELOG_ROOT;
    }

    final Map<AttributeType, List<Attribute>> userAttrs = new LinkedHashMap<AttributeType, List<Attribute>>();
    final Map<AttributeType, List<Attribute>> opAttrs = new LinkedHashMap<AttributeType, List<Attribute>>();
@@ -1247,7 +1354,7 @@
    addAttributeByType("entrydn", "entryDN", dnString, userAttrs, opAttrs);

    // REQUIRED attributes
    if (changeNumber != 0)
    if (changeNumber > 0)
    {
      addAttributeByType("changenumber", "changeNumber", String.valueOf(changeNumber), userAttrs, opAttrs);
    }
@@ -1277,7 +1384,8 @@
    {
      addAttributeByType("targetentryuuid", "targetEntryUUID", targetUUID, userAttrs, opAttrs);
    }
    addAttributeByType("changelogcookie", "changeLogCookie", cookie, userAttrs, opAttrs);
    final String cookie2 = cookie != null ? cookie : "";
    addAttributeByType("changelogcookie", "changeLogCookie", cookie2, userAttrs, opAttrs);

    final List<RawAttribute> includedAttributes = msg.getEclIncludes();
    if (includedAttributes != null && !includedAttributes.isEmpty())
@@ -1300,6 +1408,116 @@
    return new Entry(DN.decode(dnString), CHANGELOG_ENTRY_OBJECT_CLASSES, userAttrs, opAttrs);
  }

  /**
   * Used to send entries to searches on cn=changelog. This class ensures the
   * base changelog entry is sent before sending any other entry. It is also
   * used as a store when going from the "initial search" phase to the
   * "persistent search" phase.
   */
  private static class EntrySender
  {

    private final SearchOperation searchOp;
    /**
     * Used by the cookie-based searches to communicate the cookie between the
     * initial search phase and the persistent search phase. This is unused with
     * draft change number searches.
     */
    private final MultiDomainServerState cookie;
    private final AtomicBoolean hasReturnedBaseEntry = new AtomicBoolean();

    public EntrySender(SearchOperation searchOp, MultiDomainServerState cookie)
    {
      this.searchOp = searchOp;
      this.cookie = cookie;
    }

    /**
     * Sends the entry if it matches the base, scope and filter of the current search operation.
     * It will also send the base changelog entry if it needs to be sent and was not sent before.
     *
     * @return {@code true} if search should continue, {@code false} otherwise
     */
    private boolean sendEntryIfMatches(Entry entry, String cookie) throws DirectoryException
    {
      // About to send one entry: ensure the base changelog entry is sent first
      if (!sendBaseChangelogEntry(true))
      {
        // only return the base entry: stop here
        return false;
      }
      if (matchBaseAndScopeAndFilter(entry))
      {
        return searchOp.returnEntry(entry, getControls(cookie));
      }
      // maybe the next entry will match?
      return true;
    }

    /** Indicates if the provided entry matches the filter, base and scope. */
    private boolean matchBaseAndScopeAndFilter(Entry entry) throws DirectoryException
    {
      return entry.matchesBaseAndScope(searchOp.getBaseDN(), searchOp.getScope())
          && searchOp.getFilter().matchesEntry(entry);
    }

    private List<Control> getControls(String cookie)
    {
      if (cookie != null)
      {
        Control c = new EntryChangelogNotificationControl(true, cookie);
        return Arrays.asList(c);
      }
      return Collections.emptyList();
    }

    /**
     * Create and returns the base changelog entry to the underlying search operation.
     *
     * @return {@code true} if search should continue, {@code false} otherwise
     */
    private boolean sendBaseChangelogEntry(boolean hasSubordinates) throws DirectoryException
    {
      if (hasReturnedBaseEntry.compareAndSet(false, true))
      {
        final DN baseDN = searchOp.getBaseDN();
        final SearchFilter filter = searchOp.getFilter();
        final SearchScope scope = searchOp.getScope();

        if (ChangelogBackend.CHANGELOG_BASE_DN.matchesBaseAndScope(baseDN, scope))
        {
          final Entry entry = buildBaseChangelogEntry(hasSubordinates);
          if (filter.matchesEntry(entry) && !searchOp.returnEntry(entry, null))
          {
            // Abandon, size limit reached.
            return false;
          }
        }
        return !baseDN.equals(ChangelogBackend.CHANGELOG_BASE_DN)
            || !scope.equals(SearchScope.BASE_OBJECT);
      }
      return true;
    }

    private Entry buildBaseChangelogEntry(boolean hasSubordinates)
    {
      final Map<AttributeType, List<Attribute>> userAttrs =
          new LinkedHashMap<AttributeType, List<Attribute>>();
      final Map<AttributeType, List<Attribute>> operationalAttrs =
          new LinkedHashMap<AttributeType, List<Attribute>>();

      addAttributeByUppercaseName(ATTR_COMMON_NAME, ATTR_COMMON_NAME,
          ChangelogBackend.BACKEND_ID, userAttrs, operationalAttrs);
      addAttributeByUppercaseName(ATTR_SUBSCHEMA_SUBENTRY_LC, ATTR_SUBSCHEMA_SUBENTRY,
          ConfigConstants.DN_DEFAULT_SCHEMA_ROOT, userAttrs, operationalAttrs);
      addAttributeByUppercaseName("hassubordinates", "hasSubordinates",
          Boolean.toString(hasSubordinates), userAttrs, operationalAttrs);
      addAttributeByUppercaseName("entrydn", "entryDN",
          ServerConstants.DN_EXTERNAL_CHANGELOG_ROOT, userAttrs, operationalAttrs);
      return new Entry(CHANGELOG_BASE_DN, CHANGELOG_ROOT_OBJECT_CLASSES, userAttrs, operationalAttrs);
    }
  }

  private static void addAttribute(final Entry e, final String attrType, final String attrValue)
  {
    e.addAttribute(Attributes.create(attrType, attrValue), null);
@@ -1313,7 +1531,7 @@
    addAttribute(attrNameLowercase, attrNameUppercase, attrValue, userAttrs, operationalAttrs, true);
  }

  private void addAttributeByUppercaseName(String attrNameLowercase,
  private static void addAttributeByUppercaseName(String attrNameLowercase,
      String attrNameUppercase,  String attrValue,
      Map<AttributeType, List<Attribute>> userAttrs,
      Map<AttributeType, List<Attribute>> operationalAttrs)
@@ -1331,8 +1549,9 @@
    {
      attrType = DirectoryServer.getDefaultAttributeType(attrNameUppercase);
    }
    final Attribute a = addByType ?
        Attributes.create(attrType, attrValue) : Attributes.create(attrNameUppercase, attrValue);
    final Attribute a = addByType
        ? Attributes.create(attrType, attrValue)
        : Attributes.create(attrNameUppercase, attrValue);
    final List<Attribute> attrList = Collections.singletonList(a);
    if (attrType.isOperational())
    {

 opendj-sdk/opends/src/server/org/opends/server/backends/MonitorBackend.java

@@ -26,14 +26,6 @@
 */
package org.opends.server.backends;

import static org.opends.messages.BackendMessages.*;
import static org.opends.messages.ConfigMessages.*;
import static org.opends.server.config.ConfigConstants.*;
import static org.opends.server.loggers.debug.DebugLogger.debugEnabled;
import static org.opends.server.loggers.debug.DebugLogger.getTracer;
import static org.opends.server.util.ServerConstants.*;
import static org.opends.server.util.StaticUtils.*;

import java.util.*;

import org.opends.messages.Message;
@@ -51,6 +43,13 @@
import org.opends.server.util.TimeThread;
import org.opends.server.util.Validator;

import static org.opends.messages.BackendMessages.*;
import static org.opends.messages.ConfigMessages.*;
import static org.opends.server.config.ConfigConstants.*;
import static org.opends.server.loggers.debug.DebugLogger.*;
import static org.opends.server.util.ServerConstants.*;
import static org.opends.server.util.StaticUtils.*;

/**
 * This class defines a backend to hold Directory Server monitor entries. It
 * will not actually store anything, but upon request will retrieve the
@@ -65,9 +64,10 @@
   */
  private static final DebugTracer TRACER = getTracer();

  /** The set of user-defined attributes that will be included in the base
  /**
   * The set of user-defined attributes that will be included in the base
   * monitor entry.
*/
   */
  private ArrayList<Attribute> userDefinedAttributes;

  /** The set of objectclasses that will be used in monitor entries. */
@@ -349,6 +349,7 @@
  @Override
  public void finalizeBackend()
  {
    super.finalizeBackend();
    currentConfig.removeMonitorChangeListener(this);
    try
    {

 opendj-sdk/opends/src/server/org/opends/server/backends/NullBackend.java

@@ -236,6 +236,7 @@
  @Override
  public synchronized void finalizeBackend()
  {
    super.finalizeBackend();
    for (DN dn : baseDNs)
    {
      try

 opendj-sdk/opends/src/server/org/opends/server/backends/RootDSEBackend.java

@@ -290,6 +290,7 @@
  @Override
  public void finalizeBackend()
  {
    super.finalizeBackend();
    currentConfig.removeChangeListener(this);
  }


 opendj-sdk/opends/src/server/org/opends/server/backends/SchemaBackend.java

@@ -51,11 +51,12 @@
import java.util.zip.ZipEntry;
import java.util.zip.ZipInputStream;
import java.util.zip.ZipOutputStream;

import javax.crypto.Mac;

import org.opends.messages.Message;
import org.opends.server.admin.std.server.SchemaBackendCfg;
import org.opends.server.admin.server.ConfigurationChangeListener;
import org.opends.server.admin.std.server.SchemaBackendCfg;
import org.opends.server.api.AlertGenerator;
import org.opends.server.api.Backend;
import org.opends.server.api.ClientConnection;
@@ -89,8 +90,8 @@
import static org.opends.messages.ConfigMessages.*;
import static org.opends.messages.SchemaMessages.*;
import static org.opends.server.config.ConfigConstants.*;
import static org.opends.server.loggers.debug.DebugLogger.*;
import static org.opends.server.loggers.ErrorLogger.*;
import static org.opends.server.loggers.debug.DebugLogger.*;
import static org.opends.server.schema.SchemaConstants.*;
import static org.opends.server.util.ServerConstants.*;
import static org.opends.server.util.StaticUtils.*;
@@ -501,6 +502,7 @@
  @Override
  public void finalizeBackend()
  {
    super.finalizeBackend();
    currentConfig.removeSchemaChangeListener(this);

    for (DN baseDN : baseDNs)

 opendj-sdk/opends/src/server/org/opends/server/backends/TrustStoreBackend.java

@@ -26,12 +26,6 @@
 */
package org.opends.server.backends;

import static org.opends.messages.BackendMessages.*;
import static org.opends.server.config.ConfigConstants.*;
import static org.opends.server.loggers.debug.DebugLogger.*;
import static org.opends.server.util.ServerConstants.*;
import static org.opends.server.util.StaticUtils.*;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
@@ -79,6 +73,12 @@
import org.opends.server.util.SetupUtils;
import org.opends.server.util.Validator;

import static org.opends.messages.BackendMessages.*;
import static org.opends.server.config.ConfigConstants.*;
import static org.opends.server.loggers.debug.DebugLogger.*;
import static org.opends.server.util.ServerConstants.*;
import static org.opends.server.util.StaticUtils.*;

/**
 * This class defines a backend used to provide an LDAP view of public keys
 * stored in a key store.
@@ -367,6 +367,7 @@
  @Override
  public void finalizeBackend()
  {
    super.finalizeBackend();
    configuration.addTrustStoreChangeListener(this);

    try

 opendj-sdk/opends/src/server/org/opends/server/backends/jeb/BackendImpl.java

@@ -332,7 +332,7 @@
  @Override
  public void finalizeBackend()
  {
    // Deregister as a change listener.
    super.finalizeBackend();
    cfg.removeLocalDBChangeListener(this);

    // Deregister our base DNs.
@@ -371,24 +371,18 @@
      {
        TRACER.debugCaught(DebugLogLevel.ERROR, e);
      }
      Message message = ERR_JEB_DATABASE_EXCEPTION.get(e.getMessage());
      logError(message);
      logError(ERR_JEB_DATABASE_EXCEPTION.get(e.getMessage()));
    }

    // Checksum this db environment and register its offline state id/checksum.
    DirectoryServer.registerOfflineBackendStateID(this.getBackendID(),
      checksumDbEnv());

    //Deregister the alert generator.
    DirectoryServer.registerOfflineBackendStateID(getBackendID(), checksumDbEnv());
    DirectoryServer.deregisterAlertGenerator(this);

    // Make sure the thread counts are zero for next initialization.
    threadTotalCount.set(0);
    threadWriteCount.set(0);

    // Log an informational message.
    Message message = NOTE_BACKEND_OFFLINE.get(cfg.getBackendId());
    logError(message);
    logError(NOTE_BACKEND_OFFLINE.get(cfg.getBackendId()));
  }

  /** {@inheritDoc} */

 opendj-sdk/opends/src/server/org/opends/server/backends/task/TaskBackend.java

@@ -268,9 +268,9 @@
  @Override
  public void finalizeBackend()
  {
    super.finalizeBackend();
    currentConfig.removeTaskChangeListener(this);


    try
    {
      taskScheduler.stopScheduler();

 opendj-sdk/opends/src/server/org/opends/server/core/PersistentSearch.java

@@ -34,13 +34,7 @@
import org.opends.server.controls.EntryChangeNotificationControl;
import org.opends.server.controls.PersistentSearchChangeType;
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.types.CancelResult;
import org.opends.server.types.Control;
import org.opends.server.types.DN;
import org.opends.server.types.DebugLogLevel;
import org.opends.server.types.DirectoryException;
import org.opends.server.types.Entry;
import org.opends.server.types.ResultCode;
import org.opends.server.types.*;

import static org.opends.server.controls.PersistentSearchChangeType.*;
import static org.opends.server.loggers.debug.DebugLogger.*;
@@ -113,10 +107,8 @@
      psearch.isCancelled = true;

      // The persistent search can no longer be cancelled.
      psearch.searchOperation.getClientConnection().deregisterPersistentSearch(
          psearch);
      psearch.searchOperation.getClientConnection().deregisterPersistentSearch(psearch);

      //Decrement of psearch count maintained by the server.
      DirectoryServer.deregisterPersistentSearch();

      // Notify any cancellation callbacks.
@@ -161,25 +153,33 @@
  /** The reference to the associated search operation. */
  private final SearchOperation searchOperation;


  /**
   * Indicates whether to only return entries that have been updated since the
   * beginning of the search.
   */
  private final boolean changesOnly;

  /**
   * Creates a new persistent search object with the provided
   * information.
   * Creates a new persistent search object with the provided information.
   *
   * @param searchOperation
   *          The search operation for this persistent search.
   * @param changeTypes
   *          The change types for which changes should be examined.
   * @param changesOnly
   *          whether to only return entries that have been updated since the
   *          beginning of the search
   * @param returnECs
   *          Indicates whether to include entry change notification
   *          controls in search result entries sent to the client.
   *          Indicates whether to include entry change notification controls in
   *          search result entries sent to the client.
   */
  public PersistentSearch(SearchOperation searchOperation,
      Set<PersistentSearchChangeType> changeTypes, boolean returnECs)
      Set<PersistentSearchChangeType> changeTypes, boolean changesOnly,
      boolean returnECs)
  {
    this.searchOperation = searchOperation;
    this.changeTypes = changeTypes;
    this.changesOnly = changesOnly;
    this.returnECs = returnECs;
  }

@@ -241,6 +241,18 @@
  }

  /**
   * Returns whether only entries updated after the beginning of this persistent
   * search should be returned.
   *
   * @return true if only entries updated after the beginning of this search
   *         should be returned, false otherwise
   */
  public boolean isChangesOnly()
  {
    return changesOnly;
  }

  /**
   * Notifies the persistent searches that an entry has been added.
   *
   * @param entry

 opendj-sdk/opends/src/server/org/opends/server/extensions/ConfigFileHandler.java

@@ -901,6 +901,7 @@
  @Override
  public void finalizeConfigHandler()
  {
    finalizeBackend();
    try
    {
      DirectoryServer.deregisterBaseDN(configRootEntry.getDN());
@@ -916,13 +917,6 @@

  /** {@inheritDoc} */
  @Override
  public void finalizeBackend()
  {
    // No implementation is required.
  }

  /** {@inheritDoc} */
  @Override
  public ConfigEntry getConfigRootEntry()
         throws ConfigException
  {

 opendj-sdk/opends/src/server/org/opends/server/replication/plugin/LDAPReplicationDomain.java

@@ -76,7 +76,6 @@
import org.opends.server.types.operation.*;
import org.opends.server.util.LDIFReader;
import org.opends.server.util.TimeThread;
import org.opends.server.workflowelement.externalchangelog.ECLWorkflowElement;
import org.opends.server.workflowelement.localbackend.LocalBackendModifyOperation;

import static org.opends.messages.ReplicationMessages.*;
@@ -475,7 +474,7 @@
    storeECLConfiguration(configuration);
    solveConflictFlag = isSolveConflict(configuration);

    Backend backend = getBackend();
    Backend<?> backend = getBackend();
    if (backend == null)
    {
      throw new ConfigException(ERR_SEARCHING_DOMAIN_BACKEND.get(
@@ -3490,7 +3489,7 @@
  private long exportBackend(OutputStream output, boolean checksumOutput)
      throws DirectoryException
  {
    Backend backend = getBackend();
    Backend<?> backend = getBackend();

    //  Acquire a shared lock for the backend.
    try
@@ -3623,7 +3622,7 @@
   * @throws DirectoryException
   *           If the backend could not be disabled or locked exclusively.
   */
  private void preBackendImport(Backend backend) throws DirectoryException
  private void preBackendImport(Backend<?> backend) throws DirectoryException
  {
    // Stop saving state
    stateSavingDisabled = true;
@@ -3653,10 +3652,9 @@
  @Override
  protected void importBackend(InputStream input) throws DirectoryException
  {
    Backend<?> backend = getBackend();

    LDIFImportConfig importConfig = null;

    Backend backend = getBackend();

    ImportExportContext ieCtx = getImportExportContext();
    try
    {
@@ -3742,7 +3740,7 @@
   * @param backend The backend implied in the import.
   * @exception DirectoryException Thrown when an error occurs.
   */
  private void closeBackendImport(Backend backend) throws DirectoryException
  private void closeBackendImport(Backend<?> backend) throws DirectoryException
  {
    String lockFile = LockFileManager.getBackendLockFileName(backend);
    StringBuilder failureReason = new StringBuilder();
@@ -3810,7 +3808,7 @@
   * Returns the backend associated to this domain.
   * @return The associated backend.
   */
  private Backend getBackend()
  private Backend<?> getBackend()
  {
    return DirectoryServer.getBackend(getBaseDN());
  }
@@ -4098,30 +4096,6 @@

    super.sessionInitiated(initStatus, rsState);

    // Now that we are connected , we can enable ECL if :
    // 1/ RS must in the same JVM and created an ECL_WORKFLOW_ELEMENT
    // and 2/ this domain must NOT be private
    if (!getBackend().isPrivateBackend())
    {
      try
      {
        ECLWorkflowElement wfe = (ECLWorkflowElement)
        DirectoryServer.getWorkflowElement(
            ECLWorkflowElement.ECL_WORKFLOW_ELEMENT);
        if (wfe != null)
        {
          wfe.getReplicationServer().enableECL();
        }
      }
      catch (DirectoryException de)
      {
        logError(NOTE_ERR_UNABLE_TO_ENABLE_ECL.get(
            "Replication Domain on " + getBaseDNString(),
            stackTraceToSingleLineString(de)));
        // and go on
      }
    }

    // Now for bad data set status if needed
    if (forceBadDataSet)
    {
@@ -4375,7 +4349,7 @@
  @Override
  public long countEntries() throws DirectoryException
  {
    Backend backend = getBackend();
    Backend<?> backend = getBackend();
    if (!backend.supportsLDIFExport())
    {
      Message msg = ERR_INIT_EXPORT_NOT_SUPPORTED.get(backend.getBackendID());

 opendj-sdk/opends/src/server/org/opends/server/replication/server/ECLServerHandler.java

@@ -34,12 +34,16 @@
import org.opends.messages.Category;
import org.opends.messages.Message;
import org.opends.messages.Severity;
import org.opends.server.backends.ChangelogBackend;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.MultiDomainServerState;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.common.ServerStatus;
import org.opends.server.replication.protocol.*;
import org.opends.server.replication.server.changelog.api.*;
import org.opends.server.replication.server.changelog.api.ChangeNumberIndexDB;
import org.opends.server.replication.server.changelog.api.ChangeNumberIndexRecord;
import org.opends.server.replication.server.changelog.api.ChangelogException;
import org.opends.server.replication.server.changelog.api.DBCursor;
import org.opends.server.types.*;
import org.opends.server.util.ServerConstants;

@@ -47,10 +51,8 @@
import static org.opends.server.loggers.ErrorLogger.*;
import static org.opends.server.loggers.debug.DebugLogger.*;
import static org.opends.server.replication.protocol.ProtocolVersion.*;
import static org.opends.server.replication.protocol.StartECLSessionMsg
.ECLRequestType.*;
import static org.opends.server.replication.protocol.StartECLSessionMsg
.Persistent.*;
import static org.opends.server.replication.protocol.StartECLSessionMsg.ECLRequestType.*;
import static org.opends.server.replication.protocol.StartECLSessionMsg.Persistent.*;
import static org.opends.server.util.StaticUtils.*;

/**
@@ -150,8 +152,13 @@
    private final ReplicationServerDomain rsDomain;

    /**
     * Active when there are still changes supposed eligible for the ECL. It is
     * active by default.
     * Active when there are still changes supposed eligible for the ECL.
     * Here is the lifecycle of this field:
     * <ol>
     * <li>active==true at the start of the INIT phase,</li>
     * <li>active==false when there are no more changes for a domain in the the INIT phase,</li>
     * <li>active==true if it is a persistent search on external changelog. It never moves again</li>
     * </ol>
     */
    private boolean active = true;
    private UpdateMsg nextMsg;
@@ -349,8 +356,7 @@
    super(session, queueSize, replicationServer, rcvWindowSize);
    try
    {
      DN baseDN = DN.decode(ServerConstants.DN_EXTERNAL_CHANGELOG_ROOT);
      setBaseDNAndDomain(baseDN, true);
      setBaseDNAndDomain(ChangelogBackend.CHANGELOG_BASE_DN, true);
    }
    catch(DirectoryException de)
    {
@@ -849,14 +855,6 @@
  }

  /**
   * Registers this handler into its related domain and notifies the domain.
   */
  private void registerIntoDomain()
  {
    replicationServerDomain.registerHandler(this);
  }

  /**
   * Shutdown this handler.
   */
  @Override
@@ -867,16 +865,23 @@
      TRACER.debugInfo(this + " shutdown()");
    }
    releaseCursor();
    for (DomainContext domainCtxt : domainCtxts) {
      if (!domainCtxt.unRegisterHandler()) {
        logError(Message.raw(Category.SYNC, Severity.NOTICE,
            this + " shutdown() - error when unregistering handler "
                + domainCtxt.mh));

    if (domainCtxts != null)
    {
      for (DomainContext domainCtxt : domainCtxts)
      {
        if (!domainCtxt.unRegisterHandler())
        {
          logError(Message.raw(Category.SYNC, Severity.NOTICE, this
              + " shutdown() - error when unregistering handler "
              + domainCtxt.mh));
        }
        domainCtxt.stopServer();
      }
      domainCtxt.stopServer();
      domainCtxts = null;
    }

    super.shutdown();
    domainCtxts = null;
  }

  private void releaseCursor()
@@ -1018,11 +1023,11 @@
      closeInitPhase();
    }

    registerIntoDomain();
    replicationServerDomain.registerHandler(this);

    if (debugEnabled())
    {
      TRACER.debugInfo(getClass().getCanonicalName() + " " + getOperationId()
      TRACER.debugInfo(getClass().getSimpleName() + " " + getOperationId()
          + " initialized: " + " " + dumpState() + domaimCtxtsToString(""));
    }
  }
@@ -1373,7 +1378,7 @@
          + dumpState());
    }

    // go to persistent phase if one
    // set all domains to be active again for the persistent phase
    for (DomainContext domainCtxt : domainCtxts) domainCtxt.active = true;

    if (startECLSessionMsg.getPersistent() != NON_PERSISTENT)

 opendj-sdk/opends/src/server/org/opends/server/replication/server/ECLServerWriter.java

@@ -29,7 +29,7 @@
import java.io.IOException;
import java.net.SocketException;

import org.opends.server.core.DirectoryServer;
import org.opends.server.backends.ChangelogBackend;
import org.opends.server.core.PersistentSearch;
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.replication.protocol.DoneMsg;
@@ -40,7 +40,6 @@
import org.opends.server.types.DirectoryException;
import org.opends.server.types.Entry;
import org.opends.server.workflowelement.externalchangelog.ECLSearchOperation;
import org.opends.server.workflowelement.externalchangelog.ECLWorkflowElement;

import static org.opends.messages.ReplicationMessages.*;
import static org.opends.server.loggers.ErrorLogger.*;
@@ -95,9 +94,8 @@
   */
  private PersistentSearch findPersistentSearch(ECLServerHandler handler)
  {
    ECLWorkflowElement wfe = (ECLWorkflowElement)
        DirectoryServer.getWorkflowElement(ECLWorkflowElement.ECL_WORKFLOW_ELEMENT);
    for (PersistentSearch psearch : wfe.getPersistentSearches())
    final ChangelogBackend backend = ChangelogBackend.getInstance();
    for (PersistentSearch psearch : backend.getPersistentSearches())
    {
      if (psearch.getSearchOperation().toString().equals(
          handler.getOperationId()))

 opendj-sdk/opends/src/server/org/opends/server/replication/server/ReplicationServer.java

@@ -31,11 +31,8 @@
import java.util.*;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import org.opends.messages.Category;
import org.opends.messages.Message;
import org.opends.messages.Severity;
import org.opends.server.admin.server.ConfigurationChangeListener;
import org.opends.server.admin.std.meta.ReplicationServerCfgDefn.ReplicationDBImplementation;
import org.opends.server.admin.std.meta.VirtualAttributeCfgDefn.ConflictBehavior;
@@ -45,8 +42,6 @@
import org.opends.server.backends.ChangelogBackend;
import org.opends.server.config.ConfigException;
import org.opends.server.core.DirectoryServer;
import org.opends.server.core.WorkflowImpl;
import org.opends.server.core.networkgroups.NetworkGroup;
import org.opends.server.loggers.debug.DebugLogger;
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.replication.common.*;
@@ -61,9 +56,7 @@
import org.opends.server.replication.server.changelog.je.JEChangelogDB;
import org.opends.server.replication.service.DSRSShutdownSync;
import org.opends.server.types.*;
import org.opends.server.util.ServerConstants;
import org.opends.server.util.StaticUtils;
import org.opends.server.workflowelement.externalchangelog.ECLWorkflowElement;

import static org.opends.messages.ConfigMessages.*;
import static org.opends.messages.ReplicationMessages.*;
@@ -114,12 +107,6 @@
  /** The tracer object for the debug logger. */
  private static final DebugTracer TRACER = getTracer();

  private static final String eclWorkflowID =
    "External Changelog Workflow ID";
  private ECLWorkflowElement eclwe;
  private final AtomicReference<WorkflowImpl> eclWorkflowImpl =
      new AtomicReference<WorkflowImpl>();

  /**
   * This is required for unit testing, so that we can keep track of all the
   * replication servers which are running in the VM.
@@ -178,6 +165,8 @@
    this.config = cfg;
    this.dsrsShutdownSync = dsrsShutdownSync;
    this.domainPredicate = predicate;

    enableExternalChangeLog();
    ReplicationDBImplementation dbImpl = cfg.getReplicationDBImplementation();
    if (DebugLogger.debugEnabled())
    {
@@ -191,9 +180,6 @@
    initialize();
    cfg.addChangeListener(this);

    // TODO : uncomment to branch changelog backend
    //enableExternalChangeLog();

    localPorts.add(getReplicationPort());

    // Keep track of this new instance
@@ -464,15 +450,6 @@
      listenThread = new ReplicationServerListenThread(this);
      listenThread.start();

      // Creates the ECL workflow elem so that DS (LDAPReplicationDomain)
      // can know me and really enableECL.
      if (WorkflowImpl.getWorkflow(eclWorkflowID) != null)
      {
        // Already done. Nothing to do
        return;
      }
      eclwe = new ECLWorkflowElement(this);

      if (debugEnabled())
      {
        TRACER.debugInfo("RS " + getMonitorInstanceName()
@@ -486,51 +463,10 @@
      Message message = ERR_COULD_NOT_BIND_CHANGELOG.get(
          getReplicationPort(), e.getMessage());
      logError(message);
    } catch (DirectoryException e)
    {
      //FIXME:DirectoryException is raised by initializeECL => fix err msg
      Message message = Message.raw(Category.SYNC, Severity.SEVERE_ERROR,
      "Directory Exception raised by ECL initialization: " + e.getMessage());
      logError(message);
    }
  }

  /**
   * Enable the ECL access by creating a dedicated workflow element.
   * @throws DirectoryException when an error occurs.
   */
  public void enableECL() throws DirectoryException
  {
    if (eclWorkflowImpl.get() != null)
    {
      // ECL is already enabled, do nothing
      return;
    }

    // Create the workflow for the base DN
    // and register the workflow with the server.
    final DN dn = DN.decode(ServerConstants.DN_EXTERNAL_CHANGELOG_ROOT);
    final WorkflowImpl workflowImpl = new WorkflowImpl(eclWorkflowID, dn,
        eclwe.getWorkflowElementID(), eclwe);
    if (!eclWorkflowImpl.compareAndSet(null, workflowImpl))
    {
      // ECL is being enabled, do nothing
      return;
    }

    workflowImpl.register();

    NetworkGroup.getDefaultNetworkGroup().registerWorkflow(workflowImpl);

    // FIXME:ECL should the ECL Workflow be registered in admin and internal
    // network groups?
    NetworkGroup.getAdminNetworkGroup().registerWorkflow(workflowImpl);
    NetworkGroup.getInternalNetworkGroup().registerWorkflow(workflowImpl);

    registerVirtualAttributeRules();
  }

  /**
   * Enable the external changelog if it is not already enabled.
   * <p>
   * The external changelog is provided by the changelog backend.
@@ -646,34 +582,6 @@
    }
  }

  private void shutdownECL()
  {
    WorkflowImpl eclwf = (WorkflowImpl) WorkflowImpl.getWorkflow(eclWorkflowID);
    // do it only if not already done by another RS (unit test case)
    if (eclwf != null)
    {
      // FIXME:ECL should the ECL Workflow be registered in admin and internal
      // network groups?
      NetworkGroup.getInternalNetworkGroup().deregisterWorkflow(eclWorkflowID);
      NetworkGroup.getAdminNetworkGroup().deregisterWorkflow(eclWorkflowID);

      NetworkGroup.getDefaultNetworkGroup().deregisterWorkflow(eclWorkflowID);

      deregisterVirtualAttributeRules();

      eclwf.deregister();
      eclwf.finalizeWorkflow();
    }

    eclwe = (ECLWorkflowElement) DirectoryServer
        .getWorkflowElement("EXTERNAL CHANGE LOG");
    if (eclwe != null)
    {
      DirectoryServer.deregisterWorkflowElement(eclwe);
      eclwe.finalizeWorkflowElement();
    }
  }

  /**
   * Get the ReplicationServerDomain associated to the base DN given in
   * parameter.
@@ -844,9 +752,7 @@
      domain.shutdown();
    }

    // TODO : switch to second method when changelog backend is branched
    shutdownECL();
    //shutdownExternalChangelog();
    shutdownExternalChangelog();

    try
    {

 opendj-sdk/opends/src/server/org/opends/server/replication/server/changelog/file/FileChangelogDB.java

@@ -37,6 +37,7 @@
import org.opends.messages.MessageBuilder;
import org.opends.server.admin.std.server.ReplicationServerCfg;
import org.opends.server.api.DirectoryThread;
import org.opends.server.backends.ChangelogBackend;
import org.opends.server.config.ConfigException;
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.replication.common.CSN;
@@ -797,6 +798,8 @@
    final FileReplicaDB replicaDB = pair.getFirst();
    replicaDB.add(updateMsg);

    ChangelogBackend.getInstance().notifyEntryAdded(baseDN, 0, null, updateMsg);

    final ChangeNumberIndexer indexer = cnIndexer.get();
    if (indexer != null)
    {

 opendj-sdk/opends/src/server/org/opends/server/replication/server/changelog/je/ChangeNumberIndexer.java

@@ -31,6 +31,7 @@

import org.opends.messages.Message;
import org.opends.server.api.DirectoryThread;
import org.opends.server.backends.ChangelogBackend;
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.MultiDomainServerState;
@@ -491,9 +492,9 @@
          // OK, the oldest change is older than the medium consistency point
          // let's publish it to the CNIndexDB.
          final String previousCookie = mediumConsistencyRUV.toString();
          final ChangeNumberIndexRecord record =
              new ChangeNumberIndexRecord(previousCookie, baseDN, csn);
          changelogDB.getChangeNumberIndexDB().addRecord(record);
          final long changeNumber = changelogDB.getChangeNumberIndexDB().addRecord(
              new ChangeNumberIndexRecord(previousCookie, baseDN, csn));
          notifyEntryAddedToChangelog(baseDN, changeNumber, previousCookie, msg);
          moveForwardMediumConsistencyPoint(csn, baseDN);
        }
        catch (InterruptedException ignored)
@@ -523,6 +524,29 @@
  }

  /**
   * Notifies the {@link ChangelogBackend} that a new entry has been added.
   *
   * @param baseDN
   *          the baseDN of the newly added entry.
   * @param changeNumber
   *          the change number of the newly added entry. It will be greater
   *          than zero for entries added to the change number index and less
   *          than or equal to zero for entries added to any replica DB
   * @param cookieString
   *          a string representing the cookie of the newly added entry. This is
   *          only meaningful for entries added to the change number index
   * @param msg
   *          the update message of the newly added entry
   * @throws ChangelogException
   *           If a problem occurs while notifying of the newly added entry.
   */
  protected void notifyEntryAddedToChangelog(DN baseDN, long changeNumber,
      String cookieString, UpdateMsg msg) throws ChangelogException
  {
    ChangelogBackend.getInstance().notifyEntryAdded(baseDN, changeNumber, cookieString, msg);
  }

  /**
   * Nothing can be done about it.
   * <p>
   * Rely on the DirectoryThread uncaught exceptions handler for logging error +

 opendj-sdk/opends/src/server/org/opends/server/replication/server/changelog/je/JEChangelogDB.java

@@ -37,6 +37,7 @@
import org.opends.messages.MessageBuilder;
import org.opends.server.admin.std.server.ReplicationServerCfg;
import org.opends.server.api.DirectoryThread;
import org.opends.server.backends.ChangelogBackend;
import org.opends.server.config.ConfigException;
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.replication.common.CSN;
@@ -846,6 +847,8 @@
    final JEReplicaDB replicaDB = pair.getFirst();
    replicaDB.add(updateMsg);

    ChangelogBackend.getInstance().notifyEntryAdded(baseDN, 0, null, updateMsg);

    final ChangeNumberIndexer indexer = cnIndexer.get();
    if (indexer != null)
    {

 opendj-sdk/opends/src/server/org/opends/server/workflowelement/externalchangelog/ECLSearchOperation.java

@@ -34,6 +34,7 @@
import org.opends.messages.Severity;
import org.opends.server.api.ClientConnection;
import org.opends.server.api.plugin.PluginResult;
import org.opends.server.backends.ChangelogBackend;
import org.opends.server.config.ConfigConstants;
import org.opends.server.controls.*;
import org.opends.server.core.*;
@@ -54,13 +55,12 @@

import static org.opends.messages.CoreMessages.*;
import static org.opends.messages.ReplicationMessages.*;
import static org.opends.server.backends.ChangelogBackend.*;
import static org.opends.server.config.ConfigConstants.*;
import static org.opends.server.loggers.ErrorLogger.*;
import static org.opends.server.loggers.debug.DebugLogger.*;
import static org.opends.server.replication.protocol.StartECLSessionMsg
.ECLRequestType.*;
import static org.opends.server.replication.protocol.StartECLSessionMsg
.Persistent.*;
import static org.opends.server.replication.protocol.StartECLSessionMsg.ECLRequestType.*;
import static org.opends.server.replication.protocol.StartECLSessionMsg.Persistent.*;
import static org.opends.server.util.LDIFWriter.*;
import static org.opends.server.util.ServerConstants.*;
import static org.opends.server.util.StaticUtils.*;
@@ -119,22 +119,6 @@
  private static final AttributeType MODIFIERS_NAME_TYPE =
      DirectoryConfig.getAttributeType(OP_ATTR_MODIFIERS_NAME_LC, true);


  /** The associated DN. */
  private static final DN CHANGELOG_ROOT_DN;
  static
  {
    try
    {
      CHANGELOG_ROOT_DN = DN
          .decode(ServerConstants.DN_EXTERNAL_CHANGELOG_ROOT);
    }
    catch (Exception e)
    {
      throw new RuntimeException(e);
    }
  }

  /**
   * The replication server in which the search on ECL is to be performed.
   */
@@ -300,7 +284,10 @@
      // If there's a persistent search, then register it with the server.
      if (persistentSearch != null)
      {
        wfe.registerPersistentSearch(persistentSearch);
        ChangelogBackend.getInstance().registerPersistentSearch(persistentSearch);
        // TODO JNR Add callback on cancel,
        // see ECLWorkflowElement.registerPersistentSearch().
        // This will be removed very soon anyway.
        persistentSearch.enable();
      }

@@ -529,6 +516,7 @@

          persistentSearch = new PersistentSearch(this,
              psearchControl.getChangeTypes(),
              psearchControl.getChangesOnly(),
              psearchControl.getReturnECs());

          // If we're only interested in changes, then we don't actually want
@@ -607,7 +595,7 @@
      ECLUpdateMsg update = eclServerHandler.getNextECLUpdate();

      // Return root entry if requested.
      if (CHANGELOG_ROOT_DN.matchesBaseAndScope(baseDN, getScope()))
      if (CHANGELOG_BASE_DN.matchesBaseAndScope(baseDN, getScope()))
      {
        final Entry entry = createRootEntry(update != null);
        if (filter.matchesEntry(entry) && !returnEntry(entry, null))
@@ -618,7 +606,7 @@
        }
      }

      if (baseDN.equals(CHANGELOG_ROOT_DN)
      if (baseDN.equals(CHANGELOG_BASE_DN)
          && getScope().equals(SearchScope.BASE_OBJECT))
      {
        // Only the change log root entry was requested. There is no need to
@@ -924,9 +912,9 @@
    addAttributeByUppercaseName("hassubordinates", "hasSubordinates",
        Boolean.toString(hasSubordinates), userAttrs, operationalAttrs);
    addAttributeByUppercaseName("entrydn", "entryDN",
        CHANGELOG_ROOT_DN.toNormalizedString(), userAttrs, operationalAttrs);
        DN_EXTERNAL_CHANGELOG_ROOT, userAttrs, operationalAttrs);

    return new Entry(CHANGELOG_ROOT_DN, CHANGELOG_ROOT_OBJECT_CLASSES,
    return new Entry(CHANGELOG_BASE_DN, CHANGELOG_ROOT_OBJECT_CLASSES,
        userAttrs, operationalAttrs);
  }


 opendj-sdk/opends/src/server/org/opends/server/workflowelement/externalchangelog/ECLWorkflowElement.java

@@ -22,49 +22,28 @@
 *
 *
 *      Copyright 2009 Sun Microsystems, Inc.
 *      Portions Copyright 2012 ForgeRock AS
 *      Portions Copyright 2012-2014 ForgeRock AS
 */
package org.opends.server.workflowelement.externalchangelog;



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

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

import org.opends.server.admin.std.server.WorkflowElementCfg;
import org.opends.server.core.DirectoryServer;
import org.opends.server.core.PersistentSearch;
import org.opends.server.core.SearchOperation;
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.replication.server.ReplicationServer;
import org.opends.server.types.CanceledOperationException;
import org.opends.server.types.DirectoryException;
import org.opends.server.types.Operation;
import org.opends.server.workflowelement.LeafWorkflowElement;




/**
 * This class defines a workflow element for the external changelog (ECL);
 * e-g an entity that handles the processing of an operation against the ECL.
 */
public class ECLWorkflowElement extends
    LeafWorkflowElement<WorkflowElementCfg>
public class ECLWorkflowElement extends LeafWorkflowElement<WorkflowElementCfg>
{
  /**
   * The tracer object for the debug logger.
   */
  private static final DebugTracer TRACER = getTracer();

  /**
   *The set of persistent searches registered with this work flow element.
   */
  private final List<PersistentSearch> persistentSearches =
    new CopyOnWriteArrayList<PersistentSearch>();

  /**
   * A string indicating the type of the workflow element.
@@ -75,7 +54,7 @@
   * The replication server object to which we will submits request
   * on the ECL. Retrieved from the local DirectoryServer.
   */
  private ReplicationServer replicationServer;
  private final ReplicationServer replicationServer;

  /**
   * Creates a new instance of the External Change Log workflow element.
@@ -91,26 +70,16 @@
    DirectoryServer.registerWorkflowElement(this);
  }

  /**
   * {@inheritDoc}
   */
  /** {@inheritDoc} */
  @Override
  public void finalizeWorkflowElement()
  {
    // null all fields so that any use of the finalized object will raise
    // an NPE
    // null all fields so that any use of the finalized object will raise a NPE
    super.initialize(ECL_WORKFLOW_ELEMENT, null);

    // Cancel all persistent searches.
    for (PersistentSearch psearch : persistentSearches) {
      psearch.cancel();
    }
    persistentSearches.clear();
  }

  /**
   * {@inheritDoc}
   */
  /** {@inheritDoc} */
  @Override
  public void execute(Operation operation) throws CanceledOperationException {
    switch (operation.getOperationType())
    {
@@ -171,45 +140,6 @@
  }

  /**
   * Registers the provided persistent search operation with this
   * workflow element so that it will be notified of any
   * add, delete, modify, or modify DN operations that are performed.
   *
   * @param persistentSearch
   *          The persistent search operation to register with this
   *          workflow element.
   */
  void registerPersistentSearch(PersistentSearch persistentSearch)
  {
    PersistentSearch.CancellationCallback callback =
      new PersistentSearch.CancellationCallback()
    {
      public void persistentSearchCancelled(PersistentSearch psearch)
      {
        psearch.getSearchOperation().cancel(null);
        persistentSearches.remove(psearch);
      }
    };

    persistentSearches.add(persistentSearch);
    persistentSearch.registerCancellationCallback(callback);
  }



  /**
   * Gets the list of persistent searches currently active against
   * this workflow element.
   *
   * @return The list of persistent searches currently active against
   *         this workflow element.
   */
  public List<PersistentSearch> getPersistentSearches()
  {
    return persistentSearches;
  }

  /**
   * Returns the associated replication server.
   * @return the rs.
   */

 opendj-sdk/opends/src/server/org/opends/server/workflowelement/localbackend/LocalBackendAddOperation.java

@@ -184,7 +184,7 @@
        @Override
        public void run()
        {
          for (PersistentSearch psearch : wfe.getPersistentSearches())
          for (PersistentSearch psearch : backend.getPersistentSearches())
          {
            psearch.processAdd(entry);
          }

 opendj-sdk/opends/src/server/org/opends/server/workflowelement/localbackend/LocalBackendDeleteOperation.java

@@ -175,7 +175,7 @@
        @Override
        public void run()
        {
          for (PersistentSearch psearch : wfe.getPersistentSearches())
          for (PersistentSearch psearch : backend.getPersistentSearches())
          {
            psearch.processDelete(entry);
          }

 opendj-sdk/opends/src/server/org/opends/server/workflowelement/localbackend/LocalBackendModifyDNOperation.java

@@ -203,7 +203,7 @@
        @Override
        public void run()
        {
          for (PersistentSearch psearch : wfe.getPersistentSearches())
          for (PersistentSearch psearch : backend.getPersistentSearches())
          {
            psearch.processModifyDN(newEntry, currentEntry.getDN());
          }

 opendj-sdk/opends/src/server/org/opends/server/workflowelement/localbackend/LocalBackendModifyOperation.java

@@ -317,7 +317,7 @@
        @Override
        public void run()
        {
          for (PersistentSearch psearch : wfe.getPersistentSearches())
          for (PersistentSearch psearch : backend.getPersistentSearches())
          {
            psearch.processModify(modifiedEntry, currentEntry);
          }
@@ -637,7 +637,7 @@
        Control c = iter.next();
        String  oid = c.getOID();

        if (oid.equals(OID_LDAP_ASSERTION))
        if (OID_LDAP_ASSERTION.equals(oid))
        {
          LDAPAssertionRequestControl assertControl =
                getRequestControl(LDAPAssertionRequestControl.DECODER);
@@ -697,19 +697,19 @@
                                de.getMessageObject()));
          }
        }
        else if (oid.equals(OID_LDAP_NOOP_OPENLDAP_ASSIGNED))
        else if (OID_LDAP_NOOP_OPENLDAP_ASSIGNED.equals(oid))
        {
          noOp = true;
        }
        else if (oid.equals(OID_PERMISSIVE_MODIFY_CONTROL))
        else if (OID_PERMISSIVE_MODIFY_CONTROL.equals(oid))
        {
          permissiveModify = true;
        }
        else if (oid.equals(OID_LDAP_READENTRY_PREREAD))
        else if (OID_LDAP_READENTRY_PREREAD.equals(oid))
        {
          preReadRequest = getRequestControl(LDAPPreReadRequestControl.DECODER);
        }
        else if (oid.equals(OID_LDAP_READENTRY_POSTREAD))
        else if (OID_LDAP_READENTRY_POSTREAD.equals(oid))
        {
          if (c instanceof LDAPPostReadRequestControl)
          {
@@ -721,7 +721,7 @@
            iter.set(postReadRequest);
          }
        }
        else if (oid.equals(OID_PROXIED_AUTH_V1))
        else if (OID_PROXIED_AUTH_V1.equals(oid))
        {
          // Log usage of legacy proxy authz V1 control.
          addAdditionalLogItem(AdditionalLogItem.keyOnly(getClass(),
@@ -742,7 +742,7 @@
          setAuthorizationEntry(authorizationEntry);
          setProxiedAuthorizationDN(getDN(authorizationEntry));
        }
        else if (oid.equals(OID_PROXIED_AUTH_V2))
        else if (OID_PROXIED_AUTH_V2.equals(oid))
        {
          // The requester must have the PROXIED_AUTH privilege in order to
          // be able to use this control.
@@ -759,7 +759,7 @@
          setAuthorizationEntry(authorizationEntry);
          setProxiedAuthorizationDN(getDN(authorizationEntry));
        }
        else if (oid.equals(OID_PASSWORD_POLICY_CONTROL))
        else if (OID_PASSWORD_POLICY_CONTROL.equals(oid))
        {
          pwPolicyControlRequested = true;
        }
@@ -825,13 +825,11 @@
      // See if the attribute is one which controls the privileges available for
      // a user.  If it is, then the client must have the PRIVILEGE_CHANGE
      // privilege.
      if (t.hasName(OP_ATTR_PRIVILEGE_NAME))
      if (t.hasName(OP_ATTR_PRIVILEGE_NAME)
          && !clientConnection.hasPrivilege(Privilege.PRIVILEGE_CHANGE, this))
      {
        if (! clientConnection.hasPrivilege(Privilege.PRIVILEGE_CHANGE, this))
        {
          throw new DirectoryException(ResultCode.INSUFFICIENT_ACCESS_RIGHTS,
                  ERR_MODIFY_CHANGE_PRIVILEGE_INSUFFICIENT_PRIVILEGES.get());
        }
        throw new DirectoryException(ResultCode.INSUFFICIENT_ACCESS_RIGHTS,
                ERR_MODIFY_CHANGE_PRIVILEGE_INSUFFICIENT_PRIVILEGES.get());
      }

      // If the modification is not updating the password attribute,
@@ -1065,11 +1063,11 @@
      numPasswords = passwordsToAdd;
    }

    // If there were multiple password values, then make sure that's
    // OK.
    if ((!isInternalOperation())
        && (!pwPolicyState.getAuthenticationPolicy()
            .isAllowMultiplePasswordValues()) && (passwordsToAdd > 1))
    // If there were multiple password values, then make sure that's OK.
    final PasswordPolicy authPolicy = pwPolicyState.getAuthenticationPolicy();
    if (!isInternalOperation()
        && !authPolicy.isAllowMultiplePasswordValues()
        && passwordsToAdd > 1)
    {
      pwpErrorType = PasswordPolicyErrorType.PASSWORD_MOD_NOT_ALLOWED;
      throw new DirectoryException(ResultCode.CONSTRAINT_VIOLATION,
@@ -1085,9 +1083,8 @@
    {
      if (pwPolicyState.passwordIsPreEncoded(v.getValue()))
      {
        if ((!isInternalOperation())
            && !pwPolicyState.getAuthenticationPolicy()
                .isAllowPreEncodedPasswords())
        if (!isInternalOperation()
            && !authPolicy.isAllowPreEncodedPasswords())
        {
          pwpErrorType = PasswordPolicyErrorType.INSUFFICIENT_PASSWORD_QUALITY;
          throw new DirectoryException(ResultCode.CONSTRAINT_VIOLATION,
@@ -1100,15 +1097,13 @@
      }
      else
      {
        if (m.getModificationType() == ModificationType.ADD)
        if (m.getModificationType() == ModificationType.ADD
            // Make sure that the password value does not already exist.
            && pwPolicyState.passwordMatches(v.getValue()))
        {
          // Make sure that the password value doesn't already exist.
          if (pwPolicyState.passwordMatches(v.getValue()))
          {
            pwpErrorType = PasswordPolicyErrorType.PASSWORD_IN_HISTORY;
            throw new DirectoryException(ResultCode.ATTRIBUTE_OR_VALUE_EXISTS,
                ERR_MODIFY_PASSWORD_EXISTS.get());
          }
          pwpErrorType = PasswordPolicyErrorType.PASSWORD_IN_HISTORY;
          throw new DirectoryException(ResultCode.ATTRIBUTE_OR_VALUE_EXISTS,
              ERR_MODIFY_PASSWORD_EXISTS.get());
        }

        if (newPasswords == null)
@@ -1196,7 +1191,7 @@
      else
      {
        List<Attribute> attrList = currentEntry.getAttribute(pwAttr.getAttributeType());
        if ((attrList == null) || (attrList.isEmpty()))
        if (attrList == null || attrList.isEmpty())
        {
          throw new DirectoryException(ResultCode.NO_SUCH_ATTRIBUTE,
              ERR_MODIFY_NO_EXISTING_VALUES.get());
@@ -1214,51 +1209,37 @@
                    .decodeAuthPassword(av.getValue().toString());
                PasswordStorageScheme<?> scheme = DirectoryServer
                    .getAuthPasswordStorageScheme(components[0].toString());
                if (scheme != null)
                if (scheme != null
                    && scheme.authPasswordMatches(v.getValue(), components[1]
                    .toString(), components[2].toString()))
                {
                  if (scheme.authPasswordMatches(v.getValue(), components[1]
                      .toString(), components[2].toString()))
                  {
                    builder.add(av);
                    found = true;
                  }
                }
              }
              else
              {
                if (av.equals(v))
                {
                  builder.add(v);
                  builder.add(av);
                  found = true;
                }
              }
              else if (av.equals(v))
              {
                builder.add(v);
                found = true;
              }
            }
            else
            else if (UserPasswordSyntax.isEncoded(av.getValue()))
            {
              if (UserPasswordSyntax.isEncoded(av.getValue()))
              String[] components = UserPasswordSyntax.decodeUserPassword(av
                  .getValue().toString());
              PasswordStorageScheme<?> scheme = DirectoryServer
                  .getPasswordStorageScheme(toLowerCase(components[0]));
              if (scheme != null
                  && scheme.passwordMatches(v.getValue(), ByteString.valueOf(components[1])))
              {
                String[] components = UserPasswordSyntax.decodeUserPassword(av
                    .getValue().toString());
                PasswordStorageScheme<?> scheme = DirectoryServer
                    .getPasswordStorageScheme(toLowerCase(components[0]));
                if (scheme != null)
                {
                  if (scheme.passwordMatches(v.getValue(), ByteString.valueOf(
                      components[1])))
                  {
                    builder.add(av);
                    found = true;
                  }
                }
                builder.add(av);
                found = true;
              }
              else
              {
                if (av.equals(v))
                {
                  builder.add(v);
                  found = true;
                }
              }
            }
            else if (av.equals(v))
            {
              builder.add(v);
              found = true;
            }
          }
        }
@@ -1425,7 +1406,7 @@
          TRACER.debugCaught(DebugLogLevel.ERROR, e);
        }

        lowerName = toLowerCase(v.getValue().toString());
        lowerName = toLowerCase(name);
      }

      ObjectClass oc = DirectoryServer.getObjectClass(lowerName);
@@ -1669,11 +1650,11 @@
    AttributeBuilder builder = new AttributeBuilder(a, true);
    for (AttributeValue existingValue : a)
    {
      String s = existingValue.getValue().toString();
      final String value = existingValue.getValue().toString();
      long currentValue;
      try
      {
        currentValue = Long.parseLong(s);
        currentValue = Long.parseLong(value);
      }
      catch (Exception e)
      {
@@ -1684,9 +1665,8 @@

        throw new DirectoryException(
            ResultCode.INVALID_ATTRIBUTE_SYNTAX,
            ERR_MODIFY_INCREMENT_REQUIRES_INTEGER_VALUE.get(String
                .valueOf(entryDN), a.getName(),
                existingValue.getValue().toString()),
            ERR_MODIFY_INCREMENT_REQUIRES_INTEGER_VALUE.get(
                String.valueOf(entryDN), a.getName(), value),
            e);
      }

@@ -1711,13 +1691,8 @@
  public void performAdditionalPasswordChangedProcessing()
         throws DirectoryException
  {
    if (pwPolicyState == null)
    {
      // Account not managed locally so nothing to do.
      return;
    }

    if (!passwordChanged)
    if (!passwordChanged
        || pwPolicyState == null) // Account not managed locally
    {
      // Nothing to do.
      return;
@@ -1748,85 +1723,63 @@


    // If any of the password values should be validated, then do so now.
    if (selfChange || !authPolicy.isSkipValidationForAdministrators())
    if (newPasswords != null
        && (selfChange || !authPolicy.isSkipValidationForAdministrators()))
    {
      if (newPasswords != null)
      HashSet<ByteString> clearPasswords = new HashSet<ByteString>(pwPolicyState.getClearPasswords());
      if (currentPasswords != null)
      {
        HashSet<ByteString> clearPasswords = new HashSet<ByteString>();
        clearPasswords.addAll(pwPolicyState.getClearPasswords());

        if (currentPasswords != null)
        if (clearPasswords.isEmpty())
        {
          if (clearPasswords.isEmpty())
          for (AttributeValue v : currentPasswords)
          {
            for (AttributeValue v : currentPasswords)
            {
              clearPasswords.add(v.getValue());
            }
          }
          else
          {
            // NOTE:  We can't rely on the fact that Set doesn't allow
            // duplicates because technically it's possible that the values
            // aren't duplicates if they are ASN.1 elements with different types
            // (like 0x04 for a standard universal octet string type versus 0x80
            // for a simple password in a bind operation).  So we have to
            // manually check for duplicates.
            for (AttributeValue v : currentPasswords)
            {
              ByteString pw = v.getValue();

              boolean found = false;
              for (ByteString s : clearPasswords)
              {
                if (s.equals(pw))
                {
                  found = true;
                  break;
                }
              }

              if (! found)
              {
                clearPasswords.add(pw);
              }
            }
            clearPasswords.add(v.getValue());
          }
        }

        for (AttributeValue v : newPasswords)
        else
        {
          MessageBuilder invalidReason = new MessageBuilder();
          if (! pwPolicyState.passwordIsAcceptable(this, modifiedEntry,
                                   v.getValue(), clearPasswords, invalidReason))
          // NOTE:  We can't rely on the fact that Set doesn't allow
          // duplicates because technically it's possible that the values
          // aren't duplicates if they are ASN.1 elements with different types
          // (like 0x04 for a standard universal octet string type versus 0x80
          // for a simple password in a bind operation).  So we have to
          // manually check for duplicates.
          for (AttributeValue v : currentPasswords)
          {
            pwpErrorType = PasswordPolicyErrorType.INSUFFICIENT_PASSWORD_QUALITY;
            throw new DirectoryException(ResultCode.CONSTRAINT_VIOLATION,
                ERR_MODIFY_PW_VALIDATION_FAILED.get(invalidReason));
            clearPasswords.add(v.getValue());
          }
        }
      }

      for (AttributeValue v : newPasswords)
      {
        MessageBuilder invalidReason = new MessageBuilder();
        if (! pwPolicyState.passwordIsAcceptable(this, modifiedEntry,
                                 v.getValue(), clearPasswords, invalidReason))
        {
          pwpErrorType = PasswordPolicyErrorType.INSUFFICIENT_PASSWORD_QUALITY;
          throw new DirectoryException(ResultCode.CONSTRAINT_VIOLATION,
              ERR_MODIFY_PW_VALIDATION_FAILED.get(invalidReason));
        }
      }
    }


    // If we should check the password history, then do so now.
    if (pwPolicyState.maintainHistory())
    if (newPasswords != null && pwPolicyState.maintainHistory())
    {
      if (newPasswords != null)
      for (AttributeValue v : newPasswords)
      {
        for (AttributeValue v : newPasswords)
        if (pwPolicyState.isPasswordInHistory(v.getValue())
            && (selfChange || !authPolicy.isSkipValidationForAdministrators()))
        {
          if (pwPolicyState.isPasswordInHistory(v.getValue())
              && (selfChange || !authPolicy.isSkipValidationForAdministrators()))
          {
            pwpErrorType = PasswordPolicyErrorType.PASSWORD_IN_HISTORY;
            throw new DirectoryException(ResultCode.CONSTRAINT_VIOLATION,
                ERR_MODIFY_PW_IN_HISTORY.get());
          }
          pwpErrorType = PasswordPolicyErrorType.PASSWORD_IN_HISTORY;
          throw new DirectoryException(ResultCode.CONSTRAINT_VIOLATION,
              ERR_MODIFY_PW_IN_HISTORY.get());
        }

        pwPolicyState.updatePasswordHistory();
      }

      pwPolicyState.updatePasswordHistory();
    }


@@ -1882,7 +1835,7 @@
      return;
    }

    if (!(passwordChanged || enabledStateChanged || wasLocked))
    if (!passwordChanged && !enabledStateChanged && !wasLocked)
    {
      // Account managed locally, but unchanged, so nothing to do.
      return;

 opendj-sdk/opends/src/server/org/opends/server/workflowelement/localbackend/LocalBackendSearchOperation.java

@@ -54,46 +54,24 @@
       implements PreOperationSearchOperation, PostOperationSearchOperation,
                  SearchEntrySearchOperation, SearchReferenceSearchOperation
{
  /**
   * The tracer object for the debug logger.
   */
  /** The tracer object for the debug logger. */
  private static final DebugTracer TRACER = getTracer();

  /** The backend in which the search is to be performed. */
  private Backend<?> backend;


  /**
   * The backend in which the search is to be performed.
   */
  private Backend backend;

  /**
   * Indicates whether we should actually process the search.  This should
   * only be false if it's a persistent search with changesOnly=true.
   */
  private boolean processSearch;

  /**
   * The client connection for the search operation.
   */
  /** The client connection for the search operation. */
  private ClientConnection clientConnection;

  /**
   * The base DN for the search.
   */
  /** The base DN for the search. */
  private DN baseDN;

  /**
   * The persistent search request, if applicable.
   */
  /** The persistent search request, if applicable. */
  private PersistentSearch persistentSearch;

  /**
   * The filter for the search.
   */
  /** The filter for the search. */
  private SearchFilter filter;



  /**
   * Creates a new operation that may be used to search for entries in a local
   * backend of the Directory Server.
@@ -120,10 +98,7 @@
      throws CanceledOperationException
  {
    this.backend = wfe.getBackend();

    clientConnection = getClientConnection();

    processSearch = true;
    this.clientConnection = getClientConnection();

    // Check for a request to cancel this operation.
    checkIfCanceled(false);
@@ -131,7 +106,7 @@
    try
    {
      BooleanHolder executePostOpPlugins = new BooleanHolder(false);
      processSearch(wfe, executePostOpPlugins);
      processSearch(executePostOpPlugins);

      // Check for a request to cancel this operation.
      checkIfCanceled(false);
@@ -157,8 +132,7 @@
    }
  }

  private void processSearch(LocalBackendWorkflowElement wfe,
      BooleanHolder executePostOpPlugins) throws CanceledOperationException
  private void processSearch(BooleanHolder executePostOpPlugins) throws CanceledOperationException
  {
    // Process the search base and filter to convert them from their raw forms
    // as provided by the client to the forms required for the rest of the
@@ -166,7 +140,7 @@
    baseDN = getBaseDN();
    filter = getFilter();

    if ((baseDN == null) || (filter == null))
    if (baseDN == null || filter == null)
    {
      return;
    }
@@ -253,8 +227,13 @@


    // If there's a persistent search, then register it with the server.
    boolean processSearchNow = true;
    if (persistentSearch != null)
    {
      // If we're only interested in changes, then we do not actually want
      // to process the search now.
      processSearchNow = !persistentSearch.isChangesOnly();

      // The Core server maintains the count of concurrent persistent searches
      // so that all the backends (Remote and Local) are aware of it. Verify
      // with the core if we have already reached the threshold.
@@ -264,7 +243,7 @@
        appendErrorMessage(ERR_MAX_PSEARCH_LIMIT_EXCEEDED.get());
        return;
      }
      wfe.registerPersistentSearch(persistentSearch);
      backend.registerPersistentSearch(persistentSearch);
      persistentSearch.enable();
    }

@@ -272,7 +251,7 @@
    // Process the search in the backend and all its subordinates.
    try
    {
      if (processSearch)
      if (processSearchNow)
      {
        backend.search(this);
      }
@@ -335,14 +314,13 @@
    LocalBackendWorkflowElement.removeAllDisallowedControls(baseDN, this);

    List<Control> requestControls  = getRequestControls();
    if ((requestControls != null) && (! requestControls.isEmpty()))
    if (requestControls != null && ! requestControls.isEmpty())
    {
      for (int i=0; i < requestControls.size(); i++)
      for (Control c : requestControls)
      {
        Control c   = requestControls.get(i);
        String  oid = c.getOID();

        if (oid.equals(OID_LDAP_ASSERTION))
        if (OID_LDAP_ASSERTION.equals(oid))
        {
          LDAPAssertionRequestControl assertControl =
                getRequestControl(LDAPAssertionRequestControl.DECODER);
@@ -421,7 +399,7 @@
                                de.getMessageObject()), de);
          }
        }
        else if (oid.equals(OID_PROXIED_AUTH_V1))
        else if (OID_PROXIED_AUTH_V1.equals(oid))
        {
          // Log usage of legacy proxy authz V1 control.
          addAdditionalLogItem(AdditionalLogItem.keyOnly(getClass(),
@@ -440,16 +418,9 @@

          Entry authorizationEntry = proxyControl.getAuthorizationEntry();
          setAuthorizationEntry(authorizationEntry);
          if (authorizationEntry == null)
          {
            setProxiedAuthorizationDN(DN.nullDN());
          }
          else
          {
            setProxiedAuthorizationDN(authorizationEntry.getDN());
          }
          setProxiedAuthorizationDN(getDN(authorizationEntry));
        }
        else if (oid.equals(OID_PROXIED_AUTH_V2))
        else if (OID_PROXIED_AUTH_V2.equals(oid))
        {
          // The requester must have the PROXIED_AUTH privilege in order to be
          // able to use this control.
@@ -464,38 +435,23 @@

          Entry authorizationEntry = proxyControl.getAuthorizationEntry();
          setAuthorizationEntry(authorizationEntry);
          if (authorizationEntry == null)
          {
            setProxiedAuthorizationDN(DN.nullDN());
          }
          else
          {
            setProxiedAuthorizationDN(authorizationEntry.getDN());
          }
          setProxiedAuthorizationDN(getDN(authorizationEntry));
        }
        else if (oid.equals(OID_PERSISTENT_SEARCH))
        else if (OID_PERSISTENT_SEARCH.equals(oid))
        {
          PersistentSearchControl psearchControl =
            getRequestControl(PersistentSearchControl.DECODER);
          final PersistentSearchControl ctrl =
              getRequestControl(PersistentSearchControl.DECODER);

          persistentSearch = new PersistentSearch(this,
                                      psearchControl.getChangeTypes(),
                                      psearchControl.getReturnECs());

          // If we're only interested in changes, then we don't actually want
          // to process the search now.
          if (psearchControl.getChangesOnly())
          {
            processSearch = false;
          }
              ctrl.getChangeTypes(), ctrl.getChangesOnly(), ctrl.getReturnECs());
        }
        else if (oid.equals(OID_LDAP_SUBENTRIES))
        else if (OID_LDAP_SUBENTRIES.equals(oid))
        {
          SubentriesControl subentriesControl =
                  getRequestControl(SubentriesControl.DECODER);
          setReturnSubentriesOnly(subentriesControl.getVisibility());
        }
        else if (oid.equals(OID_LDUP_SUBENTRIES))
        else if (OID_LDUP_SUBENTRIES.equals(oid))
        {
          // Support for legacy draft-ietf-ldup-subentry.
          addAdditionalLogItem(AdditionalLogItem.keyOnly(getClass(),
@@ -503,25 +459,25 @@

          setReturnSubentriesOnly(true);
        }
        else if (oid.equals(OID_MATCHED_VALUES))
        else if (OID_MATCHED_VALUES.equals(oid))
        {
          MatchedValuesControl matchedValuesControl =
                getRequestControl(MatchedValuesControl.DECODER);
          setMatchedValuesControl(matchedValuesControl);
        }
        else if (oid.equals(OID_ACCOUNT_USABLE_CONTROL))
        else if (OID_ACCOUNT_USABLE_CONTROL.equals(oid))
        {
          setIncludeUsableControl(true);
        }
        else if (oid.equals(OID_REAL_ATTRS_ONLY))
        else if (OID_REAL_ATTRS_ONLY.equals(oid))
        {
          setRealAttributesOnly(true);
        }
        else if (oid.equals(OID_VIRTUAL_ATTRS_ONLY))
        else if (OID_VIRTUAL_ATTRS_ONLY.equals(oid))
        {
          setVirtualAttributesOnly(true);
        }
        else if (oid.equals(OID_GET_EFFECTIVE_RIGHTS) &&
        else if (OID_GET_EFFECTIVE_RIGHTS.equals(oid) &&
          DirectoryServer.isSupportedControl(OID_GET_EFFECTIVE_RIGHTS))
        {
          // Do nothing here and let AciHandler deal with it.
@@ -538,6 +494,11 @@
    }
  }

  private DN getDN(Entry e)
  {
    return e != null ? e.getDN() : DN.nullDN();
  }

  /** Indicates if the backend supports the control corresponding to provided oid. */
  private boolean backendSupportsControl(final String oid)
  {

 opendj-sdk/opends/src/server/org/opends/server/workflowelement/localbackend/LocalBackendWorkflowElement.java

@@ -22,7 +22,7 @@
 *
 *
 *      Copyright 2008-2010 Sun Microsystems, Inc.
 *      Portions Copyright 2011-2013 ForgeRock AS
 *      Portions Copyright 2011-2014 ForgeRock AS
 */
package org.opends.server.workflowelement.localbackend;

@@ -30,7 +30,6 @@
import java.util.Iterator;
import java.util.List;
import java.util.TreeMap;
import java.util.concurrent.CopyOnWriteArrayList;

import org.opends.messages.Message;
import org.opends.messages.MessageDescriptor;
@@ -68,7 +67,7 @@
  private static final DebugTracer TRACER = getTracer();

  /** the backend associated with the local workflow element. */
  private Backend backend;
  private Backend<?> backend;


  /** the set of local backend workflow elements registered with the server. */
@@ -77,13 +76,7 @@
            new TreeMap<String, LocalBackendWorkflowElement>();

  /**
   * The set of persistent searches registered with this work flow element.
   */
  private final List<PersistentSearch> persistentSearches =
    new CopyOnWriteArrayList<PersistentSearch>();

  /**
   * a lock to guarantee safe concurrent access to the registeredLocalBackends
   * A lock to guarantee safe concurrent access to the registeredLocalBackends
   * variable.
   */
  private static final Object registeredLocalBackendsLock = new Object();
@@ -112,9 +105,8 @@
   * @param workflowElementID  the workflow element identifier
   * @param backend  the backend associated to that workflow element
   */
  private void initialize(String workflowElementID, Backend backend)
  private void initialize(String workflowElementID, Backend<?> backend)
  {
    // Initialize the workflow ID
    super.initialize(workflowElementID, BACKEND_WORKFLOW_ELEMENT);

    this.backend  = backend;
@@ -154,29 +146,16 @@
    processWorkflowElementConfig(configuration, true);
  }


  /**
   * {@inheritDoc}
   */
  /** {@inheritDoc} */
  @Override
  public void finalizeWorkflowElement()
  {
    // null all fields so that any use of the finalized object will raise
    // an NPE
    // null all fields so that any use of the finalized object will raise a NPE
    super.initialize(null, null);
    backend = null;

    // Cancel all persistent searches.
    for (PersistentSearch psearch : persistentSearches) {
      psearch.cancel();
    }
    persistentSearches.clear();
  }


  /**
   * {@inheritDoc}
   */
  /** {@inheritDoc} */
  @Override
  public boolean isConfigurationChangeAcceptable(
      LocalBackendWorkflowElementCfg configuration,
@@ -186,10 +165,7 @@
    return processWorkflowElementConfig(configuration, false);
  }


  /**
   * {@inheritDoc}
   */
  /** {@inheritDoc} */
  @Override
  public ConfigChangeResult applyConfigurationChange(
      LocalBackendWorkflowElementCfg configuration
@@ -224,7 +200,7 @@
    {
      // Read configuration.
      String newBackendID = configuration.getBackend();
      Backend newBackend  = DirectoryServer.getBackend(newBackendID);
      Backend<?> newBackend = DirectoryServer.getBackend(newBackendID);

      // If the backend is null (i.e. not found in the list of
      // registered backends, this is probably because we are looking
@@ -273,8 +249,7 @@
   *         element.
   */
  public static LocalBackendWorkflowElement createAndRegister(
      String workflowElementID,
      Backend backend)
      String workflowElementID, Backend<?> backend)
  {
    // If the requested workflow element does not exist then create one.
    LocalBackendWorkflowElement localBackend =
@@ -661,11 +636,7 @@
    }
  }



  /**
   * {@inheritDoc}
   */
  /** {@inheritDoc} */
  @Override
  public void execute(Operation operation) throws CanceledOperationException {
    switch (operation.getOperationType())
@@ -766,54 +737,11 @@
   * @return The backend associated with this local backend workflow
   *         element.
   */
  public Backend getBackend()
  public Backend<?> getBackend()
  {
    return backend;
  }



  /**
   * Registers the provided persistent search operation with this
   * local backend workflow element so that it will be notified of any
   * add, delete, modify, or modify DN operations that are performed.
   *
   * @param persistentSearch
   *          The persistent search operation to register with this
   *          local backend workflow element.
   */
  void registerPersistentSearch(PersistentSearch persistentSearch)
  {
    PersistentSearch.CancellationCallback callback =
      new PersistentSearch.CancellationCallback()
    {
      @Override
      public void persistentSearchCancelled(PersistentSearch psearch)
      {
        persistentSearches.remove(psearch);
      }
    };

    persistentSearches.add(persistentSearch);
    persistentSearch.registerCancellationCallback(callback);
  }



  /**
   * Gets the list of persistent searches currently active against
   * this local backend workflow element.
   *
   * @return The list of persistent searches currently active against
   *         this local backend workflow element.
   */
  List<PersistentSearch> getPersistentSearches()
  {
    return persistentSearches;
  }



  /**
   * Checks if an update operation can be performed against a backend. The
   * operation will be rejected based on the server and backend writability
@@ -834,7 +762,7 @@
   * @throws DirectoryException
   *           If the update operation has been rejected.
   */
  static void checkIfBackendIsWritable(Backend backend, Operation op,
  static void checkIfBackendIsWritable(Backend<?> backend, Operation op,
      DN entryDN, MessageDescriptor.Arg1<CharSequence> serverMsg,
      MessageDescriptor.Arg1<CharSequence> backendMsg)
      throws DirectoryException
@@ -870,5 +798,14 @@
      }
    }
  }
}

  /** {@inheritDoc} */
  @Override
  public String toString()
  {
    return getClass().getSimpleName()
        + " backend=" + backend
        + " workflowElementID=" + getWorkflowElementID()
        + " workflowElementTypeInfo=" + getWorkflowElementTypeInfo();
  }
}

 opendj-sdk/opends/tests/unit-tests-testng/src/server/org/opends/server/backends/ChangelogBackendTestCase.java

@@ -25,16 +25,6 @@
 */
package org.opends.server.backends;

import static org.assertj.core.api.Assertions.*;
import static org.opends.messages.ReplicationMessages.*;
import static org.opends.server.TestCaseUtils.*;
import static org.opends.server.loggers.debug.DebugLogger.*;
import static org.opends.server.replication.protocol.OperationContext.*;
import static org.opends.server.types.ResultCode.*;
import static org.opends.server.util.ServerConstants.*;
import static org.opends.server.util.StaticUtils.*;
import static org.testng.Assert.*;

import java.io.ByteArrayOutputStream;
import java.util.ArrayList;
import java.util.Collections;
@@ -108,6 +98,17 @@

import com.forgerock.opendj.util.Pair;

import static org.assertj.core.api.Assertions.*;
import static org.opends.messages.ReplicationMessages.*;
import static org.opends.server.TestCaseUtils.*;
import static org.opends.server.loggers.debug.DebugLogger.*;
import static org.opends.server.replication.protocol.OperationContext.*;
import static org.opends.server.types.ResultCode.*;
import static org.opends.server.util.CollectionUtils.*;
import static org.opends.server.util.ServerConstants.*;
import static org.opends.server.util.StaticUtils.*;
import static org.testng.Assert.*;

@SuppressWarnings("javadoc")
public class ChangelogBackendTestCase extends ReplicationTestCase
{
@@ -241,7 +242,7 @@
    }
  }

  @Test(enabled=false)
  @Test
  public void searchInCookieModeOnOneSuffixUsingEmptyCookie() throws Exception
  {
    String test = "EmptyCookie";
@@ -268,7 +269,7 @@
    debugInfo(test, "Ending search with success");
  }

  @Test(enabled=false)
  @Test
  public void searchInCookieModeOnOneSuffix() throws Exception
  {
    String test = "CookieOneSuffix";
@@ -305,7 +306,6 @@
    searchOp = searchChangelogUsingCookie("(targetdn=*" + test + "*,o=test)", cookies[3], nbEntries, SUCCESS, test);

    debugInfo(test, "Ending search with success");

  }

  @Test(enabled=false)
@@ -520,7 +520,7 @@
      debugInfo(test, "Ending test successfully");
  }

  @Test(enabled=false)
  @Test
  public void searchInDraftModeWithInvalidChangeNumber() throws Exception
  {
    String testName = "UnknownChangeNumber";
@@ -531,7 +531,7 @@
    debugInfo(testName, "Ending test with success");
  }

  @Test(enabled=false)
  @Test
  public void searchInDraftModeOnOneSuffix() throws Exception
  {
    long firstChangeNumber = 1;
@@ -547,7 +547,7 @@
    debugInfo(testName, "Ending search with success");
  }

  @Test(enabled=false)
  @Test
  public void searchInDraftModeOnOneSuffixMultipleTimes() throws Exception
  {
    replicationServer.getChangelogDB().setPurgeDelay(0);
@@ -582,7 +582,7 @@
  /**
   * Verifies that is not possible to read the changelog without the changelog-read privilege
   */
  @Test(enabled=false)
  @Test
  public void searchingWithoutPrivilegeShouldFail() throws Exception
  {
    AuthenticationInfo nonPrivilegedUser = new AuthenticationInfo();
@@ -594,7 +594,7 @@
    assertEquals(op.getErrorMessage().toMessage(), NOTE_SEARCH_CHANGELOG_INSUFFICIENT_PRIVILEGES.get());
  }

  @Test(enabled=false)
  @Test
  public void persistentSearch() throws Exception
  {
   // TODO
@@ -603,7 +603,7 @@
   // ExternalChangeLogTest#ECLReplicationServerFullTest16
  }

  @Test(enabled=false)
  @Test
  public void simultaneousPersistentSearches() throws Exception
  {
    // TODO
@@ -624,7 +624,7 @@
  /**
   * With an empty RS, a search should return only root entry.
   */
  @Test(enabled=false)
  @Test
  public void searchWhenNoChangesShouldReturnRootEntryOnly() throws Exception
  {
    String testName = "EmptyRS";
@@ -635,7 +635,7 @@
    debugInfo(testName, "Ending test successfully");
  }

  @Test(enabled=false)
  @Test
  public void operationalAndVirtualAttributesShouldNotBeVisibleOutsideRootDSE() throws Exception
  {
    String testName = "attributesVisibleOutsideRootDSE";
@@ -1082,7 +1082,7 @@
  private List<Modification> createAttributeModif(String attributeName, String valueString)
  {
    Attribute attr = Attributes.create(attributeName, valueString);
    return newList(new Modification(ModificationType.REPLACE, attr));
    return newArrayList(new Modification(ModificationType.REPLACE, attr));
  }

  private UpdateMsg generateModDNMsg(String baseDn, CSN csn, String testName) throws Exception

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

@@ -29,6 +29,7 @@
import java.io.BufferedReader;
import java.io.ByteArrayOutputStream;
import java.io.StringReader;
import java.lang.reflect.Method;
import java.net.Socket;
import java.util.*;

@@ -65,7 +66,6 @@
import org.opends.server.util.LDIFWriter;
import org.opends.server.util.TimeThread;
import org.opends.server.workflowelement.externalchangelog.ECLSearchOperation;
import org.opends.server.workflowelement.externalchangelog.ECLWorkflowElement;
import org.opends.server.workflowelement.localbackend.LocalBackendModifyDNOperation;
import org.testng.annotations.AfterClass;
import org.testng.annotations.AfterMethod;
@@ -79,6 +79,7 @@
import static org.opends.server.loggers.debug.DebugLogger.*;
import static org.opends.server.replication.protocol.OperationContext.*;
import static org.opends.server.types.ResultCode.*;
import static org.opends.server.util.CollectionUtils.*;
import static org.opends.server.util.StaticUtils.*;
import static org.testng.Assert.*;

@@ -129,7 +130,7 @@
   * When used in a search operation, it includes all attributes (user and
   * operational)
   */
  private static final Set<String> ALL_ATTRIBUTES = newSet("*", "+");
  private static final Set<String> ALL_ATTRIBUTES = newHashSet("*", "+");
  private static final List<Control> NO_CONTROL = null;

  /**
@@ -172,14 +173,6 @@
  public void PrimaryTest() throws Exception
  {
    replicationServer.getChangelogDB().setPurgeDelay(0);
    // let's enable ECl manually now that we tested that ECl is not available
    ECLWorkflowElement wfe =
        (ECLWorkflowElement) DirectoryServer
        .getWorkflowElement(ECLWorkflowElement.ECL_WORKFLOW_ELEMENT);
    if (wfe != null)
    {
      wfe.getReplicationServer().enableECL();
    }

    // Test all types of ops.
    ECLAllOps(); // Do not clean the db for the next test
@@ -346,7 +339,7 @@
  /**
   * Verifies that is not possible to read the changelog without the changelog-read privilege
   */
  @Test(enabled=true, dependsOnMethods = { "PrimaryTest"})
  @Test(enabled = false, dependsOnMethods = { "PrimaryTest" })
  public void ECLChangelogReadPrivilegeTest() throws Exception
  {
    AuthenticationInfo nonPrivilegedUser = new AuthenticationInfo();
@@ -362,7 +355,22 @@
  @Test(enabled = true)
  public void TestECLIsNotASupportedSuffix() throws Exception
  {
    ECLCompatTestLimits(0,0, false);
    try
    {
      invoke(replicationServer, "shutdownExternalChangelog");
      ECLCompatTestLimits(0, 0, false);
    }
    finally
    {
      invoke(replicationServer, "enableExternalChangeLog");
    }
  }

  private void invoke(Object obj, String methodName) throws Exception
  {
    final Method m = obj.getClass().getDeclaredMethod(methodName);
    m.setAccessible(true);
    m.invoke(obj);
  }

  /**
@@ -555,7 +563,7 @@
    ReplicationBroker server01 = null;
    LDAPReplicationDomain domain = null;
    LDAPReplicationDomain domain2 = null;
    Backend backend2 = null;
    Backend<?> backend2 = null;

    // Use different values than other tests to avoid test interactions in concurrent test runs
    final String backendId2 = tn + 2;
@@ -636,7 +644,7 @@
    ReplicationBroker s2test = null;
    ReplicationBroker s2test2 = null;

    Backend backend2 = null;
    Backend<?> backend2 = null;
    LDAPReplicationDomain domain1 = null;
    LDAPReplicationDomain domain2 = null;
    try
@@ -948,7 +956,7 @@
  }

  /** Test ECL content after a domain has been removed. */
  @Test(enabled=true, dependsOnMethods = { "PrimaryTest"})
  @Test(enabled = false, dependsOnMethods = { "PrimaryTest" })
  public void testECLAfterDomainIsRemoved() throws Exception
  {
    String testName = "testECLAfterDomainIsRemoved";
@@ -1045,9 +1053,8 @@
    String cookie = "";
    LDIFWriter ldifWriter = getLDIFWriter();

    Set<String> lastcookieattribute = newSet("lastExternalChangelogCookie");
    InternalSearchOperation searchOp = searchOnRootDSE(lastcookieattribute);
    List<SearchResultEntry> entries = searchOp.getSearchEntries();
    final Set<String> attrs = newHashSet("lastExternalChangelogCookie");
    List<SearchResultEntry> entries = searchOnRootDSE(attrs).getSearchEntries();
    if (entries != null)
    {
      for (SearchResultEntry resultEntry : entries)
@@ -1149,7 +1156,7 @@
        checkValue(resultEntry, "replicationcsn", csns[i - 1].toString());
        checkValue(resultEntry, "replicaidentifier", String.valueOf(SERVER_ID_1));
        checkValue(resultEntry, "changelogcookie", cookies[i - 1]);
        checkValue(resultEntry, "changenumber", "0");
        assertNull(getAttributeValue(resultEntry, "changenumber"));

        if (i==1)
        {
@@ -1342,8 +1349,7 @@
    return av.toString();
  }

  private static void checkValues(Entry entry, String attrName,
      Set<String> expectedValues)
  private static void checkValues(Entry entry, String attrName, Set<String> expectedValues)
  {
    final Set<String> values = new HashSet<String>();
    for (Attribute a : entry.getAttribute(attrName))
@@ -1931,7 +1937,7 @@
  /**
   * Utility - create a second backend in order to test ECL with 2 suffixes.
   */
  private static Backend initializeTestBackend(boolean createBaseEntry,
  private static Backend<?> initializeTestBackend(boolean createBaseEntry,
      String backendId) throws Exception
  {
    DN baseDN = DN.decode("o=" + backendId);
@@ -1963,9 +1969,9 @@
    return memoryBackend;
  }

  private static void removeTestBackend(Backend... backends)
  private static void removeTestBackend(Backend<?>... backends)
  {
    for (Backend backend : backends)
    for (Backend<?> backend : backends)
    {
      if (backend != null)
      {
@@ -1989,7 +1995,7 @@
    ReplicationBroker s2test = null;
    ReplicationBroker s1test2 = null;
    ReplicationBroker s2test2 = null;
    Backend backend2 = null;
    Backend<?> backend2 = null;

    try
    {
@@ -2434,7 +2440,7 @@
    // available in other entries. We u
    debugInfo(tn, "Starting test \n\n");

    Set<String> attributes = newSet("firstchangenumber", "lastchangenumber",
    Set<String> attributes = newHashSet("firstchangenumber", "lastchangenumber",
        "changelog", "lastExternalChangelogCookie");

    debugInfo(tn, " Search: " + TEST_ROOT_DN_STRING);
@@ -2603,8 +2609,8 @@

    final String backendId3 = "test3";
    final DN baseDN3 = DN.decode("o=" + backendId3);
    Backend backend2 = null;
    Backend backend3 = null;
    Backend<?> backend2 = null;
    Backend<?> backend3 = null;
    LDAPReplicationDomain domain2 = null;
    LDAPReplicationDomain domain3 = null;
    LDAPReplicationDomain domain21 = null;
@@ -2702,7 +2708,7 @@
        {
          Entry targetEntry = parseIncludedAttributes(resultEntry, targetdn);

          Set<String> eoc = newSet("person", "inetOrgPerson", "organizationalPerson", "top");
          Set<String> eoc = newHashSet("person", "inetOrgPerson", "organizationalPerson", "top");
          checkValues(targetEntry, "objectclass", eoc);

          String changeType = getAttributeValue(resultEntry, "changetype");

 opendj-sdk/opends/tests/unit-tests-testng/src/server/org/opends/server/replication/server/changelog/je/ChangeNumberIndexerTest.java

@@ -41,10 +41,7 @@
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.protocol.UpdateMsg;
import org.opends.server.replication.server.ChangelogState;
import org.opends.server.replication.server.changelog.api.ChangeNumberIndexDB;
import org.opends.server.replication.server.changelog.api.ChangeNumberIndexRecord;
import org.opends.server.replication.server.changelog.api.ChangelogDB;
import org.opends.server.replication.server.changelog.api.ReplicationDomainDB;
import org.opends.server.replication.server.changelog.api.*;
import org.opends.server.types.DN;
import org.testng.annotations.*;

@@ -635,7 +632,16 @@
        return eclEnabledDomains.contains(baseDN);
      }
    };
    cnIndexer = new ChangeNumberIndexer(changelogDB, initialState, predicate);
    cnIndexer = new ChangeNumberIndexer(changelogDB, initialState, predicate)
    {
      /** {@inheritDoc} */
      @Override
      protected void notifyEntryAddedToChangelog(DN baseDN, long changeNumber,
          String previousCookie, UpdateMsg msg) throws ChangelogException
      {
        // avoid problems with ChangelogBackend initialization
      }
    };
    cnIndexer.start();
    waitForWaitingState(cnIndexer);
  }