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

Jean-Noel Rouvignac

22.23.2014 25862fdf888ed23207ab51937a43e6f9ad41d805

 opendj3-server-dev/src/guitools/org/opends/guitools/controlpanel/util/ReadOnlyConfigFileHandler.java

@@ -90,6 +90,7 @@
  @Override
  public void finalizeConfigHandler()
  {
    finalizeBackend();
  }

  /** {@inheritDoc} */
@@ -291,12 +292,6 @@

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

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

 opendj3-server-dev/src/server/org/opends/server/api/Backend.java

@@ -29,7 +29,9 @@
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.forgerock.i18n.LocalizableMessage;
import org.forgerock.opendj.config.server.ConfigException;
@@ -40,6 +42,8 @@
import org.opends.server.core.DirectoryServer;
import org.opends.server.core.ModifyDNOperation;
import org.opends.server.core.ModifyOperation;
import org.opends.server.core.PersistentSearch;
import org.opends.server.core.PersistentSearch.CancellationCallback;
import org.opends.server.core.SearchOperation;
import org.opends.server.monitors.BackendMonitor;
import org.opends.server.types.AttributeType;
@@ -99,6 +103,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.
@@ -166,16 +174,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();
  }



@@ -887,7 +905,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.

 opendj3-server-dev/src/server/org/opends/server/backends/BackupBackend.java

@@ -201,6 +201,7 @@
  @Override
  public void finalizeBackend()
  {
    super.finalizeBackend();
    currentConfig.removeBackupChangeListener(this);

    try

 opendj3-server-dev/src/server/org/opends/server/backends/MonitorBackend.java

@@ -64,7 +64,8 @@
{
  private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();

  /** 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;
@@ -331,6 +332,7 @@
  @Override
  public void finalizeBackend()
  {
    super.finalizeBackend();
    currentConfig.removeMonitorChangeListener(this);
    try
    {

 opendj3-server-dev/src/server/org/opends/server/backends/NullBackend.java

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

 opendj3-server-dev/src/server/org/opends/server/backends/RootDSEBackend.java

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


 opendj3-server-dev/src/server/org/opends/server/backends/SchemaBackend.java

@@ -495,6 +495,7 @@
  @Override
  public void finalizeBackend()
  {
    super.finalizeBackend();
    currentConfig.removeSchemaChangeListener(this);

    for (DN baseDN : baseDNs)

 opendj3-server-dev/src/server/org/opends/server/backends/TrustStoreBackend.java

@@ -346,6 +346,7 @@
  @Override
  public void finalizeBackend()
  {
    super.finalizeBackend();
    configuration.addTrustStoreChangeListener(this);

    try

 opendj3-server-dev/src/server/org/opends/server/backends/jeb/BackendImpl.java

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

    // Deregister our base DNs.
@@ -349,10 +349,7 @@
    }

    // 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.

 opendj3-server-dev/src/server/org/opends/server/backends/task/TaskBackend.java

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


    try
    {
      taskScheduler.stopScheduler();

 opendj3-server-dev/src/server/org/opends/server/core/PersistentSearch.java

@@ -107,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.
@@ -152,25 +150,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;
  }

@@ -232,6 +238,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

 opendj3-server-dev/src/server/org/opends/server/extensions/ConfigFileHandler.java

@@ -752,6 +752,7 @@
  @Override
  public void finalizeConfigHandler()
  {
    finalizeBackend();
    try
    {
      DirectoryServer.deregisterBaseDN(configRootEntry.getDN());
@@ -764,13 +765,6 @@

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

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

 opendj3-server-dev/src/server/org/opends/server/replication/plugin/LDAPReplicationDomain.java

@@ -85,7 +85,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.forgerock.opendj.ldap.ResultCode.*;
@@ -478,7 +477,7 @@
    storeECLConfiguration(configuration);
    solveConflictFlag = isSolveConflict(configuration);

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

    //  Acquire a shared lock for the backend.
    try
@@ -3601,7 +3600,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;
@@ -3629,10 +3628,9 @@
  @Override
  protected void importBackend(InputStream input) throws DirectoryException
  {
    Backend<?> backend = getBackend();

    LDIFImportConfig importConfig = null;

    Backend backend = getBackend();

    ImportExportContext ieCtx = getImportExportContext();
    try
    {
@@ -3718,7 +3716,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();
@@ -3785,7 +3783,7 @@
   * Returns the backend associated to this domain.
   * @return The associated backend.
   */
  private Backend getBackend()
  private Backend<?> getBackend()
  {
    return DirectoryServer.getBackend(getBaseDN());
  }
@@ -4071,29 +4069,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)
      {
        logger.info(NOTE_ERR_UNABLE_TO_ENABLE_ECL,
            "Replication Domain on " + getBaseDNString(), stackTraceToSingleLineString(de));
        // and go on
      }
    }

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

 opendj3-server-dev/src/server/org/opends/server/replication/server/ECLServerHandler.java

