OPENDJ-1116 Introduce abstraction for the changelog DB

Jean-Noel Rouvignac

04.34.2013 ff3d7d233cd306e4fe50e2219fcdfc7f4ec7c920

OPENDJ-1116 Introduce abstraction for the changelog DB


ChangeNumberIndexDB.java:
All methods can now throw ChangelogException.


ReplicationServer.java:
Extracted method shutdownCNIndexDB().
Removed getFirstChangeNumber() and getLastChangeNumber().

DraftCNDbHandler.java:
Removed catch blocks and let the exception propagate.

ECLServerHandler.java:
Let the ChangelogException propagate.


ReplicaDBCursor.java:
Augmented the javadoc to explain the sort order. Follow up of r9504.

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

@@ -1347,9 +1347,11 @@
   *         provided oldestChange, <code>false</code> otherwise
   * @throws DirectoryException
   *           if any problem occur
   * @throws ChangelogException
   *           if a database problem occurs.
   */
  private boolean assignChangeNumber(final ECLUpdateMsg oldestChange)
      throws DirectoryException
      throws DirectoryException, ChangelogException
  {
    // We also need to check if the draftCNdb is consistent with
    // the changelogdb.
@@ -1451,7 +1453,7 @@
  }

  private void assignNewDraftCNAndStore(ECLUpdateMsg change)
      throws DirectoryException
      throws DirectoryException, ChangelogException
  {
    // generate a new change number and assign to this change
    change.setChangeNumber(replicationServer.getNewChangeNumber());

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

@@ -586,8 +586,7 @@
   * Enable the ECL access by creating a dedicated workflow element.
   * @throws DirectoryException when an error occurs.
   */
  public void enableECL()
  throws DirectoryException
  public void enableECL() throws DirectoryException
  {
    if (externalChangeLogWorkflowImpl!=null)
    {
@@ -661,8 +660,7 @@
    {
      Message message =
        NOTE_ERR_UNABLE_TO_ENABLE_ECL_VIRTUAL_ATTR.get(attrName, e.toString());
      throw new DirectoryException(ResultCode.OPERATIONS_ERROR,
          message, e);
      throw new DirectoryException(ResultCode.OPERATIONS_ERROR, message, e);
    }
  }

@@ -679,19 +677,14 @@
      // internalNetworkGroup?
      NetworkGroup internalNetworkGroup = NetworkGroup
          .getInternalNetworkGroup();
      internalNetworkGroup
          .deregisterWorkflow(externalChangeLogWorkflowID);
      internalNetworkGroup.deregisterWorkflow(externalChangeLogWorkflowID);

      // FIXME:ECL should the ECL Workflow be registered in adminNetworkGroup?
      NetworkGroup adminNetworkGroup = NetworkGroup
          .getAdminNetworkGroup();
      adminNetworkGroup
          .deregisterWorkflow(externalChangeLogWorkflowID);
      NetworkGroup adminNetworkGroup = NetworkGroup.getAdminNetworkGroup();
      adminNetworkGroup.deregisterWorkflow(externalChangeLogWorkflowID);

      NetworkGroup defaultNetworkGroup = NetworkGroup
          .getDefaultNetworkGroup();
      defaultNetworkGroup
          .deregisterWorkflow(externalChangeLogWorkflowID);
      NetworkGroup defaultNetworkGroup = NetworkGroup.getDefaultNetworkGroup();
      defaultNetworkGroup.deregisterWorkflow(externalChangeLogWorkflowID);

      eclwf.deregister();
      eclwf.finalizeWorkflow();
@@ -705,12 +698,27 @@
      eclwe.finalizeWorkflowElement();
    }

    shutdownCNIndexDB();
  }

  private void shutdownCNIndexDB()
  {
    synchronized (cnIndexDBLock)
    {
      if (cnIndexDB != null)
      {
        try
        {
        cnIndexDB.shutdown();
      }
        catch (ChangelogException ignored)
        {
          if (debugEnabled())
          {
            TRACER.debugCaught(DebugLogLevel.WARNING, ignored);
          }
        }
      }
    }
  }

