OPENDJ-1116 Introduce abstraction for the changelog DB

Jean-Noel Rouvignac

05.32.2013 d2ec418d6010332f55828d53c613a3c3e9d03d9e

OPENDJ-1116 Introduce abstraction for the changelog DB


Fixed a few bugs (introduced in the previous commits?):
- All code must check that the result of calling ChangeNumberIndexDB.getFirstCNIndexData() and ChangeNumberIndexDB.getLastCNIndexData() are not null (if the DB is empty, or closed, etc.).
- In ReplicationServer.clearGenerationId(), the code could have a side effect if the last entry in the DB was cleared by this method: the lastGeneratedChangeNumber would be allowed to go back!


ChangeNumberIndexDB.java:
Removed getLastChangeNumber().

DraftCNDbHandler.java:
Inlined getLastChangeNumber() to the only remaining method caller ReplicationServer.getChangeNumberIndexDB().
Finally changed the implementation of isEmpty() to something less costly.

DraftCNDbHandlerTest.java:
Inlined getFirstChangeNumber() and getLastChangeNumber().


ReplicationServer.java:
In clearGenerationId(), removed the code changing the value of lastGeneratedChangeNumber: lastGeneratedChangeNumber is not allowed to go backward.
In getChangeNumberIndexDB(), inlined here the code of DraftCNDbHandler.getLastChangeNumber() + removed useless use of MessageBuilder.
In getECLChangeNumberLimits(), added a null check for the result of getLastCNIndexData().

replication.properties:
Added a new error message.
Changed MILD_ERR_DRAFT_CHANGENUMBER_DATABASE_173 to MILD_ERR_CHANGENUMBER_DATABASE_173.


ECLServerHandler.java:
In findCookie(), replaced the DB empty checks by null checks on the return of getFirstCNIndexData() and getLastCNIndexData() since this is purely equivalent.

 opends/src/messages/messages/replication.properties

@@ -310,8 +310,8 @@
 domain %s from server %s to all other servers of the topology is forbidden as \
 the source server has some fractional configuration : only fractional servers \
 in a replicated topology does not make sense
MILD_ERR_DRAFT_CHANGENUMBER_DATABASE_173=An error occurred when accessing the \
 database of the draft change number : %s
MILD_ERR_CHANGENUMBER_DATABASE_173=An error occurred when accessing the \
 change number database : %s
SEVERE_ERR_INITIALIZATION_FAILED_NOCONN_174=The initialization failed because \
 the domain %s is not connected to a replication server
SEVERE_ERR_FRACTIONAL_COULD_NOT_RETRIEVE_CONFIG_175=Could not retrieve the \
@@ -492,3 +492,7 @@
SEVERE_ERR_REPLICATIONDB_CANNOT_PROCESS_CHANGE_RECORD_215=Replication server RS(%d) \
 failed to parse change record with changenumber %s from the database. Error: %s
SEVERE_ERR_SESSION_STARTUP_INTERRUPTED_216=%s was interrupted in the startup phase
MILD_ERR_READING_FIRST_THEN_LAST_IN_CHANGENUMBER_DATABASE_217=An error occurred \
 when accessing the change number database: impossible to read the last record \
 after having successfully read the first. Database might have been cleaned or \
 closed between successive reads.

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

@@ -571,16 +571,13 @@
    {
      // Request filter DOES NOT contain any first change number
      // So we'll generate from the first change number in the DraftCNdb
      if (cnIndexDB.isEmpty())
      {
        // FIXME JNR if we find a way to make draftCNDb.isEmpty() a non costly
        // operation, then I think we can move this check to the top of this
        // method
      final CNIndexData firstCNData = cnIndexDB.getFirstCNIndexData();
      if (firstCNData == null)
      { // DB is empty or closed
        isEndOfCNIndexDBReached = true;
        return null;
      }

      final CNIndexData firstCNData = cnIndexDB.getFirstCNIndexData();
      final long firstChangeNumber = firstCNData.getChangeNumber();
      final String crossDomainStartState = firstCNData.getPreviousCookie();
      cnIndexDBCursor = cnIndexDB.getCursorFrom(firstChangeNumber);
@@ -630,13 +627,13 @@
    {
      // startChangeNumber is between first and potential last and has never
      // been returned yet
      if (cnIndexDB.isEmpty())
      final CNIndexData lastCNData = cnIndexDB.getLastCNIndexData();
      if (lastCNData == null)
      {
        isEndOfCNIndexDBReached = true;
        return null;
      }

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

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

@@ -914,18 +914,6 @@
            TRACER.debugCaught(DebugLogLevel.WARNING, ignored);
          }
        }

        try
        {
          lastGeneratedChangeNumber = cnIndexDB.getLastChangeNumber();
        }
        catch (Exception ignored)
        {
          if (debugEnabled())
          {
            TRACER.debugCaught(DebugLogLevel.WARNING, ignored);
          }
        }
      }
    }
  }
