OPENDJ-1116 Introduce abstraction for the changelog DB

Jean-Noel Rouvignac

10.45.2013 87b6a9294f437fd6312c2fbe2292ee64babd8d7e

OPENDJ-1116 Introduce abstraction for the changelog DB

Changes after review from Matthew Swift.

Renamed CNIndexRecord to ChangeNumberIndexRecord to be consistent with ChangeNumberIndexDB name.

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

@@ -40,10 +40,7 @@
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.CNIndexRecord;
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.api.DBCursor;
import org.opends.server.replication.server.changelog.api.*;
import org.opends.server.types.*;
import org.opends.server.util.ServerConstants;

@@ -77,7 +74,7 @@
  private String operationId;

  /** Cursor on the {@link ChangeNumberIndexDB}. */
  private DBCursor<CNIndexRecord> cnIndexDBCursor;
  private DBCursor<ChangeNumberIndexRecord> cnIndexDBCursor;

  private boolean draftCompat = false;
  /**
@@ -591,7 +588,7 @@
    {
      // Request filter DOES NOT contain any start change number
      // So we'll generate from the oldest change number in the CNIndexDB
      final CNIndexRecord oldestRecord = cnIndexDB.getOldestRecord();
      final ChangeNumberIndexRecord oldestRecord = cnIndexDB.getOldestRecord();
      if (oldestRecord == null)
      { // DB is empty or closed
        isEndOfCNIndexDBReached = true;
@@ -606,8 +603,9 @@
    // Request filter DOES contain a startChangeNumber

    // Read the CNIndexDB to see whether it contains startChangeNumber
    DBCursor<CNIndexRecord> cursor = cnIndexDB.getCursorFrom(startChangeNumber);
    final CNIndexRecord startRecord = cursor.getRecord();
    DBCursor<ChangeNumberIndexRecord> cursor =
        cnIndexDB.getCursorFrom(startChangeNumber);
    final ChangeNumberIndexRecord startRecord = cursor.getRecord();
    if (startRecord != null)
    {
      // found the provided startChangeNumber, let's return it
@@ -632,7 +630,7 @@
    if (startChangeNumber < oldestChangeNumber)
    {
      cursor = cnIndexDB.getCursorFrom(oldestChangeNumber);
      final CNIndexRecord oldestRecord = cursor.getRecord();
      final ChangeNumberIndexRecord oldestRecord = cursor.getRecord();
      if (oldestRecord == null)
      {
        // This should not happen
@@ -648,7 +646,7 @@
    {
      // startChangeNumber is between oldest and potential newest and has never
      // been returned yet
      final CNIndexRecord newestRecord = cnIndexDB.getNewestRecord();
      final ChangeNumberIndexRecord newestRecord = cnIndexDB.getNewestRecord();
      if (newestRecord == null)
      {
        isEndOfCNIndexDBReached = true;
@@ -667,10 +665,12 @@
    throw new DirectoryException(ResultCode.SUCCESS, Message.raw(""));
  }

  private DBCursor<CNIndexRecord> getCursorFrom(ChangeNumberIndexDB cnIndexDB,
      long startChangeNumber) throws ChangelogException
  private DBCursor<ChangeNumberIndexRecord> getCursorFrom(
      ChangeNumberIndexDB cnIndexDB, long startChangeNumber)
      throws ChangelogException
  {
    DBCursor<CNIndexRecord> cursor = cnIndexDB.getCursorFrom(startChangeNumber);
    DBCursor<ChangeNumberIndexRecord> cursor =
        cnIndexDB.getCursorFrom(startChangeNumber);
    if (cursor.getRecord() == null)
    {
      close(cursor);
@@ -1320,7 +1320,7 @@
        return true;
      }

      final CNIndexRecord currentRecord = cnIndexDBCursor.getRecord();
      final ChangeNumberIndexRecord currentRecord = cnIndexDBCursor.getRecord();
      final CSN csnFromCNIndexDB = currentRecord.getCSN();
      final DN dnFromCNIndexDB = currentRecord.getBaseDN();

@@ -1401,8 +1401,9 @@
  private void assignNewChangeNumberAndStore(ECLUpdateMsg change)
      throws ChangelogException
  {
    final CNIndexRecord record = new CNIndexRecord(previousCookie.toString(),
        change.getBaseDN(), change.getUpdateMsg().getCSN());
    final ChangeNumberIndexRecord record =
        new ChangeNumberIndexRecord(previousCookie.toString(),
            change.getBaseDN(), change.getUpdateMsg().getCSN());
    // store in CNIndexDB the pair
    // (change number of the current change, state before this change)
    change.setChangeNumber(

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

@@ -51,10 +51,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.CNIndexRecord;
import org.opends.server.replication.server.changelog.api.ChangeNumberIndexDB;
import org.opends.server.replication.server.changelog.api.ChangelogDB;
import org.opends.server.replication.server.changelog.api.ChangelogException;
import org.opends.server.replication.server.changelog.api.*;
import org.opends.server.replication.server.changelog.je.JEChangelogDB;
import org.opends.server.types.*;
import org.opends.server.util.LDIFReader;
@@ -1399,8 +1396,8 @@
    try
    {
      final ChangeNumberIndexDB cnIndexDB = getChangeNumberIndexDB();
      final CNIndexRecord oldestRecord = cnIndexDB.getOldestRecord();
      final CNIndexRecord newestRecord = cnIndexDB.getNewestRecord();
      final ChangeNumberIndexRecord oldestRecord = cnIndexDB.getOldestRecord();
      final ChangeNumberIndexRecord newestRecord = cnIndexDB.getNewestRecord();

      boolean dbEmpty = true;
      long oldestChangeNumber = 0;
@@ -1453,7 +1450,7 @@
        {
          // There are records in the CNIndexDB (so already returned to clients)
          // BUT
          // There is nothing related to this domain in the newest CNIndexRecord
          // There is nothing related to this domain in the newest record
          // (maybe this domain was disabled when this record was returned).
          // In that case, are counted the changes from the time of the most
          // recent change

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

@@ -29,10 +29,10 @@

/**
 * This class stores an index of all the changes seen by this server in the form
 * of {@link CNIndexRecord}s. The records are 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.
 * of {@link ChangeNumberIndexRecord}s. The records are 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"
@@ -53,22 +53,22 @@
  /**
   * Get the oldest record stored in this DB.
   *
   * @return Returns the oldest {@link CNIndexRecord} in this DB, null when the
   *         DB is empty or closed
   * @return Returns the oldest {@link ChangeNumberIndexRecord} in this DB, null
   *         when the DB is empty or closed
   * @throws ChangelogException
   *           if a database problem occurs.
   */
  CNIndexRecord getOldestRecord() throws ChangelogException;
  ChangeNumberIndexRecord getOldestRecord() throws ChangelogException;

  /**
   * Get the newest record stored in this DB.
   *
   * @return Returns the newest {@link CNIndexRecord} in this DB, null when the
   *         DB is empty or closed
   * @return Returns the newest {@link ChangeNumberIndexRecord} in this DB, null
   *         when the DB is empty or closed
   * @throws ChangelogException
   *           if a database problem occurs.
   */
  CNIndexRecord getNewestRecord() throws ChangelogException;
  ChangeNumberIndexRecord getNewestRecord() throws ChangelogException;

  /**
   * Add an update to the list of messages that must be saved to this DB managed
@@ -80,12 +80,12 @@
   * for lazily building the ChangeNumberIndexDB.
   *
   * @param record
   *          The {@link CNIndexRecord} to add to this DB.
   *          The {@link ChangeNumberIndexRecord} to add to this DB.
   * @return the change number associated to this record on adding to this DB
   * @throws ChangelogException
   *           if a database problem occurs.
   */
  long addRecord(CNIndexRecord record) throws ChangelogException;
  long addRecord(ChangeNumberIndexRecord record) throws ChangelogException;

  /**
   * Generate a new {@link DBCursor} that allows to browse the db managed by
@@ -99,7 +99,7 @@
   * @throws ChangelogException
   *           if a database problem occurs.
   */
  DBCursor<CNIndexRecord> getCursorFrom(long startChangeNumber)
  DBCursor<ChangeNumberIndexRecord> getCursorFrom(long startChangeNumber)
      throws ChangelogException;

}

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