@@ -34,6 +34,7 @@
import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.i18n.slf4j.LocalizedLogger;
import org.forgerock.opendj.ldap.ResultCode;
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;
@@ -155,8 +156,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;
@@ -354,8 +360,7 @@
    super(session, queueSize, replicationServer, rcvWindowSize);
    try
    {
      DN baseDN = DN.valueOf(ServerConstants.DN_EXTERNAL_CHANGELOG_ROOT);
      setBaseDNAndDomain(baseDN, true);
      setBaseDNAndDomain(ChangelogBackend.CHANGELOG_BASE_DN, true);
    }
    catch(DirectoryException de)
    {
@@ -853,14 +858,6 @@
  }

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

  /**
   * Shutdown this handler.
   */
  @Override
@@ -871,14 +868,20 @@
      logger.trace(this + " shutdown()");
    }
    releaseCursor();
    for (DomainContext domainCtxt : domainCtxts) {
      if (!domainCtxt.unRegisterHandler()) {
        logger.error(LocalizableMessage.raw(this + " shutdown() - error when unregistering handler "+ domainCtxt.mh));
    if (domainCtxts != null)
    {
      for (DomainContext domainCtxt : domainCtxts)
      {
        if (!domainCtxt.unRegisterHandler())
        {
          logger.error(LocalizableMessage.raw(this + " shutdown() - error when unregistering handler " + domainCtxt.mh));
        }
        domainCtxt.stopServer();
      }
      domainCtxt.stopServer();
      domainCtxts = null;
    }

    super.shutdown();
    domainCtxts = null;
  }

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

    registerIntoDomain();
    replicationServerDomain.registerHandler(this);

    if (logger.isTraceEnabled())
    {
      logger.trace(getClass().getCanonicalName() + " " + getOperationId()
      logger.trace(getClass().getSimpleName() + " " + getOperationId()
          + " initialized: " + " " + dumpState() + domaimCtxtsToString(""));
    }
  }
@@ -1372,7 +1375,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)

 opendj3-server-dev/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.forgerock.i18n.slf4j.LocalizedLogger;
import org.opends.server.replication.protocol.DoneMsg;
@@ -39,7 +39,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.util.StaticUtils.*;
@@ -89,9 +88,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()))

 opendj3-server-dev/src/server/org/opends/server/replication/server/ReplicationServer.java

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

import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.i18n.slf4j.LocalizedLogger;
@@ -46,8 +45,6 @@
import org.opends.server.api.VirtualAttributeProvider;
import org.opends.server.backends.ChangelogBackend;
import org.opends.server.core.DirectoryServer;
import org.opends.server.core.WorkflowImpl;
import org.opends.server.core.networkgroups.NetworkGroup;
import org.opends.server.replication.common.*;
import org.opends.server.replication.plugin.MultimasterReplication;
import org.opends.server.replication.protocol.*;
@@ -60,8 +57,6 @@
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.workflowelement.externalchangelog.ECLWorkflowElement;