@@ -1638,16 +1626,19 @@
        if (cnIndexDB == null)
        {
          cnIndexDB = new DraftCNDbHandler(this, this.dbEnv);
          lastGeneratedChangeNumber = cnIndexDB.getLastChangeNumber();
          final CNIndexData lastCNData = cnIndexDB.getLastCNIndexData();
          // initialization of the lastGeneratedChangeNumebr from the DB content
          // if DB is empty => lastCNData is null => Default to 0
          lastGeneratedChangeNumber =
              (lastCNData != null) ? lastCNData.getChangeNumber() : 0;
        }
        return cnIndexDB;
      }
      catch (Exception e)
      {
        TRACER.debugCaught(DebugLogLevel.ERROR, e);
        MessageBuilder mb = new MessageBuilder();
        mb.append(ERR_DRAFT_CHANGENUMBER_DATABASE.get(""));
        throw new DirectoryException(OPERATIONS_ERROR, mb.toMessage(), e);
        Message message = ERR_CHANGENUMBER_DATABASE.get(e.getMessage());
        throw new DirectoryException(OPERATIONS_ERROR, message, e);
      }
    }
  }
@@ -1715,6 +1706,14 @@
      String domainForLastCN = null;
      if (firstCNData != null)
      {
        if (lastCNData == null)
        {
          // Edge case: DB was cleaned or closed in between call to getFirst*()
          // and getLast*(). The only remaining solution is to fail fast.
          throw new ChangelogException(
              ERR_READING_FIRST_THEN_LAST_IN_CHANGENUMBER_DATABASE.get());
        }

        dbEmpty = false;
        firstChangeNumber = firstCNData.getChangeNumber();
        lastChangeNumber = lastCNData.getChangeNumber();

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

@@ -72,15 +72,6 @@
  CNIndexData getLastCNIndexData() 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() throws ChangelogException;

  /**
   * Add an update to the list of messages that must be saved to this DB managed
   * by this DB.
   * <p>

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

@@ -169,18 +169,6 @@
    return db.readLastCNIndexData();
  }

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

  /**
   * Get the number of changes.
   * @return Returns the number of changes.
@@ -190,26 +178,11 @@
    return db.count();
  }

  /**
   * {@inheritDoc}
   * <p>
   * FIXME Find a way to implement this method in a more efficient manner.
   * {@link com.sleepycat.je.Database#count()} javadoc mentions:
   * <blockquote>Note that this method does scan a significant portion of the
   * database and should be considered a fairly expensive
   * operation.</blockquote>
   * <p>
   * It could be faster to:
   * <ul>
   * <li>open a cursor, check if the next entry exits, then close the cursor
   * </li>
   * <li>call <code>db.readFirstChangeNumber() != 0</code></li>
   * </ul>
   */
  /** {@inheritDoc} */
  @Override
  public boolean isEmpty()
  public boolean isEmpty() throws ChangelogException
  {
    return count() == 0;
    return getLastCNIndexData() == null;
  }

  /**

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

@@ -37,7 +37,6 @@
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;
@@ -108,9 +107,9 @@
      handler.add(new CNIndexData(cn3, value3, baseDN3, csn3));

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

      DraftCNDBCursor dbc = handler.getReadCursor(firstChangeNumber);
      try
@@ -134,12 +133,13 @@
      handler.setPurgeDelay(100);

      // Check the db is cleared.
      while (handler.count() != 0)
      while (!handler.isEmpty())
      {
        Thread.sleep(200);
      }
      assertEquals(getFirstChangeNumber(handler), 0);
      assertEquals(getLastChangeNumber(handler), 0);
      assertNull(handler.getFirstCNIndexData());
      assertNull(handler.getLastCNIndexData());
      assertEquals(handler.count(), 0);
    }
    finally
    {
@@ -235,10 +235,11 @@
      Thread.sleep(500);

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

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

      assertEquals(getPreviousCookie(handler, cn1), value1);
      assertEquals(getPreviousCookie(handler, cn2), value2);
@@ -256,9 +257,10 @@
      handler.clear();

      // Check the db is cleared.
      assertEquals(getFirstChangeNumber(handler), 0);
      assertEquals(getLastChangeNumber(handler), 0);
      assertNull(handler.getFirstCNIndexData());
      assertNull(handler.getLastCNIndexData());
      assertEquals(handler.count(), 0);
      assertTrue(handler.isEmpty());
    }
    finally
    {
@@ -272,16 +274,6 @@
    }
  }

  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);