OPENDJ-1116 Introduce abstraction for the changelog DB

Jean-Noel Rouvignac

05.51.2013 84cf626ebcae1b535abe9efd3eed5cdf78bdd319

OPENDJ-1116 Introduce abstraction for the changelog DB


Added CNIndexData class to replace the conjunct use of the (CSN, baseDN, previousCookie) tuple along with the changeNumber.
It allowed to replace:
- several parameters list with all theses types
- several groups of methods returning instances of these types with duplicate code in the implementation.

Remainder of r9504: changed changeNumber from int to long.


CNIndexData.java: ADDED

*.java:
Consequence of introducing CNIndexData class.


ECLServerHandler.java:
Renamed releaseIterator() into releaseCursor().
In getNextECLUpdate(), aggregated consecutive if statements + added a debugInfo().
Renamed assignNewDraftCNAndStore() to assignNewChangeNumberAndStore().
Changed changeNumber from int to long.

ChangeNumberIndexDB.java:
Replaced getCSN(), getBaseDN() and getPreviousCookie() by getCNIndexData().
Renamed getFirstChangeNumber() to getFirstCNIndexData().
Added getLastCNIndexData().
Changed add() signature.

DraftCNDbHandler.java
Consequence of the change to implemented interface ChangeNumberIndexDB.
Changed changeNumber from int to long.
Added getChangeNumber().


ChangeNumberIndexDBCursor.java:
Replaced getCSN(), getBaseDN() and getPreviousCookie() by getCNIndexData().

DraftCNDbIterator.java
Consequence of the change to implemented interface ChangeNumberIndexDBCursor.


DraftCNData.java:
Added changeNumber field + getter.
Replaced fields value, baseDN and csn by cnIndexData + getters.
Renamed readFirstChangeNumber() and readLastChangeNumber() to readFirstCNIndexData() and readLastCNIndexData().
Added newCNIndexData().
In inner class DraftCNDBCursor, Field key is now a ReplicationDraftCNKey (was DatabaseEntry) + DraftCNData field has been replaced with CNIndexData + entry is initialized on construction.
Replaced currentValue(), currentBaseDN(), currentCSN() and currentKey() by currentData().
Removed getKey().

ReplicationDraftCNKey.java:
Added a default ctor.
In getChangeNumber(), matched the code in DraftCNDB.

DraftCNDbHandlerTest.java:
Added assertEqualTo(), getFirstChangeNumber(), getLastChangeNumber(), getPreviousCookie().

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

@@ -41,7 +41,10 @@
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.CNIndexData;
import org.opends.server.replication.server.changelog.api.ChangeNumberIndexDB;
import org.opends.server.replication.server.changelog.api.ChangeNumberIndexDBCursor;
import org.opends.server.replication.server.changelog.api.ChangelogException;
import org.opends.server.types.*;
import org.opends.server.util.ServerConstants;

@@ -71,7 +74,7 @@
  /**
   * Specifies the last changer number requested.
   */
  private int lastChangeNumber = 0;
  private long lastChangeNumber = 0;
  /**
   * Specifies whether the change number db has been read until its end.
   */