@@ -1385,17 +1393,7 @@
          }
        }

        try
        {
          cnIndexDB.shutdown();
        }
        catch (Exception ignored)
        {
          if (debugEnabled())
          {
            TRACER.debugCaught(DebugLogLevel.WARNING, ignored);
          }
        }
        shutdownCNIndexDB();

        lastGeneratedChangeNumber = 0;
        cnIndexDB = null;
@@ -1639,7 +1637,7 @@
        if (cnIndexDB == null)
        {
          cnIndexDB = new DraftCNDbHandler(this, this.dbEnv);
          lastGeneratedChangeNumber = getLastChangeNumber();
          lastGeneratedChangeNumber = cnIndexDB.getLastChangeNumber();
        }
        return cnIndexDB;
      }
@@ -1654,40 +1652,6 @@
  }

  /**
   * Get the value of the first change number, 0 when db is empty.
   *
   * @return the first value.
   */
  public long getFirstChangeNumber()
  {
    synchronized (cnIndexDBLock)
    {
      if (cnIndexDB != null)
      {
        return cnIndexDB.getFirstChangeNumber();
      }
      return 0;
    }
  }

  /**
   * Get the value of the last change number, 0 when db is empty.
   *
   * @return the last value.
   */
  public long getLastChangeNumber()
  {
    synchronized (cnIndexDBLock)
    {
      if (cnIndexDB != null)
      {
        return cnIndexDB.getLastChangeNumber();
      }
      return 0;
    }
  }

  /**
   * Generate a new change number.
   *
   * @return The generated change number
@@ -1739,6 +1703,8 @@
    boolean dbEmpty = false;
    final ChangeNumberIndexDB cnIndexDB = getChangeNumberIndexDB();

    try
    {
    long firstChangeNumber = cnIndexDB.getFirstChangeNumber();
    Map<String, ServerState> domainsServerStateForLastCN = null;
    CSN csnForLastCN = null;
@@ -1758,8 +1724,8 @@
      String lastCNGenState = cnIndexDB.getPreviousCookie(lastChangeNumber);
      if (lastCNGenState != null && lastCNGenState.length() > 0)
      {
        domainsServerStateForLastCN =
            MultiDomainServerState.splitGenStateToServerStates(lastCNGenState);
          domainsServerStateForLastCN = MultiDomainServerState
              .splitGenStateToServerStates(lastCNGenState);
      }

      csnForLastCN = cnIndexDB.getCSN(lastChangeNumber);
@@ -1823,6 +1789,11 @@
    }
    return new long[] { firstChangeNumber, lastChangeNumber };
  }
    catch (ChangelogException e)
    {
      throw new DirectoryException(ResultCode.OPERATIONS_ERROR, e);
    }
  }

  /**
   * Returns the last (newest) cookie value.

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

@@ -49,8 +49,10 @@
   * @param changeNumber
   *          the provided change number.
   * @return the associated CSN, null when none.
   * @throws ChangelogException
   *           if a database problem occurs.
   */
  public CSN getCSN(long changeNumber);
  public CSN getCSN(long changeNumber) throws ChangelogException;

  /**
   * Get the baseDN associated to a provided change number.
@@ -58,8 +60,10 @@
   * @param changeNumber
   *          the provided change number.
   * @return the baseDN, null when none.
   * @throws ChangelogException
   *           if a database problem occurs.
   */
  public String getBaseDN(long changeNumber);
  public String getBaseDN(long changeNumber) throws ChangelogException;

  /**
   * Get the previous cookie associated to a provided change number.
@@ -67,22 +71,28 @@
   * @param changeNumber
   *          the provided change number.
   * @return the previous cookie, null when none.
   * @throws ChangelogException
   *           if a database problem occurs.
   */
  String getPreviousCookie(long changeNumber);
  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();
  long getFirstChangeNumber() throws ChangelogException;

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

  /**
   * Add an update to the list of messages that must be saved to this DB managed
@@ -99,8 +109,11 @@
   *          The associated baseDN.
   * @param csn
   *          The associated replication CSN.
   * @throws ChangelogException
   *           if a database problem occurs.
   */
  void add(long changeNumber, String previousCookie, String baseDN, CSN csn);
  void add(long changeNumber, String previousCookie, String baseDN, CSN csn)
      throws ChangelogException;

  /**
   * Generate a new {@link ChangeNumberIndexDBCursor} that allows to browse the
@@ -113,7 +126,7 @@
   *         this DBHandler and starting at the position defined by a given
   *         changeNumber.
   * @throws ChangelogException
   *           if a database problem happened.
   *           if a database problem occurs.
   */
  ChangeNumberIndexDBCursor getCursorFrom(long startChangeNumber)
      throws ChangelogException;