import static org.opends.messages.ConfigMessages.*;
import static org.opends.messages.ReplicationMessages.*;
@@ -109,12 +104,6 @@
  /** To know whether a domain is enabled for the external changelog. */
  private final ECLEnabledDomainPredicate domainPredicate;

  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.
@@ -173,6 +162,8 @@
    this.config = cfg;
    this.dsrsShutdownSync = dsrsShutdownSync;
    this.domainPredicate = predicate;

    enableExternalChangeLog();
    ReplicationDBImplementation dbImpl = cfg.getReplicationDBImplementation();
    logger.trace("Using %s as DB implementation for changelog DB", dbImpl);
    if (dbImpl == ReplicationDBImplementation.JE)
@@ -188,9 +179,6 @@
    initialize();
    cfg.addChangeListener(this);

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

    localPorts.add(getReplicationPort());

    // Keep track of this new instance
@@ -458,15 +446,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 (logger.isTraceEnabled())
      {
        logger.trace("RS " + getMonitorInstanceName() + " successfully initialized");
@@ -477,50 +456,10 @@
    } catch (IOException e)
    {
      logger.error(ERR_COULD_NOT_BIND_CHANGELOG, getReplicationPort(), e.getMessage());
    } catch (DirectoryException e)
    {
      //FIXME:DirectoryException is raised by initializeECL => fix err msg
      logger.error(LocalizableMessage.raw(
          "Directory Exception raised by ECL initialization: %s", e.getMessage()));
    }
  }

  /**
   * 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.valueOf(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.
@@ -636,34 +575,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.
@@ -901,9 +812,7 @@
      domain.shutdown();
    }

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

    try
    {

 opendj3-server-dev/src/server/org/opends/server/replication/server/changelog/je/ChangeNumberIndexer.java

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

import org.forgerock.i18n.slf4j.LocalizedLogger;
import org.opends.server.api.DirectoryThread;
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;
@@ -524,6 +525,7 @@
  protected void notifyEntryAddedToChangelog(DN baseDN, long changeNumber,
      MultiDomainServerState cookie, UpdateMsg msg) throws ChangelogException
  {
    ChangelogBackend.getInstance().notifyEntryAdded(baseDN, changeNumber, cookie.toString(), msg);
  }

  /**

 opendj3-server-dev/src/server/org/opends/server/replication/server/changelog/je/JEChangelogDB.java

@@ -39,6 +39,7 @@
import org.forgerock.opendj.config.server.ConfigException;
import org.opends.server.admin.std.server.ReplicationServerCfg;
import org.opends.server.api.DirectoryThread;
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;
@@ -838,6 +839,8 @@
    final JEReplicaDB replicaDB = pair.getFirst();
    replicaDB.add(updateMsg);

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

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

 opendj3-server-dev/src/server/org/opends/server/workflowelement/externalchangelog/ECLSearchOperation.java

@@ -37,6 +37,7 @@
import org.forgerock.opendj.ldap.SearchScope;
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.*;
@@ -62,6 +63,7 @@

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.replication.protocol.StartECLSessionMsg.ECLRequestType.*;
import static org.opends.server.replication.protocol.StartECLSessionMsg.Persistent.*;
@@ -120,22 +122,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
          .valueOf(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.
   */
@@ -298,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();
      }

@@ -518,6 +507,7 @@

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

          // If we're only interested in changes, then we don't actually want
@@ -596,7 +586,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))
@@ -607,7 +597,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
@@ -907,9 +897,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);
  }


 opendj3-server-dev/src/server/org/opends/server/workflowelement/externalchangelog/ECLWorkflowElement.java

@@ -26,15 +26,11 @@
 */
