OPENDJ-1116 Introduce abstraction for the changelog DB

Jean-Noel Rouvignac

30.48.2013 55e5dd13a9d3d646a3a426cf4f0790f0cded735e

OPENDJ-1116 Introduce abstraction for the changelog DB


Moved the lastGeneratedChangeNumber to the ChangeNumberIndexDB implementation because it is the responsibility of the DB to manage such field (think of it like relational databases's sequence numbers).


ReplicationServer.java:
Moved lastGeneratedChangeNumber + relevant code to DraftCNDbHandler.
Removed getNewChangeNumber(), replaced with TODO ChangeNumberIndexDB.getLastGeneratedChangeNumber()

DraftCNDbHandler.java
Moved lastGeneratedChangeNumber here and converted it to AtomicLong + moved relevant code here from ReplicationServer.
In getMonitorData(), extracted methods, createChangeNumberAttribute() and getChangeNumber().

ChangeNumberIndexDB.java:
Added nextChangeNumber() and getLastGeneratedChangeNumber().

ECLServerHandler.java:
Consequence of the changes to ReplicationServer.

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

@@ -1453,12 +1453,13 @@
  private void assignNewChangeNumberAndStore(ECLUpdateMsg change)
      throws DirectoryException, ChangelogException
  {
    // generate a new change number and assign to this change
    change.setChangeNumber(replicationServer.getNewChangeNumber());
    ChangeNumberIndexDB cnIndexDB = replicationServer.getChangeNumberIndexDB();

    change.setChangeNumber(cnIndexDB.nextChangeNumber());

    // store in CNIndexDB the pair
    // (change number of the current change, state before this change)
    replicationServer.getChangeNumberIndexDB().addRecord(new CNIndexRecord(
    cnIndexDB.addRecord(new CNIndexRecord(
        change.getChangeNumber(),
        previousCookie.toString(),
        change.getBaseDN(),

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

@@ -150,13 +150,6 @@
   */
  private ChangeNumberIndexDB cnIndexDB;

  /**
   * The last value generated of the change number.
   * <p>
   * Guarded by cnIndexDBLock
   **/
  private long lastGeneratedChangeNumber = 0;

  /** Used for protecting {@link ChangeNumberIndexDB} related state. */
  private final Object cnIndexDBLock = new Object();

@@ -1307,7 +1300,6 @@

        shutdownCNIndexDB();

        lastGeneratedChangeNumber = 0;
        cnIndexDB = null;
      }
    }
@@ -1528,11 +1520,6 @@
        if (cnIndexDB == null)
        {
          cnIndexDB = this.changelogDB.newChangeNumberIndexDB();
          final CNIndexRecord lastCNRecord = cnIndexDB.getLastRecord();
          // initialization of the lastGeneratedChangeNumber from the DB content
          // if DB is empty => last record does not exist => default to 0
          lastGeneratedChangeNumber =
              (lastCNRecord != null) ? lastCNRecord.getChangeNumber() : 0;
        }
        return cnIndexDB;
      }
@@ -1547,19 +1534,6 @@
  }

  /**
   * Generate a new change number.
   *
   * @return The generated change number
   */
  public long getNewChangeNumber()
  {
    synchronized (cnIndexDBLock)
    {
      return ++lastGeneratedChangeNumber;
    }
  }

  /**
   * Get first and last change number.
   *
   * @param crossDomainEligibleCSN
@@ -1683,8 +1657,9 @@
      {
        // The database was empty, just keep increasing numbers since last time
        // we generated one change number.
        firstChangeNumber += lastGeneratedChangeNumber;
        lastChangeNumber += lastGeneratedChangeNumber;
        long lastGeneratedCN = this.cnIndexDB.getLastGeneratedChangeNumber();
        firstChangeNumber += lastGeneratedCN;
        lastChangeNumber += lastGeneratedCN;
      }
      return new long[] { firstChangeNumber, lastChangeNumber };
    }

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

@@ -43,6 +43,19 @@
 */
public interface ChangeNumberIndexDB
{
  /**
   * Generates the next change number.
   *
   * @return The newly generated change number
   */
  long nextChangeNumber();

  /**
   * Returns the last generated change number.
   *
   * @return the lastGeneratedChangeNumber
   */
  long getLastGeneratedChangeNumber();

  /**
   * Get the record associated to a provided change number.

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

@@ -30,6 +30,7 @@
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;

import org.opends.messages.MessageBuilder;
@@ -82,6 +83,8 @@
   * even use it!
   */
  private long lastChangeNumber = NO_KEY;
  /** The last generated value for the change number. */
  private final AtomicLong lastGeneratedChangeNumber;
  private DbMonitorProvider dbMonitor = new DbMonitorProvider();
  private boolean shutdown = false;
  private boolean trimDone = false;
@@ -120,8 +123,14 @@

    // DB initialization
    db = new DraftCNDB(dbenv);
    firstChangeNumber = getChangeNumber(db.readFirstRecord());
    lastChangeNumber = getChangeNumber(db.readLastRecord());
    final CNIndexRecord firstRecord = db.readFirstRecord();
    final CNIndexRecord lastRecord = db.readLastRecord();
    firstChangeNumber = getChangeNumber(firstRecord);
    lastChangeNumber = getChangeNumber(lastRecord);
    // initialization of the lastGeneratedChangeNumber from the DB content
    // if DB is empty => last record does not exist => default to 0
    lastGeneratedChangeNumber =
        new AtomicLong((lastRecord != null) ? lastRecord.getChangeNumber() : 0);

    // Trimming thread
    thread = new DirectoryThread(this, "Replication DraftCN db");
@@ -165,6 +174,20 @@
    return db.readLastRecord();
  }

  /** {@inheritDoc} */
  @Override
  public long nextChangeNumber()
  {
    return lastGeneratedChangeNumber.incrementAndGet();
  }

  /** {@inheritDoc} */
  @Override
  public long getLastGeneratedChangeNumber()
  {
    return lastGeneratedChangeNumber.get();
  }

  /**
   * Get the number of changes.
   * @return Returns the number of changes.
@@ -416,40 +439,39 @@
    public List<Attribute> getMonitorData()
    {
      List<Attribute> attributes = new ArrayList<Attribute>();

      try
      {
        CNIndexRecord firstCNRecord = db.readFirstRecord();
        String firstCN = String.valueOf(firstCNRecord.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
      {
        CNIndexRecord lastCNRecord = db.readLastRecord();
        if (lastCNRecord != null)
        {
          String lastCN = String.valueOf(lastCNRecord.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(createChangeNumberAttribute(true));
      attributes.add(createChangeNumberAttribute(false));
      attributes.add(Attributes.create("count", Long.toString(count())));
      return attributes;
    }

    private Attribute createChangeNumberAttribute(boolean isFirst)
    {
      final String attributeName =
          isFirst ? "first-draft-changenumber" : "last-draft-changenumber";
      final String changeNumber = String.valueOf(getChangeNumber(isFirst));
      return Attributes.create(attributeName, changeNumber);
    }

    private long getChangeNumber(boolean isFirst)
    {
      try
      {
        CNIndexRecord record =
            isFirst ? db.readFirstRecord() : db.readLastRecord();
        if (record != null)
        {
          return record.getChangeNumber();
        }
      }
      catch (ChangelogException e)
      {
        if (debugEnabled())
          TRACER.debugCaught(DebugLogLevel.WARNING, e);
      }
      return 0;
    }

    /**
     * {@inheritDoc}
     */