OPENDJ-1482 : SEVERE_ERROR (cursor still opened on the log) when running re...

Nicolas Capponi

28.06.2014 46992e79c3b43e79c556f747cab12f93614aeae0

OPENDJ-1482 : SEVERE_ERROR (cursor still opened on the log) when running
replication topology with file-based changelog

The root cause is that records may be appended more than once to the 
changelog and in an order that is breaking the key ordering. 

Update Log.java to reject any append of a record that has a key that would
break the ordering of records in the log (ie, key lower than or equal to the last
key appended).

 opends/src/server/org/opends/server/replication/server/changelog/file/Log.java

@@ -25,9 +25,9 @@
 */
package org.opends.server.replication.server.changelog.file;

import static org.opends.messages.ReplicationMessages.*;
import static org.opends.server.loggers.debug.DebugLogger.*;
import static org.opends.server.util.StaticUtils.*;
import static org.opends.messages.ReplicationMessages.*;

import java.io.Closeable;
import java.io.File;
@@ -47,6 +47,9 @@
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.forgerock.util.Reject;
import org.opends.messages.Category;
import org.opends.messages.Message;
import org.opends.messages.Severity;
import org.opends.server.loggers.ErrorLogger;
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.replication.server.changelog.api.ChangelogException;
@@ -153,6 +156,13 @@
  private final TreeMap<K, LogFile<K, V>> logFiles = new TreeMap<K, LogFile<K, V>>();

  /**
   * The last key appended to the log. In order to keep the ordering of the keys
   * in the log, any attempt to append a record with a key lower or equal to
   * this key will silently fail.
   */
  private K lastAppendedKey;

  /**
   * The list of non-empty cursors opened on this log. Opened cursors may have
   * to be updated when rotating the head log file.
   */
@@ -354,6 +364,12 @@
      {
        return;
      }
      if (recordIsBreakingKeyOrdering(record))
      {
        ErrorLogger.logError(Message.raw(Category.SYNC, Severity.NOTICE,
            "Rejecting append to log '%s' for record: [%s]", logPath.getPath(), record.toString()));
        return;
      }
      LogFile<K, V> headLogFile = getHeadLogFile();
      if (headLogFile.getSizeInBytes() > sizeLimitPerLogFileInBytes)
      {
@@ -363,6 +379,7 @@
        headLogFile = getHeadLogFile();
      }
      headLogFile.append(record);
      lastAppendedKey = record.getKey();
    }
    finally
    {
@@ -371,6 +388,15 @@
  }

  /**
   * Indicates if the provided record has a key that would break the key
   * ordering in the log.
   */
  private boolean recordIsBreakingKeyOrdering(final Record<K, V> record)
  {
    return lastAppendedKey != null && record.getKey().compareTo(lastAppendedKey) <= 0;
  }

  /**
   * Synchronize all records added with the file system, ensuring that records
   * are effectively persisted.
   * <p>
@@ -814,8 +840,10 @@

  private void openHeadLogFile() throws ChangelogException
  {
    logFiles.put(recordParser.getMaxKey(),
        LogFile.newAppendableLogFile(new File(logPath,  HEAD_LOG_FILE_NAME), recordParser));
    final LogFile<K, V> head = LogFile.newAppendableLogFile(new File(logPath,  HEAD_LOG_FILE_NAME), recordParser);
    Record<K,V> newestRecord = head.getNewestRecord();
    lastAppendedKey = newestRecord == null ? null : newestRecord.getKey();
    logFiles.put(recordParser.getMaxKey(), head);
  }

  private void openReadOnlyLogFile(final File logFilePath) throws ChangelogException