package org.opends.server.workflowelement.externalchangelog;



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.forgerock.i18n.slf4j.LocalizedLogger;
import org.opends.server.replication.server.ReplicationServer;
@@ -43,25 +39,15 @@
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>
{
  private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();

  /**
   *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.
   */
  public static final String ECL_WORKFLOW_ELEMENT = "EXTERNAL CHANGE LOG";
@@ -70,7 +56,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.
@@ -86,26 +72,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())
    {
@@ -166,45 +142,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.
   */

 opendj3-server-dev/src/server/org/opends/server/workflowelement/localbackend/LocalBackendAddOperation.java

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

 opendj3-server-dev/src/server/org/opends/server/workflowelement/localbackend/LocalBackendDeleteOperation.java

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

 opendj3-server-dev/src/server/org/opends/server/workflowelement/localbackend/LocalBackendModifyDNOperation.java

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

 opendj3-server-dev/src/server/org/opends/server/workflowelement/localbackend/LocalBackendModifyOperation.java

@@ -355,7 +355,7 @@
        @Override
        public void run()
        {
          for (PersistentSearch psearch : wfe.getPersistentSearches())
          for (PersistentSearch psearch : backend.getPersistentSearches())
          {
            psearch.processModify(modifiedEntry, currentEntry);
          }
@@ -666,7 +666,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);
@@ -718,19 +718,19 @@
                    entryDN, 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)
          {
@@ -742,7 +742,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(),
@@ -763,7 +763,7 @@
          setAuthorizationEntry(authorizationEntry);
          setProxiedAuthorizationDN(getName(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.
@@ -780,7 +780,7 @@
          setAuthorizationEntry(authorizationEntry);
          setProxiedAuthorizationDN(getName(authorizationEntry));
        }
        else if (oid.equals(OID_PASSWORD_POLICY_CONTROL))
        else if (OID_PASSWORD_POLICY_CONTROL.equals(oid))
        {
          pwPolicyControlRequested = true;
        }
@@ -846,13 +846,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,
@@ -1086,11 +1084,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,
@@ -1106,9 +1104,8 @@
    {
      if (pwPolicyState.passwordIsPreEncoded(v))
      {
        if ((!isInternalOperation())
            && !pwPolicyState.getAuthenticationPolicy()
                .isAllowPreEncodedPasswords())
        if (!isInternalOperation()
            && !authPolicy.isAllowPreEncodedPasswords())
        {
          pwpErrorType = PasswordPolicyErrorType.INSUFFICIENT_PASSWORD_QUALITY;
          throw new DirectoryException(ResultCode.CONSTRAINT_VIOLATION,
@@ -1121,15 +1118,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))
        {
          // Make sure that the password value doesn't already exist.
          if (pwPolicyState.passwordMatches(v))
          {
            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)
@@ -1214,7 +1209,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());
@@ -1232,48 +1227,37 @@
                    .decodeAuthPassword(av.toString());
                PasswordStorageScheme<?> scheme = DirectoryServer
                    .getAuthPasswordStorageScheme(components[0].toString());
                if (scheme != null)
                {
                  if (scheme.authPasswordMatches(v,
                      components[1].toString(), components[2].toString()))
                  {
                    builder.add(av);
                    found = true;
                  }
                }
              }
              else
              {
                if (av.equals(v))
                {
                  builder.add(v);
                  found = true;
                }
              }
            }
            else
            {
              if (UserPasswordSyntax.isEncoded(av))
              {
                String[] components = UserPasswordSyntax.decodeUserPassword(av.toString());
                PasswordStorageScheme<?> scheme = DirectoryServer
                    .getPasswordStorageScheme(toLowerCase(components[0]));
                if (scheme != null
                    && scheme.passwordMatches(v, ByteString.valueOf(components[1])))
                    && scheme.authPasswordMatches(v,
                        components[1].toString(), components[2].toString()))
                {
                  builder.add(av);
                  found = true;
                }
              }
              else
              else if (av.equals(v))
              {
                if (av.equals(v))
                {
                  builder.add(v);
                  found = true;
                }
                builder.add(v);
                found = true;
              }
            }
            else if (UserPasswordSyntax.isEncoded(av))
            {
              String[] components = UserPasswordSyntax.decodeUserPassword(av.toString());
              PasswordStorageScheme<?> scheme = DirectoryServer
                  .getPasswordStorageScheme(toLowerCase(components[0]));
              if (scheme != null
                  && scheme.passwordMatches(v, ByteString.valueOf(components[1])))
              {
                builder.add(av);
                found = true;
              }
            }
            else if (av.equals(v))
            {
              builder.add(v);
              found = true;
            }
          }
        }