@@ -522,7 +525,7 @@
   * @throws DirectoryException
   *           When an error is raised.
   */
  private void initializeCLSearchFromChangeNumber(int startChangeNumber)
  private void initializeCLSearchFromChangeNumber(long startChangeNumber)
      throws DirectoryException
  {
    try
@@ -535,13 +538,13 @@
    catch(DirectoryException de)
    {
      TRACER.debugCaught(DebugLogLevel.ERROR, de);
      releaseIterator();
      releaseCursor();
      throw de;
    }
    catch(Exception e)
    {
      TRACER.debugCaught(DebugLogLevel.ERROR, e);
      releaseIterator();
      releaseCursor();
      throw new DirectoryException(
          ResultCode.OPERATIONS_ERROR,
          Message.raw(Category.SYNC,
@@ -561,9 +564,8 @@
   * @throws DirectoryException
   *           if a database problem occurred
   */
  private String findCookie(final int startChangeNumber)
      throws ChangelogException,
      DirectoryException
  private String findCookie(final long startChangeNumber)
      throws ChangelogException, DirectoryException
  {
    final ChangeNumberIndexDB cnIndexDB =
        replicationServer.getChangeNumberIndexDB();
@@ -581,9 +583,9 @@
        return null;
      }

      final long firstChangeNumber = cnIndexDB.getFirstChangeNumber();
      final String crossDomainStartState =
          cnIndexDB.getPreviousCookie(firstChangeNumber);
      final CNIndexData firstCNData = cnIndexDB.getFirstCNIndexData();
      final long firstChangeNumber = firstCNData.getChangeNumber();
      final String crossDomainStartState = firstCNData.getPreviousCookie();
      cnIndexDBCursor = cnIndexDB.getCursorFrom(firstChangeNumber);
      return crossDomainStartState;
    }
@@ -591,11 +593,11 @@
    // Request filter DOES contain a startChangeNumber

    // Read the draftCNDb to see whether it contains startChangeNumber
    String crossDomainStartState =
        cnIndexDB.getPreviousCookie(startChangeNumber);
    if (crossDomainStartState != null)
    CNIndexData startCNData = cnIndexDB.getCNIndexData(startChangeNumber);
    if (startCNData != null)
    {
      // found the provided startChangeNumber, let's return it
      final String crossDomainStartState = startCNData.getPreviousCookie();
      cnIndexDBCursor = cnIndexDB.getCursorFrom(startChangeNumber);
      return crossDomainStartState;
    }
@@ -615,9 +617,10 @@
    // the DB, let's use the lower limit.
    if (startChangeNumber < firstChangeNumber)
    {
      crossDomainStartState = cnIndexDB.getPreviousCookie(firstChangeNumber);
      if (crossDomainStartState != null)
      CNIndexData firstCNData = cnIndexDB.getCNIndexData(firstChangeNumber);
      if (firstCNData != null)
      {
        final String crossDomainStartState = firstCNData.getPreviousCookie();
        cnIndexDBCursor = cnIndexDB.getCursorFrom(firstChangeNumber);
        return crossDomainStartState;
      }
@@ -636,8 +639,9 @@
        return null;
      }

      final long lastKey = cnIndexDB.getLastChangeNumber();
      crossDomainStartState = cnIndexDB.getPreviousCookie(lastKey);
      final CNIndexData lastCNData = cnIndexDB.getLastCNIndexData();
      final long lastKey = lastCNData.getChangeNumber();
      final String crossDomainStartState = lastCNData.getPreviousCookie();
      cnIndexDBCursor = cnIndexDB.getCursorFrom(lastKey);
      return crossDomainStartState;

@@ -897,7 +901,7 @@
  {
    if (debugEnabled())
      TRACER.debugInfo(this + " shutdown()");
    releaseIterator();
    releaseCursor();
    for (DomainContext domainCtxt : domainCtxts) {
      if (!domainCtxt.unRegisterHandler()) {
        logError(Message.raw(Category.SYNC, Severity.NOTICE,
@@ -910,7 +914,7 @@
    domainCtxts = null;
  }

  private void releaseIterator()
  private void releaseCursor()
  {
    if (this.cnIndexDBCursor != null)
    {
@@ -1256,13 +1260,10 @@
        oldestContext.currentState.update(
            change.getUpdateMsg().getCSN());

        if (oldestContext.currentState.cover(oldestContext.stopState))
        {
          oldestContext.active = false;
        }
        if (draftCompat
            && lastChangeNumber > 0
            && change.getChangeNumber() > lastChangeNumber)
        if (oldestContext.currentState.cover(oldestContext.stopState)
            || (draftCompat
                && lastChangeNumber > 0
                && change.getChangeNumber() > lastChangeNumber))
        {
          oldestContext.active = false;
        }
@@ -1278,8 +1279,9 @@
      if (searchPhase == PERSISTENT_PHASE)
      {
        if (debugEnabled())
          clDomCtxtsToString("In getNextECLUpdate (persistent): " +
          "looking for the generalized oldest change");
          TRACER.debugInfo(clDomCtxtsToString(
              "In getNextECLUpdate (persistent): "
                  + "looking for the generalized oldest change"));

        for (DomainContext domainCtxt : domainCtxts) {
          domainCtxt.getNextEligibleMessageForDomain(operationId);
@@ -1300,7 +1302,7 @@

          if (draftCompat)
          {
            assignNewDraftCNAndStore(change);
            assignNewChangeNumberAndStore(change);
          }
          oldestChange = change;
        }
@@ -1317,21 +1319,19 @@

    if (oldestChange != null)
    {
      final CSN csn = oldestChange.getUpdateMsg().getCSN();
      if (debugEnabled())
        TRACER.debugInfo("getNextECLUpdate updates previousCookie:"
          + oldestChange.getUpdateMsg().getCSN());
        TRACER.debugInfo("getNextECLUpdate updates previousCookie:" + csn);

      // Update the current state
      previousCookie.update(
          oldestChange.getBaseDN(),
          oldestChange.getUpdateMsg().getCSN());
      previousCookie.update(oldestChange.getBaseDN(), csn);

      // Set the current value of global state in the returned message
      oldestChange.setCookie(previousCookie);

      if (debugEnabled())
        TRACER.debugInfo("getNextECLUpdate returns result oldest change =" +
                oldestChange);
        TRACER.debugInfo("getNextECLUpdate returns result oldestChange="
            + oldestChange);

    }
    return oldestChange;
@@ -1370,14 +1370,15 @@
      if (isEndOfCNIndexDBReached)
      {
        // we are at the end of the DraftCNdb in the append mode
        assignNewDraftCNAndStore(oldestChange);
        assignNewChangeNumberAndStore(oldestChange);
        return true;
      }


      // the next change from the CNIndexDB
      CSN csnFromDraftCNDb = cnIndexDBCursor.getCSN();
      String dnFromDraftCNDb = cnIndexDBCursor.getBaseDN();
      final CNIndexData cnIndexData = cnIndexDBCursor.getCNIndexData();
      final CSN csnFromDraftCNDb = cnIndexData.getCSN();
      final String dnFromDraftCNDb = cnIndexData.getBaseDN();

      if (debugEnabled())
        TRACER.debugInfo("assignChangeNumber() generating change number "
@@ -1392,10 +1393,10 @@
      {
        if (debugEnabled())
          TRACER.debugInfo("assignChangeNumber() generating change number "
              + " assigning changeNumber=" + cnIndexDBCursor.getChangeNumber()
              + " assigning changeNumber=" + cnIndexData.getChangeNumber()
              + " to change=" + oldestChange);

        oldestChange.setChangeNumber(cnIndexDBCursor.getChangeNumber());
        oldestChange.setChangeNumber(cnIndexData.getChangeNumber());
        return true;
      }

@@ -1429,8 +1430,8 @@

        if (debugEnabled())
          TRACER.debugInfo("assignChangeNumber() generating change number has"
              + "skipped to  changeNumber=" + cnIndexDBCursor.getChangeNumber()
              + " csn=" + cnIndexDBCursor.getCSN() + " End of CNIndexDB ?"
              + "skipped to  changeNumber=" + cnIndexData.getChangeNumber()
              + " csn=" + cnIndexData.getCSN() + " End of CNIndexDB ?"
              + isEndOfCNIndexDBReached);
      }
      catch (ChangelogException e)
@@ -1452,7 +1453,7 @@
    return sameDN && sameCSN;
  }

  private void assignNewDraftCNAndStore(ECLUpdateMsg change)
  private void assignNewChangeNumberAndStore(ECLUpdateMsg change)
      throws DirectoryException, ChangelogException
  {
    // generate a new change number and assign to this change
@@ -1460,11 +1461,11 @@

    // store in CNIndexDB the pair
    // (change number of the current change, state before this change)
    replicationServer.getChangeNumberIndexDB().add(
    replicationServer.getChangeNumberIndexDB().add(new CNIndexData(
        change.getChangeNumber(),
        previousCookie.toString(),
        change.getBaseDN(),
        change.getUpdateMsg().getCSN());
        change.getUpdateMsg().getCSN()));
  }

  /**
@@ -1499,7 +1500,7 @@
    }

    // End of INIT_PHASE => always release the iterator
    releaseIterator();
    releaseCursor();
  }

  /**

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

@@ -52,6 +52,7 @@
import org.opends.server.replication.common.*;
import org.opends.server.replication.plugin.MultimasterReplication;
import org.opends.server.replication.protocol.*;
import org.opends.server.replication.server.changelog.api.CNIndexData;
import org.opends.server.replication.server.changelog.api.ChangeNumberIndexDB;
import org.opends.server.replication.server.changelog.api.ChangelogException;
import org.opends.server.replication.server.changelog.je.DbHandler;
@@ -1699,37 +1700,36 @@
     *     (this diff is done domain by domain)
     */

    long lastChangeNumber;
    boolean dbEmpty = false;
    final ChangeNumberIndexDB cnIndexDB = getChangeNumberIndexDB();

    try
    {
      long firstChangeNumber = cnIndexDB.getFirstChangeNumber();
      boolean dbEmpty = true;
      long firstChangeNumber = 0;
      long lastChangeNumber = 0;

      final CNIndexData firstCNData = cnIndexDB.getFirstCNIndexData();
      final CNIndexData lastCNData = cnIndexDB.getLastCNIndexData();

      Map<String, ServerState> domainsServerStateForLastCN = null;
      CSN csnForLastCN = null;
      String domainForLastCN = null;
      if (firstChangeNumber < 1)
      if (firstCNData != null)
      {
        dbEmpty = true;
        firstChangeNumber = 0;
        lastChangeNumber = 0;
      }
      else
      {
        lastChangeNumber = cnIndexDB.getLastChangeNumber();
        dbEmpty = false;
        firstChangeNumber = firstCNData.getChangeNumber();
        lastChangeNumber = lastCNData.getChangeNumber();

        // Get the generalized state associated with the current last change
        // number and initializes from it the startStates table
        String lastCNGenState = cnIndexDB.getPreviousCookie(lastChangeNumber);
        String lastCNGenState = lastCNData.getPreviousCookie();
        if (lastCNGenState != null && lastCNGenState.length() > 0)
        {
          domainsServerStateForLastCN = MultiDomainServerState
              .splitGenStateToServerStates(lastCNGenState);
        }

        csnForLastCN = cnIndexDB.getCSN(lastChangeNumber);
        domainForLastCN = cnIndexDB.getBaseDN(lastChangeNumber);
        csnForLastCN = lastCNData.getCSN();
        domainForLastCN = lastCNData.getBaseDN();
      }

      long newestDate = 0;

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

New file
@@ -0,0 +1,113 @@
/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License, Version 1.0 only
 * (the "License").  You may not use this file except in compliance
 * with the License.
 *
 * You can obtain a copy of the license at
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE
 * or https://OpenDS.dev.java.net/OpenDS.LICENSE.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE.  If applicable,
 * add the following below this CDDL HEADER, with the fields enclosed
 * by brackets "[]" replaced with your own identifying information:
 *      Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 *
 *
 *      Copyright 2013 ForgeRock AS
 */
package org.opends.server.replication.server.changelog.api;

import org.opends.server.replication.common.CSN;

/**
 * The Change Number Index Data class represents records stored in the
 * {@link ChangeNumberIndexDB}.
 */
public class CNIndexData
{

  /** This is the key used to store the rest of the . */
  private long changeNumber;
  private String previousCookie;
  private String baseDN;
  private CSN csn;

  /**
   * Builds an instance of this class.
   *
   * @param changeNumber
   *          the change number
   * @param previousCookie
   *          the previous cookie
   * @param baseDN
   *          the baseDN
   * @param csn
   *          the replication CSN field
   */
  public CNIndexData(long changeNumber, String previousCookie, String baseDN,
      CSN csn)
  {
    super();
    this.changeNumber = changeNumber;
    this.previousCookie = previousCookie;
    this.baseDN = baseDN;
    this.csn = csn;
  }

  /**
   * Getter for the baseDN field.
   *
   * @return the baseDN
   */
  public String getBaseDN()
  {
    return baseDN;
  }

  /**
   * Getter for the replication CSN field.
   *
   * @return The replication CSN field.
   */
  public CSN getCSN()
  {
    return csn;
  }

  /**
   * Getter for the change number field.
   *
   * @return The change number field.
   */
  public long getChangeNumber()
  {
    return changeNumber;
  }

  /**
   * Get the previous cookie field.
   *
   * @return the previous cookie.
   */
  public String getPreviousCookie()
  {
    return previousCookie;
  }

  /** {@inheritDoc} */
  @Override
  public String toString()
  {
    return "changeNumber=" + changeNumber + " csn=" + csn + " baseDN=" + baseDN
        + " previousCookie=" + previousCookie;
  }
}

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

@@ -26,13 +26,12 @@
 */
package org.opends.server.replication.server.changelog.api;

import org.opends.server.replication.common.CSN;

/**
 * This class stores an index of all the changes seen by this server. The index
 * is sorted by a global ordering as defined in the CSN class. The index links a
 * <code>changeNumber</code> to the corresponding {@link CSN}. The {@link CSN}
 * then links to a corresponding change in one of the {@link ReplicaDB}s.
 * This class stores an index of all the changes seen by this server in the form
 * of {@link CNIndexData}. The index is sorted by a global ordering as defined
 * in the CSN class. The index links a <code>changeNumber</code> to the
 * corresponding CSN. The CSN then links to a corresponding change in one of the
 * ReplicaDBs.
 *
 * @see <a href=
 * "https://wikis.forgerock.org/confluence/display/OPENDJ/OpenDJ+Domain+Names"
@@ -44,46 +43,33 @@
{

  /**
   * Get the CSN associated to a provided change number.
   * Get the {@link CNIndexData} record associated to a provided change number.
   *
   * @param changeNumber
   *          the provided change number.
   * @return the associated CSN, null when none.
   * @return the {@link CNIndexData} record, null when none.
   * @throws ChangelogException
   *           if a database problem occurs.
   */
  public CSN getCSN(long changeNumber) throws ChangelogException;
  CNIndexData getCNIndexData(long changeNumber) throws ChangelogException;

  /**
   * Get the baseDN associated to a provided change number.
   * Get the first {@link CNIndexData} record stored in this DB.
   *
   * @param changeNumber
   *          the provided change number.
   * @return the baseDN, null when none.
   * @return Returns the first {@link CNIndexData} record in this DB.
   * @throws ChangelogException
   *           if a database problem occurs.
   */
  public String getBaseDN(long changeNumber) throws ChangelogException;
  CNIndexData getFirstCNIndexData() throws ChangelogException;

  /**
   * Get the previous cookie associated to a provided change number.
   * Get the last {@link CNIndexData} record stored in this DB.
   *
   * @param changeNumber
   *          the provided change number.
   * @return the previous cookie, null when none.
   * @return Returns the last {@link CNIndexData} record in this DB
   * @throws ChangelogException
   *           if a database problem occurs.
   */
  String getPreviousCookie(long changeNumber) throws ChangelogException;

  /**
   * Get the first change number stored in this DB.
   *
   * @return Returns the first change number in this DB.
   * @throws ChangelogException
   *           if a database problem occurs.
   */
  long getFirstChangeNumber() throws ChangelogException;
  CNIndexData getLastCNIndexData() throws ChangelogException;

  /**
   * Get the last change number stored in this DB.
@@ -101,19 +87,12 @@
   * This method is blocking if the size of the list of message is larger than
   * its maximum.
   *
   * @param changeNumber
   *          The change number for this record in this DB.
   * @param previousCookie
   *          The value of the previous cookie.
   * @param baseDN
   *          The associated baseDN.
   * @param csn
   *          The associated replication CSN.
   * @param cnIndexData
   *          The {@link CNIndexData} record to add to this DB.
   * @throws ChangelogException
   *           if a database problem occurs.
   */
  void add(long changeNumber, String previousCookie, String baseDN, CSN csn)
      throws ChangelogException;
  void add(CNIndexData cnIndexData) throws ChangelogException;

  /**
   * Generate a new {@link ChangeNumberIndexDBCursor} that allows to browse the

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

@@ -28,8 +28,6 @@

import java.io.Closeable;

import org.opends.server.replication.common.CSN;

/**
 * Iterator into the changelog database. Once it is not used anymore, a
 * ChangelogDBIterator must be closed to release all the resources into the
@@ -39,25 +37,11 @@
{

  /**
   * Getter for the replication CSN field.
   * Getter for the {@link CNIndexData} record.
   *
   * @return The replication CSN field.
   * @return The replication CNIndexData record.
   */
  CSN getCSN();

  /**
   * Getter for the baseDN field.
   *
   * @return The service ID.
   */
  String getBaseDN();

  /**
   * Getter for the change number field.
   *
   * @return The change number field.
   */
  long getChangeNumber();
  CNIndexData getCNIndexData();

  /**
   * Skip to the next record of the database.

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

@@ -34,7 +34,7 @@
import org.opends.messages.Message;
import org.opends.messages.MessageBuilder;
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.server.changelog.api.CNIndexData;
import org.opends.server.replication.server.changelog.api.ChangelogException;
import org.opends.server.types.DebugLogLevel;

@@ -56,7 +56,6 @@
public class DraftCNDB
{
  private static final DebugTracer TRACER = getTracer();
  private static final int DATABASE_EMPTY = 0;

  private Database db;
  private ReplicationDbEnv dbenv;
@@ -83,22 +82,19 @@

  /**
   * Add an entry to the database.
   * @param changeNumber the provided change number.
   *
   * @param value        the provided value to be stored associated
   *                     with this change number.
   * @param domainBaseDN the provided domainBaseDn to be stored associated
   *                     with this change number.
   * @param csn the provided replication CSN to be
   *                     stored associated with this change number.
   * @param cnIndexData
   *          the provided {@link CNIndexData} to be stored.
   */
  public void addEntry(long changeNumber, String value, String domainBaseDN,
      CSN csn)
  public void addEntry(CNIndexData cnIndexData)
  {
    try
    {
      final long changeNumber = cnIndexData.getChangeNumber();
      DatabaseEntry key = new ReplicationDraftCNKey(changeNumber);
      DatabaseEntry data = new DraftCNData(value, domainBaseDN, csn);
      DatabaseEntry data =
          new DraftCNData(changeNumber, cnIndexData.getPreviousCookie(),
              cnIndexData.getBaseDN(), cnIndexData.getCSN());

      // Use a transaction so that we can override durability.
      Transaction txn = null;
@@ -220,9 +216,12 @@

  /**
   * Read the first Change from the database, 0 when none.
   *
   * @return the first change number.
   * @throws ChangelogException
   *           if a database problem occurred
   */
  public int readFirstChangeNumber()
  public CNIndexData readFirstCNIndexData() throws ChangelogException
  {
    try
    {
@@ -233,18 +232,18 @@
        // If the DB has been closed then return immediately.
        if (isDBClosed())
        {
          return DATABASE_EMPTY;
          return null;
        }

        cursor = db.openCursor(null, null);
        DatabaseEntry key = new DatabaseEntry();
        ReplicationDraftCNKey key = new ReplicationDraftCNKey();
        DatabaseEntry entry = new DatabaseEntry();
        if (cursor.getFirst(key, entry, LockMode.DEFAULT) != SUCCESS)
        {
          return DATABASE_EMPTY;
          return null;
        }

        return Integer.parseInt(decodeUTF8(key.getData()));
        return newCNIndexData(key, entry);
      }
      finally
      {
@@ -254,10 +253,17 @@
    catch (DatabaseException e)
    {
      dbenv.shutdownOnException(e);
      return DATABASE_EMPTY;
      return null;
    }
  }

  private CNIndexData newCNIndexData(ReplicationDraftCNKey key,
      DatabaseEntry data) throws ChangelogException
  {
    return new DraftCNData(key.getChangeNumber(), data.getData())
        .getCNIndexData();
  }

  /**
   * Return the record count.
   * @return the record count.
@@ -270,7 +276,7 @@
      // If the DB has been closed then return immediately.
      if (isDBClosed())
      {
        return DATABASE_EMPTY;
        return 0;
      }

      return db.count();
@@ -283,14 +289,17 @@
    {
      dbCloseLock.readLock().unlock();
    }
    return DATABASE_EMPTY;
    return 0;
  }

  /**
   * Read the last change number from the database.
   *
   * @return the last change number.
   * @throws ChangelogException
   *           if a database problem occurred
   */
  public int readLastChangeNumber()
  public CNIndexData readLastCNIndexData() throws ChangelogException
  {
    try
    {
@@ -301,18 +310,18 @@
        // If the DB has been closed then return immediately.
        if (isDBClosed())
        {
          return DATABASE_EMPTY;
          return null;
        }

        cursor = db.openCursor(null, null);
        DatabaseEntry key = new DatabaseEntry();
        ReplicationDraftCNKey key = new ReplicationDraftCNKey();
        DatabaseEntry entry = new DatabaseEntry();
        if (cursor.getLast(key, entry, LockMode.DEFAULT) != SUCCESS)
        {
          return DATABASE_EMPTY;
          return null;
        }

        return Integer.parseInt(decodeUTF8(key.getData()));
        return newCNIndexData(key, entry);
      }
      finally
      {
@@ -322,7 +331,7 @@
    catch (DatabaseException e)
    {
      dbenv.shutdownOnException(e);
      return DATABASE_EMPTY;
      return null;
    }
  }

@@ -348,9 +357,9 @@
     * Will be set non null for a write cursor.
     */
    private final Transaction txn;
    private final DatabaseEntry key;
    private final DatabaseEntry entry;
    private DraftCNData cnData;
    private final ReplicationDraftCNKey key;
    private final DatabaseEntry entry = new DatabaseEntry();
    private CNIndexData cnIndexData;
    private boolean isClosed = false;


@@ -365,7 +374,6 @@
    private DraftCNDBCursor(long startChangeNumber) throws ChangelogException
    {
      this.key = new ReplicationDraftCNKey(startChangeNumber);
      this.entry = new DatabaseEntry();

      // Take the lock. From now on, whatever error that happen in the life
      // of this cursor should end by unlocking that lock. We must also
@@ -405,12 +413,12 @@
            }
            else
            {
              cnData = new DraftCNData(entry.getData());
              cnIndexData = newCNIndexData(this.key, entry);
            }
          }
          else
          {
            cnData = new DraftCNData(entry.getData());
            cnIndexData = newCNIndexData(this.key, entry);
          }
        }

@@ -431,15 +439,12 @@
      }
    }



    private DraftCNDBCursor() throws ChangelogException
    {
      Transaction localTxn = null;
      Cursor localCursor = null;

      this.key = new DatabaseEntry();
      this.entry = new DatabaseEntry();
      this.key = new ReplicationDraftCNKey();

      // We'll go on only if no close or no clear is running
      dbCloseLock.readLock().lock();
@@ -544,105 +549,29 @@
    }

    /**
     * Getter for the value field of the current cursor.
     * @return The current value field.
     */
    public String currentValue()
    {
      if (isClosed)
      {
        return null;
      }

      try
      {
        if (cnData != null)
        {
          return cnData.getValue();
        }
      }
      catch(Exception e)
      {
        TRACER.debugCaught(DebugLogLevel.ERROR, e);
      }
      return null;
    }

    /**
     * Getter for the baseDN field of the current cursor.
     * @return The current baseDN.
     */
    public String currentBaseDN()
    {
      if (isClosed)
      {
        return null;
      }

      try
      {
        if (cnData != null)
        {
          return cnData.getBaseDN();
        }
      }
      catch(Exception e)
      {
        TRACER.debugCaught(DebugLogLevel.ERROR, e);
      }
      return null;
    }

    /**
     * Getter for the integer value of the current cursor, representing
     * the current change number being processed.
     * Returns the {@link CNIndexData} at the current position of the cursor.
     *
     * @return the current change number as an integer.
     * @return The current {@link CNIndexData}.
     */
    public int currentKey()
    public CNIndexData currentData()
    {
      if (isClosed)
      {
        return -1;
        return null;
      }

      try
      {
        String str = decodeUTF8(key.getData());
        return Integer.parseInt(str);
        return cnIndexData;
      }
      catch (Exception e)
      {
        TRACER.debugCaught(DebugLogLevel.ERROR, e);
      }
      return -1;
    }

    /**
     * Returns the replication CSN associated with the current key.
     * @return the replication CSN
     */
    public CSN currentCSN()
    {
      if (isClosed)
      {
        return null;
      }

      try
      {
        if (cnData != null)
        {
          return cnData.getCSN();
        }
      }
      catch(Exception e)
      {
        TRACER.debugCaught(DebugLogLevel.ERROR, e);
      }
      return null;
    }


    /**
     * Go to the next record on the cursor.
     * @return the next record on this cursor.
@@ -659,10 +588,10 @@
        OperationStatus status = cursor.getNext(key, entry, LockMode.DEFAULT);
        if (status != OperationStatus.SUCCESS)
        {
          cnData = null;
          cnIndexData = null;
          return false;
        }
        cnData = new DraftCNData(entry.getData());
        cnIndexData = newCNIndexData(this.key, entry);
      }
      catch(Exception e)
      {
@@ -693,20 +622,6 @@
      }
    }

    /**
     * Returns the current key associated with this cursor.
     *
     * @return The current key associated with this cursor.
     */
    public DatabaseEntry getKey()
    {
      if (isClosed)
      {
        return null;
      }

      return key;
    }
  }

  /**

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

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

import org.opends.messages.Message;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.server.changelog.api.CNIndexData;
import org.opends.server.replication.server.changelog.api.ChangelogException;

import com.sleepycat.je.DatabaseEntry;
@@ -46,18 +47,25 @@

  private static final long serialVersionUID = 1L;

  private String value;
  private String baseDN;
  private CSN csn;
  private long changeNumber;
  private CNIndexData cnIndexData;

  /**
   * Creates a record to be stored in the DraftCNDB.
   * @param previousCookie The previous cookie.
   * @param baseDN The baseDN (domain DN).
   * @param csn The replication CSN.
   *
   * @param changeNumber
   *          the change number
   * @param previousCookie
   *          The previous cookie
   * @param baseDN
   *          The baseDN (domain DN)
   * @param csn
   *          The replication CSN
   */
  public DraftCNData(String previousCookie, String baseDN, CSN csn)
  public DraftCNData(long changeNumber, String previousCookie, String baseDN,
      CSN csn)
  {
    this.changeNumber = changeNumber;
    String record =
        previousCookie + FIELD_SEPARATOR + baseDN + FIELD_SEPARATOR + csn;
    setData(getBytes(record));
@@ -65,29 +73,38 @@

  /**
   * Creates a record to be stored in the DraftCNDB from the provided byte[].
   * @param data the provided byte[].
   * @throws ChangelogException a.
   *
   * @param changeNumber
   *          the change number
   * @param data
   *          the provided byte[]
   * @throws ChangelogException
   *           if a database problem occurred
   */
  public DraftCNData(byte[] data) throws ChangelogException
  public DraftCNData(long changeNumber, byte[] data) throws ChangelogException
  {
    decodeData(data);
    this.changeNumber = changeNumber;
    this.cnIndexData = decodeData(changeNumber, data);
  }

  /**
   * Decode a record into fields.
   * @param data the provided byte array.
   * @throws ChangelogException when a problem occurs.
   * Decode and returns a {@link CNIndexData} record.
   *
   * @param changeNumber
   * @param data
   *          the provided byte array.
   * @return the decoded {@link CNIndexData} record
   * @throws ChangelogException
   *           when a problem occurs.
   */
  public void decodeData(byte[] data) throws ChangelogException
  private CNIndexData decodeData(long changeNumber, byte[] data)
      throws ChangelogException
  {
    try
    {
      String stringData = new String(data, "UTF-8");

      String[] str = stringData.split(FIELD_SEPARATOR, 3);
      value = str[0];
      baseDN = str[1];
      csn = new CSN(str[2]);
      return new CNIndexData(changeNumber, str[0], str[1], new CSN(str[2]));
    }
    catch (UnsupportedEncodingException e)
    {
@@ -98,43 +115,17 @@
  }

  /**
   * Getter for the value.
   * Getter for the decoded {@link CNIndexData} record.
   *
   * @return the value.
   * @throws ChangelogException when a problem occurs.
   */
  public String getValue() throws ChangelogException
  {
    if (value == null)
      decodeData(getData());
    return this.value;
  }

  /**
   * Getter for the service ID.
   *
   * @return The baseDN
   * @throws ChangelogException when a problem occurs.
   */
  public String getBaseDN() throws ChangelogException
  {
    if (value == null)
      decodeData(getData());
    return this.baseDN;
  }

  /**
   * Getter for the replication CSN.
   *
   * @return the replication CSN.
   * @return the CNIndexData record.
   * @throws ChangelogException
   *           when a problem occurs.
   */
  public CSN getCSN() throws ChangelogException
  public CNIndexData getCNIndexData() throws ChangelogException
  {
    if (value == null)
      decodeData(getData());
    return this.csn;
    if (cnIndexData == null)
      cnIndexData = decodeData(changeNumber, getData());
    return cnIndexData;
  }

  /**
@@ -144,19 +135,7 @@
  @Override
  public String toString()
  {
    StringBuilder buffer = new StringBuilder();
    toString(buffer);
    return buffer.toString();
    return "DraftCNData : [" + cnIndexData + "]";
  }

  /**
   * Dump a string representation of these data into the provided buffer.
   * @param buffer the provided buffer.
   */
  public void toString(StringBuilder buffer)
  {
    buffer.append("DraftCNData : [value=").append(value);
    buffer.append("] [serviceID=").append(baseDN);
    buffer.append("] [csn=").append(csn).append("]");
  }
}

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

@@ -48,6 +48,7 @@
import org.opends.server.replication.server.changelog.je.DraftCNDB.*;
import org.opends.server.types.Attribute;
import org.opends.server.types.Attributes;
import org.opends.server.types.DebugLogLevel;
import org.opends.server.types.InitializationException;

import static org.opends.messages.ReplicationMessages.*;
@@ -78,12 +79,12 @@
   * FIXME Is this field that useful? {@link #getFirstChangeNumber()} does not
   * even use it!
   */
  private int firstChangeNumber = NO_KEY;
  private long firstChangeNumber = NO_KEY;
  /**
   * FIXME Is this field that useful? {@link #getLastChangeNumber()} does not
   * even use it!
   */
  private int lastChangeNumber = NO_KEY;
  private long lastChangeNumber = NO_KEY;
  private DbMonitorProvider dbMonitor = new DbMonitorProvider();
  private boolean shutdown = false;
  private boolean trimDone = false;
@@ -122,8 +123,8 @@

    // DB initialization
    db = new DraftCNDB(dbenv);
    firstChangeNumber = db.readFirstChangeNumber();
    lastChangeNumber = db.readLastChangeNumber();
    firstChangeNumber = getChangeNumber(db.readFirstCNIndexData());
    lastChangeNumber = getChangeNumber(db.readLastCNIndexData());

    // Trimming thread
    thread = new DirectoryThread(this, "Replication DraftCN db");
@@ -134,34 +135,50 @@
    DirectoryServer.registerMonitorProvider(dbMonitor);
  }

  private long getChangeNumber(CNIndexData cnIndexData)
      throws ChangelogException
  {
    if (cnIndexData != null)
    {
      return cnIndexData.getChangeNumber();
    }
    return 0;
  }

  /** {@inheritDoc} */
  @Override
  public synchronized void add(long changeNumber, String previousCookie,
      String baseDN, CSN csn)
  public void add(CNIndexData cnIndexData) throws ChangelogException
  {
    db.addEntry(changeNumber, previousCookie, baseDN, csn);
    db.addEntry(cnIndexData);

    if (debugEnabled())
      TRACER.debugInfo(
          "In DraftCNDbhandler.add, added: "
        + " key=" + changeNumber
        + " previousCookie=" + previousCookie
        + " baseDN=" + baseDN
        + " csn=" + csn);
      TRACER.debugInfo("In DraftCNDbhandler.add, added: " + cnIndexData);
  }

  /** {@inheritDoc} */
  @Override
  public long getFirstChangeNumber()
  public CNIndexData getFirstCNIndexData() throws ChangelogException
  {
    return db.readFirstChangeNumber();
    return db.readFirstCNIndexData();
  }

  /** {@inheritDoc} */
  @Override
  public long getLastChangeNumber()
  public CNIndexData getLastCNIndexData() throws ChangelogException
  {
    return db.readLastChangeNumber();
    return db.readLastCNIndexData();
  }

  /** {@inheritDoc} */
  @Override
  public long getLastChangeNumber() throws ChangelogException
  {
    final CNIndexData data = getLastCNIndexData();
    if (data != null)
    {
      return data.getChangeNumber();
    }
    return 0;
  }

  /**
@@ -331,8 +348,8 @@
          }

          // From the draftCNDb change record, get the domain and CSN
          final CSN csn = cursor.currentCSN();
          final String baseDN = cursor.currentBaseDN();
          final CNIndexData data = cursor.currentData();
          final String baseDN = data.getBaseDN();
          if (baseDNToClear != null && baseDNToClear.equalsIgnoreCase(baseDN))
          {
            cursor.delete();
@@ -350,10 +367,11 @@
            continue;
          }

          final CSN csn = data.getCSN();
          final ServerState startState = domain.getStartState();
          final CSN fcsn = startState.getCSN(csn.getServerId());

          final int currentChangeNumber = cursor.currentKey();
          final long currentChangeNumber = data.getChangeNumber();

          if (csn.older(fcsn))
          {
@@ -366,7 +384,7 @@
          {
            Map<String, ServerState> csnStartStates =
                MultiDomainServerState.splitGenStateToServerStates(
                        cursor.currentValue());
                        data.getPreviousCookie());
            csnVector = csnStartStates.get(baseDN);

            if (debugEnabled())
@@ -430,12 +448,37 @@
    public List<Attribute> getMonitorData()
    {
      List<Attribute> attributes = new ArrayList<Attribute>();
      attributes.add(Attributes.create("first-draft-changenumber",
          Integer.toString(db.readFirstChangeNumber())));
      attributes.add(Attributes.create("last-draft-changenumber",
          Integer.toString(db.readLastChangeNumber())));
      attributes.add(Attributes.create("count",
          Long.toString(count())));

      try
      {
        CNIndexData firstCNData = db.readFirstCNIndexData();
        String firstCN = String.valueOf(firstCNData.getChangeNumber());
        attributes.add(Attributes.create("first-draft-changenumber", firstCN));
      }
      catch (ChangelogException e)
      {
        if (debugEnabled())
          TRACER.debugCaught(DebugLogLevel.WARNING, e);
        attributes.add(Attributes.create("first-draft-changenumber", "0"));
      }

      try
      {
        CNIndexData lastCNData = db.readLastCNIndexData();
        if (lastCNData != null)
        {
          String lastCN = String.valueOf(lastCNData.getChangeNumber());
          attributes.add(Attributes.create("last-draft-changenumber", lastCN));
        }
      }
      catch (ChangelogException e)
      {
        if (debugEnabled())
          TRACER.debugCaught(DebugLogLevel.WARNING, e);
        attributes.add(Attributes.create("last-draft-changenumber", "0"));
      }

      attributes.add(Attributes.create("count", Long.toString(count())));
      return attributes;
    }

@@ -482,8 +525,8 @@
  public void clear() throws ChangelogException
  {
    db.clear();
    firstChangeNumber = db.readFirstChangeNumber();
    lastChangeNumber = db.readLastChangeNumber();
    firstChangeNumber = getChangeNumber(db.readFirstCNIndexData());
    lastChangeNumber = getChangeNumber(db.readLastCNIndexData());
  }

  private ReentrantLock lock = new ReentrantLock();
@@ -516,13 +559,14 @@

  /** {@inheritDoc} */
  @Override
  public String getPreviousCookie(long changeNumber) throws ChangelogException
  public CNIndexData getCNIndexData(long changeNumber)
      throws ChangelogException
  {
    DraftCNDBCursor cursor = null;
    try
    {
      cursor = db.openReadCursor(changeNumber);
      return cursor.currentValue();
      return cursor.currentData();
    }
    finally
    {
@@ -530,35 +574,4 @@
    }
  }

  /** {@inheritDoc} */
  @Override
  public CSN getCSN(long changeNumber) throws ChangelogException
  {
    DraftCNDBCursor cursor = null;
    try
    {
      cursor = db.openReadCursor(changeNumber);
      return cursor.currentCSN();
    }
    finally
    {
      close(cursor);
    }
  }

  /** {@inheritDoc} */
  @Override
  public String getBaseDN(long changeNumber) throws ChangelogException
  {
    DraftCNDBCursor cursor = null;
    try
    {
      cursor = db.openReadCursor(changeNumber);
      return cursor.currentBaseDN();
    }
    finally
    {
      close(cursor);
    }
  }
}

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

@@ -29,7 +29,6 @@

import org.opends.messages.Message;
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.server.changelog.api.*;
import org.opends.server.replication.server.changelog.je.DraftCNDB.*;
import org.opends.server.types.DebugLogLevel;
@@ -68,13 +67,13 @@

  /** {@inheritDoc} */
  @Override
  public String getBaseDN()
  public CNIndexData getCNIndexData()
  {
    try
    {
      return this.draftCNDbCursor.currentBaseDN();
      return this.draftCNDbCursor.currentData();
    }
    catch(Exception e)
    catch (Exception e)
    {
      TRACER.debugCaught(DebugLogLevel.ERROR, e);
      return null;
@@ -83,28 +82,6 @@

  /** {@inheritDoc} */
  @Override
  public CSN getCSN()
  {
    try
    {
      return this.draftCNDbCursor.currentCSN();
    }
    catch(Exception e)
    {
      TRACER.debugCaught(DebugLogLevel.ERROR, e);
      return null;
    }
  }

  /** {@inheritDoc} */
  @Override
  public long getChangeNumber()
  {
    return ((ReplicationDraftCNKey) draftCNDbCursor.getKey()).getChangeNumber();
  }

  /** {@inheritDoc} */
  @Override
  public boolean next() throws ChangelogException
  {
    if (draftCNDbCursor != null)

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

@@ -31,6 +31,8 @@

import com.sleepycat.je.DatabaseEntry;

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

/**
 * Useful to create ReplicationServer keys from sequence numbers.
 */
@@ -39,8 +41,18 @@
  private static final long serialVersionUID = 1L;

  /**
   * Creates a ReplicationDraftCNKey that can start anywhere in the DB.
   */
  public ReplicationDraftCNKey()
  {
    super();
  }

  /**
   * Creates a new ReplicationKey from the given change number.
   * @param changeNumber The change number to use.
   *
   * @param changeNumber
   *          The change number to use.
   */
  public ReplicationDraftCNKey(long changeNumber)
  {
@@ -63,6 +75,6 @@
   */
  public long getChangeNumber()
  {
    return Long.valueOf(new String(getData()));
    return Long.valueOf(decodeUTF8(getData()));
  }
}

 opends/tests/unit-tests-testng/src/server/org/opends/server/replication/server/changelog/je/DraftCNDbHandlerTest.java

@@ -36,6 +36,8 @@
import org.opends.server.replication.common.CSNGenerator;
import org.opends.server.replication.server.ReplServerFakeConfiguration;
import org.opends.server.replication.server.ReplicationServer;
import org.opends.server.replication.server.changelog.api.CNIndexData;
import org.opends.server.replication.server.changelog.api.ChangeNumberIndexDB;
import org.opends.server.replication.server.changelog.api.ChangeNumberIndexDBCursor;
import org.opends.server.replication.server.changelog.api.ChangelogException;
import org.opends.server.replication.server.changelog.je.DraftCNDB.DraftCNDBCursor;
@@ -45,21 +47,17 @@
import static org.testng.Assert.*;

/**
 * Test the DraftCNDbHandler class with 2 kinds of cleaning of the db :
 * - periodic trim
 * - call to clear method()
 * Test the DraftCNDbHandler class with 2 kinds of cleaning of the db : -
 * periodic trim - call to clear method()
 */
@SuppressWarnings("javadoc")
public class DraftCNDbHandlerTest extends ReplicationTestCase
{
  /**
   * This test makes basic operations of a DraftCNDb :
   * - create the db
   * - add records
   * - read them with a cursor
   * - set a very short trim period
   * - wait for the db to be trimmed / here since the changes are not stored in
   *   the replication changelog, the draftCNDb will be cleared.
   * This test makes basic operations of a DraftCNDb : - create the db - add
   * records - read them with a cursor - set a very short trim period - wait for
   * the db to be trimmed / here since the changes are not stored in the
   * replication changelog, the draftCNDb will be cleared.
   */
  @Test()
  void testDraftCNDbHandlerTrim() throws Exception
@@ -99,40 +97,32 @@
      String baseDN2 = "baseDN2";
      String baseDN3 = "baseDN3";

      CSNGenerator gen = new CSNGenerator( 1, 0);
      CSNGenerator gen = new CSNGenerator(1, 0);
      CSN csn1 = gen.newCSN();
      CSN csn2 = gen.newCSN();
      CSN csn3 = gen.newCSN();

      // Add records
      handler.add(cn1, value1, baseDN1, csn1);
      handler.add(cn2, value2, baseDN2, csn2);
      handler.add(cn3, value3, baseDN3, csn3);
      handler.add(new CNIndexData(cn1, value1, baseDN1, csn1));
      handler.add(new CNIndexData(cn2, value2, baseDN2, csn2));
      handler.add(new CNIndexData(cn3, value3, baseDN3, csn3));

      // The ChangeNumber should not get purged
      final long firstChangeNumber = handler.getFirstChangeNumber();
      final long firstChangeNumber = getFirstChangeNumber(handler);
      assertEquals(firstChangeNumber, cn1);
      assertEquals(handler.getLastChangeNumber(), cn3);
      assertEquals(getLastChangeNumber(handler), cn3);

      DraftCNDBCursor dbc = handler.getReadCursor(firstChangeNumber);
      try
      {
        assertEquals(dbc.currentCSN(), csn1);
        assertEquals(dbc.currentBaseDN(), baseDN1);
        assertEquals(dbc.currentValue(), value1);
        assertEqualTo(dbc.currentData(), csn1, baseDN1, value1);
        assertTrue(dbc.toString().length() != 0);

        assertTrue(dbc.next());

        assertEquals(dbc.currentCSN(), csn2);
        assertEquals(dbc.currentBaseDN(), baseDN2);
        assertEquals(dbc.currentValue(), value2);
        assertEqualTo(dbc.currentData(), csn2, baseDN2, value2);

        assertTrue(dbc.next());

        assertEquals(dbc.currentCSN(), csn3);
        assertEquals(dbc.currentBaseDN(), baseDN3);
        assertEquals(dbc.currentValue(), value3);
        assertEqualTo(dbc.currentData(), csn3, baseDN3, value3);

        assertFalse(dbc.next());
      }
@@ -144,12 +134,12 @@
      handler.setPurgeDelay(100);

      // Check the db is cleared.
      while(handler.count()!=0)
      while (handler.count() != 0)
      {
        Thread.sleep(200);
      }
      assertEquals(handler.getFirstChangeNumber(), 0);
      assertEquals(handler.getLastChangeNumber(), 0);
      assertEquals(getFirstChangeNumber(handler), 0);
      assertEquals(getLastChangeNumber(handler), 0);
    }
    finally
    {
@@ -163,6 +153,14 @@
    }
  }

  private void assertEqualTo(CNIndexData data, CSN csn, String baseDN,
      String cookie)
  {
    assertEquals(data.getCSN(), csn);
    assertEquals(data.getBaseDN(), baseDN);
    assertEquals(data.getPreviousCookie(), cookie);
  }

  private File createCleanDir() throws IOException
  {
    String buildRoot = System.getProperty(TestCaseUtils.PROPERTY_BUILD_ROOT);
@@ -225,26 +223,26 @@
      String baseDN2 = "baseDN2";
      String baseDN3 = "baseDN3";

      CSNGenerator gen = new CSNGenerator( 1, 0);
      CSNGenerator gen = new CSNGenerator(1, 0);
      CSN csn1 = gen.newCSN();
      CSN csn2 = gen.newCSN();
      CSN csn3 = gen.newCSN();

      // Add records
      handler.add(cn1, value1, baseDN1, csn1);
      handler.add(cn2, value2, baseDN2, csn2);
      handler.add(cn3, value3, baseDN3, csn3);
      handler.add(new CNIndexData(cn1, value1, baseDN1, csn1));
      handler.add(new CNIndexData(cn2, value2, baseDN2, csn2));
      handler.add(new CNIndexData(cn3, value3, baseDN3, csn3));
      Thread.sleep(500);

      // Checks
      assertEquals(handler.getFirstChangeNumber(), cn1);
      assertEquals(handler.getLastChangeNumber(), cn3);
      assertEquals(getFirstChangeNumber(handler), cn1);
      assertEquals(getLastChangeNumber(handler), cn3);

      assertEquals(handler.count(), 3, "Db count");

      assertEquals(handler.getPreviousCookie(cn1), value1);
      assertEquals(handler.getPreviousCookie(cn2), value2);
      assertEquals(handler.getPreviousCookie(cn3), value3);
      assertEquals(getPreviousCookie(handler, cn1), value1);
      assertEquals(getPreviousCookie(handler, cn2), value2);
      assertEquals(getPreviousCookie(handler, cn3), value3);

      ChangeNumberIndexDBCursor cursor = handler.getCursorFrom(cn1);
      assertCursorReadsInOrder(cursor, cn1, cn2, cn3);
@@ -258,8 +256,8 @@
      handler.clear();

      // Check the db is cleared.
      assertEquals(handler.getFirstChangeNumber(), 0);
      assertEquals(handler.getLastChangeNumber(), 0);
      assertEquals(getFirstChangeNumber(handler), 0);
      assertEquals(getLastChangeNumber(handler), 0);
      assertEquals(handler.count(), 0);
    }
    finally
@@ -274,6 +272,29 @@
    }
  }

  private long getFirstChangeNumber(ChangeNumberIndexDB handler) throws Exception
  {
    return handler.getFirstCNIndexData().getChangeNumber();
  }

  private long getLastChangeNumber(ChangeNumberIndexDB handler) throws Exception
  {
    return handler.getLastCNIndexData().getChangeNumber();
  }

  private String getPreviousCookie(DraftCNDbHandler handler, long changeNumber) throws Exception
  {
    ChangeNumberIndexDBCursor cursor = handler.getCursorFrom(changeNumber);
    try
    {
      return cursor.getCNIndexData().getPreviousCookie();
    }
    finally
    {
      StaticUtils.close(cursor);
    }
  }

  private void assertCursorReadsInOrder(ChangeNumberIndexDBCursor cursor,
      int... sns) throws ChangelogException
  {
@@ -281,7 +302,7 @@
    {
      for (int i = 0; i < sns.length; i++)
      {
        assertEquals(cursor.getChangeNumber(), sns[i]);
        assertEquals(cursor.getCNIndexData().getChangeNumber(), sns[i]);
        final boolean isNotLast = i + 1 < sns.length;
        assertEquals(cursor.next(), isNotLast);
      }