File was renamed from opends/src/server/org/opends/server/replication/server/changelog/api/CNIndexRecord.java
@@ -30,10 +30,10 @@
import org.opends.server.types.DN;

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

  /** This is the key used to store the rest of the . */
@@ -54,8 +54,8 @@
   * @param csn
   *          the replication CSN field
   */
  public CNIndexRecord(long changeNumber, String previousCookie, DN baseDN,
      CSN csn)
  public ChangeNumberIndexRecord(long changeNumber, String previousCookie,
      DN baseDN, CSN csn)
  {
    this.changeNumber = changeNumber;
    this.previousCookie = previousCookie;
@@ -72,9 +72,9 @@
   *          the baseDN
   * @param csn
   *          the replication CSN field
   * @see #CNIndexRecord(long, String, DN, CSN)
   * @see #ChangeNumberIndexRecord(long, String, DN, CSN)
   */
  public CNIndexRecord(String previousCookie, DN baseDN, CSN csn)
  public ChangeNumberIndexRecord(String previousCookie, DN baseDN, CSN csn)
  {
    this(0, previousCookie, baseDN, csn);
  }

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

@@ -33,8 +33,7 @@

import org.opends.messages.MessageBuilder;
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.replication.server.changelog.api.CNIndexRecord;
import org.opends.server.replication.server.changelog.api.ChangelogException;
import org.opends.server.replication.server.changelog.api.*;
import org.opends.server.types.DebugLogLevel;

import com.sleepycat.je.*;
@@ -83,9 +82,9 @@
   * Add a record to the database.
   *
   * @param record
   *          the provided {@link CNIndexRecord} to be stored.
   *          the provided {@link ChangeNumberIndexRecord} to be stored.
   */
  public void addRecord(CNIndexRecord record)
  public void addRecord(ChangeNumberIndexRecord record)
  {
    try
    {
@@ -220,7 +219,7 @@
   * @throws ChangelogException
   *           if a database problem occurred
   */
  public CNIndexRecord readFirstRecord() throws ChangelogException
  public ChangeNumberIndexRecord readFirstRecord() throws ChangelogException
  {
    try
    {
@@ -256,7 +255,7 @@
    }
  }

  private CNIndexRecord newCNIndexRecord(ReplicationDraftCNKey key,
  private ChangeNumberIndexRecord newCNIndexRecord(ReplicationDraftCNKey key,
      DatabaseEntry data) throws ChangelogException
  {
    return new DraftCNData(key.getChangeNumber(), data.getData()).getRecord();
@@ -297,7 +296,7 @@
   * @throws ChangelogException
   *           if a database problem occurred
   */
  public CNIndexRecord readLastRecord() throws ChangelogException
  public ChangeNumberIndexRecord readLastRecord() throws ChangelogException
  {
    try
    {
@@ -357,7 +356,7 @@
    private final Transaction txn;
    private final ReplicationDraftCNKey key;
    private final DatabaseEntry entry = new DatabaseEntry();
    private CNIndexRecord record;
    private ChangeNumberIndexRecord record;
    private boolean isClosed = false;


@@ -561,11 +560,12 @@
    }

    /**
     * Returns the {@link CNIndexRecord} at the current position of the cursor.
     * Returns the {@link ChangeNumberIndexRecord} at the current position of
     * the cursor.
     *
     * @return The current {@link CNIndexRecord}.
     * @return The current {@link ChangeNumberIndexRecord}.
     */
    public CNIndexRecord currentRecord()
    public ChangeNumberIndexRecord currentRecord()
    {
      if (isClosed)
      {

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

@@ -31,8 +31,7 @@

import org.opends.messages.Message;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.server.changelog.api.CNIndexRecord;
import org.opends.server.replication.server.changelog.api.ChangelogException;
import org.opends.server.replication.server.changelog.api.*;
import org.opends.server.types.DN;
import org.opends.server.types.DirectoryException;

@@ -50,7 +49,7 @@
  private static final long serialVersionUID = 1L;

  private long changeNumber;
  private CNIndexRecord record;
  private ChangeNumberIndexRecord record;

  /**
   * Creates a record to be stored in the DraftCNDB.
@@ -90,16 +89,16 @@
  }

  /**
   * Decode and returns a {@link CNIndexRecord}.
   * Decode and returns a {@link ChangeNumberIndexRecord}.
   *
   * @param changeNumber
   * @param data
   *          the provided byte array.
   * @return the decoded {@link CNIndexRecord}
   * @return the decoded {@link ChangeNumberIndexRecord}
   * @throws ChangelogException
   *           when a problem occurs.
   */
  private CNIndexRecord decodeData(long changeNumber, byte[] data)
  private ChangeNumberIndexRecord decodeData(long changeNumber, byte[] data)
      throws ChangelogException
  {
    try
@@ -107,7 +106,8 @@
      String stringData = new String(data, "UTF-8");
      String[] str = stringData.split(FIELD_SEPARATOR, 3);
      final DN baseDN = DN.decode(str[1]);
      return new CNIndexRecord(changeNumber, str[0], baseDN, new CSN(str[2]));
      final CSN csn = new CSN(str[2]);
      return new ChangeNumberIndexRecord(changeNumber, str[0], baseDN, csn);
    }
    catch (UnsupportedEncodingException e)
    {
@@ -124,11 +124,11 @@
  /**
   * Getter for the decoded record.
   *
   * @return the {@link CNIndexRecord} record.
   * @return the {@link ChangeNumberIndexRecord} record.
   * @throws ChangelogException
   *           when a problem occurs.
   */
  public CNIndexRecord getRecord() throws ChangelogException
  public ChangeNumberIndexRecord getRecord() throws ChangelogException
  {
    if (record == null)
      record = decodeData(changeNumber, getData());

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

@@ -123,8 +123,8 @@

    // DB initialization
    db = new DraftCNDB(dbenv);
    final CNIndexRecord oldestRecord = db.readFirstRecord();
    final CNIndexRecord newestRecord = db.readLastRecord();
    final ChangeNumberIndexRecord oldestRecord = db.readFirstRecord();
    final ChangeNumberIndexRecord newestRecord = db.readLastRecord();
    oldestChangeNumber = getChangeNumber(oldestRecord);
    newestChangeNumber = getChangeNumber(newestRecord);
    // initialization of the lastGeneratedChangeNumber from the DB content
@@ -147,7 +147,8 @@
    trimmingThread.start();
  }

  private long getChangeNumber(CNIndexRecord record) throws ChangelogException
  private long getChangeNumber(ChangeNumberIndexRecord record)
      throws ChangelogException
  {
    if (record != null)
    {
@@ -158,12 +159,13 @@

  /** {@inheritDoc} */
  @Override
  public long addRecord(CNIndexRecord record) throws ChangelogException
  public long addRecord(ChangeNumberIndexRecord record)
      throws ChangelogException
  {
    long changeNumber = nextChangeNumber();
    final CNIndexRecord newRecord =
        new CNIndexRecord(changeNumber, record.getPreviousCookie(), record
            .getBaseDN(), record.getCSN());
    final ChangeNumberIndexRecord newRecord =
        new ChangeNumberIndexRecord(changeNumber, record.getPreviousCookie(),
            record.getBaseDN(), record.getCSN());
    db.addRecord(newRecord);

    if (debugEnabled())
@@ -173,14 +175,14 @@

  /** {@inheritDoc} */
  @Override
  public CNIndexRecord getOldestRecord() throws ChangelogException
  public ChangeNumberIndexRecord getOldestRecord() throws ChangelogException
  {
    return db.readFirstRecord();
  }

  /** {@inheritDoc} */
  @Override
  public CNIndexRecord getNewestRecord() throws ChangelogException
  public ChangeNumberIndexRecord getNewestRecord() throws ChangelogException
  {
    return db.readLastRecord();
  }
@@ -221,7 +223,7 @@

  /** {@inheritDoc} */
  @Override
  public DBCursor<CNIndexRecord> getCursorFrom(long startChangeNumber)
  public DBCursor<ChangeNumberIndexRecord> getCursorFrom(long startChangeNumber)
      throws ChangelogException
  {
    return new JEChangeNumberIndexDBCursor(db, startChangeNumber);
@@ -355,7 +357,7 @@
            return;
          }

          final CNIndexRecord record = cursor.currentRecord();
          final ChangeNumberIndexRecord record = cursor.currentRecord();
          if (baseDNToClear != null && baseDNToClear.equals(record.getBaseDN()))
          {
            cursor.delete();
@@ -478,7 +480,7 @@
    {
      try
      {
        CNIndexRecord record =
        final ChangeNumberIndexRecord record =
            isFirst ? db.readFirstRecord() : db.readLastRecord();
        if (record != null)
        {

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

@@ -38,7 +38,8 @@
 * This class allows to iterate through the changes received from a given
 * LDAP Server Identifier.
 */
public class JEChangeNumberIndexDBCursor implements DBCursor<CNIndexRecord>
public class JEChangeNumberIndexDBCursor implements
    DBCursor<ChangeNumberIndexRecord>
{
  private static final DebugTracer TRACER = getTracer();
  private DraftCNDBCursor draftCNDbCursor;
@@ -62,7 +63,7 @@

  /** {@inheritDoc} */
  @Override
  public CNIndexRecord getRecord()
  public ChangeNumberIndexRecord getRecord()
  {
    try
    {

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

@@ -35,7 +35,7 @@
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.server.ReplServerFakeConfiguration;
import org.opends.server.replication.server.ReplicationServer;
import org.opends.server.replication.server.changelog.api.CNIndexRecord;
import org.opends.server.replication.server.changelog.api.ChangeNumberIndexRecord;
import org.opends.server.replication.server.changelog.api.ChangelogException;
import org.opends.server.replication.server.changelog.api.DBCursor;
import org.opends.server.types.DN;
@@ -86,16 +86,16 @@
      CSN[] csns = newCSNs(1, 0, 3);

      // Add records
      long cn1 = cnIndexDB.addRecord(new CNIndexRecord(value1, baseDN1, csns[0]));
                 cnIndexDB.addRecord(new CNIndexRecord(value2, baseDN2, csns[1]));
      long cn3 = cnIndexDB.addRecord(new CNIndexRecord(value3, baseDN3, csns[2]));
      long cn1 = addRecord(cnIndexDB, value1, baseDN1, csns[0]);
                 addRecord(cnIndexDB, value2, baseDN2, csns[1]);
      long cn3 = addRecord(cnIndexDB, value3, baseDN3, csns[2]);

      // The ChangeNumber should not get purged
      final long oldestCN = cnIndexDB.getOldestRecord().getChangeNumber();
      assertEquals(oldestCN, cn1);
      assertEquals(cnIndexDB.getNewestRecord().getChangeNumber(), cn3);

      DBCursor<CNIndexRecord> cursor = cnIndexDB.getCursorFrom(oldestCN);
      DBCursor<ChangeNumberIndexRecord> cursor = cnIndexDB.getCursorFrom(oldestCN);
      try
      {
        assertEqualTo(cursor.getRecord(), csns[0], baseDN1, value1);
@@ -131,7 +131,13 @@
    }
  }

  private void assertEqualTo(CNIndexRecord record, CSN csn, DN baseDN, String cookie)
  private long addRecord(JEChangeNumberIndexDB cnIndexDB, String cookie, DN baseDN, CSN csn)
      throws ChangelogException
  {
    return cnIndexDB.addRecord(new ChangeNumberIndexRecord(cookie, baseDN, csn));
  }

  private void assertEqualTo(ChangeNumberIndexRecord record, CSN csn, DN baseDN, String cookie)
  {
    assertEquals(record.getCSN(), csn);
    assertEquals(record.getBaseDN(), baseDN);
@@ -190,9 +196,9 @@
      CSN[] csns = newCSNs(1, 0, 3);

      // Add records
      long cn1 = cnIndexDB.addRecord(new CNIndexRecord(value1, baseDN1, csns[0]));
      long cn2 = cnIndexDB.addRecord(new CNIndexRecord(value2, baseDN2, csns[1]));
      long cn3 = cnIndexDB.addRecord(new CNIndexRecord(value3, baseDN3, csns[2]));
      long cn1 = addRecord(cnIndexDB, value1, baseDN1, csns[0]);
      long cn2 = addRecord(cnIndexDB, value2, baseDN2, csns[1]);
      long cn3 = addRecord(cnIndexDB, value3, baseDN3, csns[2]);

      // Checks
      assertEquals(cnIndexDB.getOldestRecord().getChangeNumber(), cn1);
@@ -205,7 +211,7 @@
      assertEquals(getPreviousCookie(cnIndexDB, cn2), value2);
      assertEquals(getPreviousCookie(cnIndexDB, cn3), value3);

      DBCursor<CNIndexRecord> cursor = cnIndexDB.getCursorFrom(cn1);
      DBCursor<ChangeNumberIndexRecord> cursor = cnIndexDB.getCursorFrom(cn1);
      assertCursorReadsInOrder(cursor, cn1, cn2, cn3);

      cursor = cnIndexDB.getCursorFrom(cn2);
@@ -239,7 +245,7 @@
  private String getPreviousCookie(JEChangeNumberIndexDB cnIndexDB,
      long changeNumber) throws Exception
  {
    DBCursor<CNIndexRecord> cursor = cnIndexDB.getCursorFrom(changeNumber);
    DBCursor<ChangeNumberIndexRecord> cursor = cnIndexDB.getCursorFrom(changeNumber);
    try
    {
      return cursor.getRecord().getPreviousCookie();
@@ -250,7 +256,7 @@
    }
  }

  private void assertCursorReadsInOrder(DBCursor<CNIndexRecord> cursor,
  private void assertCursorReadsInOrder(DBCursor<ChangeNumberIndexRecord> cursor,
      long... cns) throws ChangelogException
  {
    try