@@ -1706,13 +1690,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;
@@ -1743,70 +1722,62 @@


    // 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())
          clearPasswords.addAll(currentPasswords);
        }
        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 (ByteString pw : currentPasswords)
          {
            clearPasswords.addAll(currentPasswords);
          }
          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 (ByteString pw : currentPasswords)
            if (!clearPasswords.contains(pw))
            {
              if (!clearPasswords.contains(pw))
              {
                clearPasswords.add(pw);
              }
              clearPasswords.add(pw);
            }
          }
        }
      }

        for (ByteString v : newPasswords)
      for (ByteString v : newPasswords)
      {
        LocalizableMessageBuilder invalidReason = new LocalizableMessageBuilder();
        if (! pwPolicyState.passwordIsAcceptable(this, modifiedEntry,
                                 v, clearPasswords, invalidReason))
        {
          LocalizableMessageBuilder invalidReason = new LocalizableMessageBuilder();
          if (! pwPolicyState.passwordIsAcceptable(this, modifiedEntry,
                                   v, clearPasswords, invalidReason))
          {
            pwpErrorType = PasswordPolicyErrorType.INSUFFICIENT_PASSWORD_QUALITY;
            throw new DirectoryException(ResultCode.CONSTRAINT_VIOLATION,
                ERR_MODIFY_PW_VALIDATION_FAILED.get(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 (ByteString v : newPasswords)
      {
        for (ByteString v : newPasswords)
        if (pwPolicyState.isPasswordInHistory(v)
            && (selfChange || !authPolicy.isSkipValidationForAdministrators()))
        {
          if (pwPolicyState.isPasswordInHistory(v)
              && (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();
    }


@@ -1862,7 +1833,7 @@
      return;
    }

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

 opendj3-server-dev/src/server/org/opends/server/workflowelement/localbackend/LocalBackendSearchOperation.java

@@ -56,41 +56,21 @@
{
  private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();

  /** 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.
@@ -117,10 +97,7 @@
      throws CanceledOperationException
  {
    this.backend = wfe.getBackend();

    clientConnection = getClientConnection();

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

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

      // Check for a request to cancel this operation.
      checkIfCanceled(false);
@@ -154,8 +131,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
@@ -163,7 +139,7 @@
    baseDN = getBaseDN();
    filter = getFilter();

    if ((baseDN == null) || (filter == null))
    if (baseDN == null || filter == null)
    {
      return;
    }
@@ -245,8 +221,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.
@@ -256,7 +237,7 @@
        appendErrorMessage(ERR_MAX_PSEARCH_LIMIT_EXCEEDED.get());
        return;
      }
      wfe.registerPersistentSearch(persistentSearch);
      backend.registerPersistentSearch(persistentSearch);
      persistentSearch.enable();
    }

@@ -264,7 +245,7 @@
    // Process the search in the backend and all its subordinates.
    try
    {
      if (processSearch)
      if (processSearchNow)
      {
        backend.search(this);
      }
@@ -320,14 +301,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);
@@ -397,7 +377,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(),
@@ -416,16 +396,9 @@

          Entry authorizationEntry = proxyControl.getAuthorizationEntry();
          setAuthorizationEntry(authorizationEntry);
          if (authorizationEntry == null)
          {
            setProxiedAuthorizationDN(DN.rootDN());
          }
          else
          {
            setProxiedAuthorizationDN(authorizationEntry.getName());
          }
          setProxiedAuthorizationDN(getName(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.
@@ -440,38 +413,23 @@

          Entry authorizationEntry = proxyControl.getAuthorizationEntry();
          setAuthorizationEntry(authorizationEntry);
          if (authorizationEntry == null)
          {
            setProxiedAuthorizationDN(DN.rootDN());
          }
          else
          {
            setProxiedAuthorizationDN(authorizationEntry.getName());
          }
          setProxiedAuthorizationDN(getName(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(),
@@ -479,25 +437,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.
@@ -514,6 +472,11 @@
    }
  }

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

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

 opendj3-server-dev/src/server/org/opends/server/workflowelement/localbackend/LocalBackendWorkflowElement.java

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

import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.i18n.LocalizableMessageDescriptor;
@@ -65,7 +64,7 @@
  private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();

  /** 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. */
@@ -74,13 +73,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();
@@ -109,9 +102,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;
@@ -151,29 +143,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,
@@ -183,10 +162,7 @@
    return processWorkflowElementConfig(configuration, false);
  }


  /**
   * {@inheritDoc}
   */
  /** {@inheritDoc} */
  @Override
  public ConfigChangeResult applyConfigurationChange(
      LocalBackendWorkflowElementCfg configuration
@@ -221,7 +197,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
@@ -270,8 +246,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 =
@@ -655,11 +630,7 @@
    }
  }



  /**
   * {@inheritDoc}
   */
  /** {@inheritDoc} */
  @Override
  public void execute(Operation operation) throws CanceledOperationException {
    switch (operation.getOperationType())
@@ -760,54 +731,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
@@ -828,7 +756,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, LocalizableMessageDescriptor.Arg1<Object> serverMsg,
      LocalizableMessageDescriptor.Arg1<Object> backendMsg)
      throws DirectoryException
@@ -864,5 +792,14 @@
      }
    }
  }
}

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

 opendj3-server-dev/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.*;

@@ -72,7 +73,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;
@@ -85,6 +85,7 @@
import static org.opends.server.TestCaseUtils.*;
import static org.opends.server.controls.PersistentSearchChangeType.*;
import static org.opends.server.replication.protocol.OperationContext.*;
import static org.opends.server.util.CollectionUtils.*;
import static org.opends.server.util.StaticUtils.*;
import static org.testng.Assert.*;

@@ -135,7 +136,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;

  /**
@@ -178,14 +179,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
@@ -352,7 +345,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();
@@ -368,7 +361,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);
  }

  /**
@@ -561,7 +569,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;
@@ -642,7 +650,7 @@
    ReplicationBroker s2test = null;
    ReplicationBroker s2test2 = null;

    Backend backend2 = null;
    Backend<?> backend2 = null;
    LDAPReplicationDomain domain1 = null;
    LDAPReplicationDomain domain2 = null;
    try
@@ -954,7 +962,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";
@@ -1051,9 +1059,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)
@@ -1155,7 +1162,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)
        {
@@ -1347,8 +1354,7 @@
    return a.iterator().next().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))
@@ -1936,7 +1942,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.valueOf("o=" + backendId);
@@ -1968,9 +1974,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)
      {
@@ -1994,7 +2000,7 @@
    ReplicationBroker s2test = null;
    ReplicationBroker s1test2 = null;
    ReplicationBroker s2test2 = null;
    Backend backend2 = null;
    Backend<?> backend2 = null;

    try
    {
@@ -2439,7 +2445,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);
@@ -2608,8 +2614,8 @@

    final String backendId3 = "test3";
    final DN baseDN3 = DN.valueOf("o=" + backendId3);
    Backend backend2 = null;
    Backend backend3 = null;
    Backend<?> backend2 = null;
    Backend<?> backend3 = null;
    LDAPReplicationDomain domain2 = null;
    LDAPReplicationDomain domain3 = null;
    LDAPReplicationDomain domain21 = null;
@@ -2707,7 +2713,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");