@@ -123,14 +136,16 @@
   *
   * @return <code>true</code> if this database is empty, <code>false</code>
   *         otherwise
   * @throws ChangelogException
   *           if a database problem occurs.
   */
  boolean isEmpty();
  boolean isEmpty() throws ChangelogException;

  /**
   * Clear the changes from this DB (from both memory cache and DB storage).
   *
   * @throws ChangelogException
   *           When an exception occurs while removing the changes from this DB.
   *           if a database problem occurs.
   */
  void clear() throws ChangelogException;

@@ -142,13 +157,16 @@
   *          The baseDN for which we want to remove all records from this DB,
   *          null means all.
   * @throws ChangelogException
   *           When an exception occurs while removing the changes from this DB.
   *           if a database problem occurs.
   */
  void clear(String baseDNToClear) throws ChangelogException;

  /**
   * Shutdown this DB.
   *
   * @throws ChangelogException
   *           if a database problem occurs.
   */
  void shutdown();
  void shutdown() throws ChangelogException;

}

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

@@ -33,6 +33,9 @@
/**
 * This cursor allows to iterate through the changes received from a given
 * replica (Directory Server) in the topology.
 * <p>
 * Instances of this class are sorted in the order defined by the CSN of the
 * current {@link UpdateMsg}, i.e. the cursor with the oldest CSN comes first.
 */
public interface ReplicaDBCursor extends Closeable, Comparable<ReplicaDBCursor>
{

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

@@ -516,7 +516,7 @@

  /** {@inheritDoc} */
  @Override
  public String getPreviousCookie(long changeNumber)
  public String getPreviousCookie(long changeNumber) throws ChangelogException
  {
    DraftCNDBCursor cursor = null;
    try
@@ -524,11 +524,6 @@
      cursor = db.openReadCursor(changeNumber);
      return cursor.currentValue();
    }
    catch(Exception e)
    {
      debugException("getValue", changeNumber, e);
      return null;
    }
    finally
    {
      close(cursor);
@@ -537,7 +532,7 @@

  /** {@inheritDoc} */
  @Override
  public CSN getCSN(long changeNumber)
  public CSN getCSN(long changeNumber) throws ChangelogException
  {
    DraftCNDBCursor cursor = null;
    try
@@ -545,11 +540,6 @@
      cursor = db.openReadCursor(changeNumber);
      return cursor.currentCSN();
    }
    catch(Exception e)
    {
      debugException("getCSN", changeNumber, e);
      return null;
    }
    finally
    {
      close(cursor);
@@ -558,7 +548,7 @@

  /**{@inheritDoc}*/
  @Override
  public String getBaseDN(long changeNumber)
  public String getBaseDN(long changeNumber) throws ChangelogException
  {
    DraftCNDBCursor cursor = null;
    try
@@ -566,25 +556,9 @@
      cursor = db.openReadCursor(changeNumber);
      return cursor.currentBaseDN();
    }
    catch(Exception e)
    {
      debugException("getBaseDN", changeNumber, e);
      return null;
    }
    finally
    {
      close(cursor);
    }
  }

  private void debugException(String methodName, long changeNumber, Exception e)
  {
    if (debugEnabled())
      TRACER.debugInfo("In DraftCNDbHandler." + methodName + "(), read: "
          + " key=" + changeNumber + " value returned is null"
          + " first="+ db.readFirstChangeNumber()
          + " last=" + db.readLastChangeNumber()
          + " count=" + db.count()
          + " exception " + e + " " + e.getMessage());
  }
}