OPENDJ-1116 Introduce abstraction for the changelog DB

Jean-Noel Rouvignac

30.48.2013 acbee86d9d0116af38fde530023a9ed5cc685270

 opendj-sdk/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(),

 opendj-sdk/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 };
    }

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

 opendj-sdk/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,38 +439,37 @@
    public List<Attribute> getMonitorData()
    {
      List<Attribute> attributes = new ArrayList<Attribute>();
      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 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)
        CNIndexRecord record =
            isFirst ? db.readFirstRecord() : db.readLastRecord();
        if (record != null)
        {
          String lastCN = String.valueOf(lastCNRecord.getChangeNumber());
          attributes.add(Attributes.create("last-draft-changenumber", lastCN));
          return record.getChangeNumber();
        }
      }
      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;
      return 0;
    }

    /**