From 030cd9b2dbc0d7016114f75e8b93190e34d1e213 Mon Sep 17 00:00:00 2001
From: Nicolas Capponi <nicolas.capponi@forgerock.com>
Date: Fri, 28 Nov 2014 14:11:01 +0000
Subject: [PATCH] OPENDJ-1611 CR-5492 Filter replicaIds when filling late queue in MessageHandler
---
opends/src/server/org/opends/server/replication/server/MessageHandler.java | 23 ++++++++++++++++++-----
1 files changed, 18 insertions(+), 5 deletions(-)
diff --git a/opends/src/server/org/opends/server/replication/server/MessageHandler.java b/opends/src/server/org/opends/server/replication/server/MessageHandler.java
index f3366b2..ac7ce11 100644
--- a/opends/src/server/org/opends/server/replication/server/MessageHandler.java
+++ b/opends/src/server/org/opends/server/replication/server/MessageHandler.java
@@ -28,6 +28,7 @@
import java.util.ArrayList;
import java.util.List;
+import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import org.opends.messages.Message;
@@ -255,12 +256,16 @@
* Get the next update that must be sent to the consumer
* from the message queue or from the database.
*
- * @param synchronous specifies what to do when the queue is empty.
+ * @param connectedReplicaIds
+ * Ids of replicas to accept when returning a message from
+ * the late queue.
+ * @param synchronous specifies what to do when the queue is empty.
* when true, the method blocks; when false the method return null.
+ *
* @return The next update that must be sent to the consumer.
* null when synchronous is false and queue is empty.
*/
- protected UpdateMsg getNextMessage(boolean synchronous)
+ protected UpdateMsg getNextMessage(Set<Integer> connectedReplicaIds, boolean synchronous)
{
while (activeConsumer)
{
@@ -293,7 +298,7 @@
* restart as usual
* load this change on the delayList
*/
- fillLateQueue();
+ fillLateQueue(connectedReplicaIds);
if (lateQueue.isEmpty())
{
// we could not find any messages in the changelog
@@ -387,7 +392,11 @@
return null;
}
- private void fillLateQueue()
+ /**
+ * Fills the late queue with the most recent changes, accepting only the
+ * messages from provided replica ids.
+ */
+ private void fillLateQueue(Set<Integer> connectedReplicaIds)
{
DBCursor<UpdateMsg> cursor = null;
try
@@ -395,7 +404,11 @@
cursor = replicationServerDomain.getCursorFrom(serverState);
while (cursor.next() && isLateQueueBelowThreshold())
{
- lateQueue.add(cursor.getRecord());
+ final UpdateMsg record = cursor.getRecord();
+ if (connectedReplicaIds.contains(record.getCSN().getServerId()))
+ {
+ lateQueue.add(record);
+ }
}
}
catch (ChangelogException e)
--
Gitblit v1.10.0