OPENDJ-3337 Make sure ReplicationOffline messages are not repeatedly return...

Fabio Pistolesi

11.16.2016 7160f59040db3b159e1d73d9ba58055e8806163d

OPENDJ-3337 Make sure ReplicationOffline messages are not repeatedly returned by cursors.

ReplicationOffline messages need special treatment, as they do not update server states.
Prevent returning already processed messages, especially when they're the last message for the domain.
Moreover set a server as following the replication when there are no new messages to send.

 opendj-server-legacy/src/main/java/org/opends/server/replication/server/MessageHandler.java

@@ -304,6 +304,13 @@
          synchronized (msgQueue) // TODO JNR why synchronize(msgQueue) here?
          {
            msg = lateQueue.removeFirst();
            // By default a server is always not following. A weird case where messages not representing
            // an operation may happen, making the late queue repeatedly fill and be emptied without ever
            // getting the server out of state "not following".
            if (lateQueue.isEmpty() && msgQueue.isEmpty())
            {
              following = true;
            }
          }
          if (updateServerState(msg))
          {

 opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java

@@ -423,7 +423,9 @@
      }
      return true;
    }
    return false;
    // Replica offline messages should not get to a connected DS, they are meant to be
    // exchanged only between RSes
    return updateMsg instanceof ReplicaOfflineMsg;
  }

  private PreparedAssuredInfo getPreparedAssuredInfo(UpdateMsg updateMsg,

 opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/ReplicaCursor.java

@@ -11,7 +11,7 @@
 * Header, with the fields enclosed by brackets [] replaced by your own identifying
 * information: "Portions Copyright [year] [name of copyright owner]".
 *
 * Copyright 2014-2015 ForgeRock AS.
 * Copyright 2014-2016 ForgeRock AS.
 */
package org.opends.server.replication.server.changelog.file;

@@ -74,8 +74,20 @@
   */
  public void setOfflineCSN(CSN offlineCSN)
  {
    this.replicaOfflineMsg.set(
        offlineCSN != null ? new ReplicaOfflineMsg(offlineCSN) : null);
    if (offlineCSN != null)
    {
      ReplicaOfflineMsg prevOfflineMsg = this.replicaOfflineMsg.get();
      if (prevOfflineMsg == null || prevOfflineMsg.getCSN().isOlderThan(offlineCSN))
      {
        // Do not spin if the the message for this replica has been changed. Either a newer
        // message has arrived or the next cursor iteration will pick it up.
        this.replicaOfflineMsg.compareAndSet(prevOfflineMsg, new ReplicaOfflineMsg(offlineCSN));
      }
    }
    else
    {
      this.replicaOfflineMsg.set(null);
    }
  }

  /** {@inheritDoc} */