From 6e14a8394d193af0fa32b83d3cc424787d41eadd Mon Sep 17 00:00:00 2001
From: Jean-Noel Rouvignac <jean-noel.rouvignac@forgerock.com>
Date: Thu, 26 Jun 2014 13:19:57 +0000
Subject: [PATCH] OPENDJ-1453 (CR-3870) Replica offline messages should be synced with updates
---
opends/src/server/org/opends/server/replication/server/changelog/file/FileChangelogDB.java | 26 ++++++++++++++++++++++++--
1 files changed, 24 insertions(+), 2 deletions(-)
diff --git a/opends/src/server/org/opends/server/replication/server/changelog/file/FileChangelogDB.java b/opends/src/server/org/opends/server/replication/server/changelog/file/FileChangelogDB.java
index b83a4f1..7c3163b 100644
--- a/opends/src/server/org/opends/server/replication/server/changelog/file/FileChangelogDB.java
+++ b/opends/src/server/org/opends/server/replication/server/changelog/file/FileChangelogDB.java
@@ -45,6 +45,7 @@
import org.opends.server.replication.server.changelog.api.*;
import org.opends.server.replication.server.changelog.je.ChangeNumberIndexer;
import org.opends.server.replication.server.changelog.je.CompositeDBCursor;
+import org.opends.server.replication.server.changelog.je.ReplicaOfflineCursor;
import org.opends.server.types.DN;
import org.opends.server.types.DebugLogLevel;
import org.opends.server.util.StaticUtils;
@@ -388,7 +389,7 @@
{
if (debugEnabled())
{
- TRACER.debugError("clear the FileChangelogDB");
+ TRACER.debugInfo("clear the FileChangelogDB");
}
if (!dbDirectory.exists())
{
@@ -631,18 +632,39 @@
throws ChangelogException
{
final Set<Integer> serverIds = getDomainMap(baseDN).keySet();
+ final ChangelogState state = replicationEnv.readChangelogState();
final Map<DBCursor<UpdateMsg>, Void> cursors = new HashMap<DBCursor<UpdateMsg>, Void>(serverIds.size());
for (int serverId : serverIds)
{
// get the last already sent CSN from that server to get a cursor
final CSN lastCSN = startAfterServerState != null ? startAfterServerState.getCSN(serverId) : null;
- cursors.put(getCursorFrom(baseDN, serverId, lastCSN), null);
+ final DBCursor<UpdateMsg> replicaDBCursor = getCursorFrom(baseDN, serverId, lastCSN);
+ final CSN offlineCSN = getOfflineCSN(state, baseDN, serverId, startAfterServerState);
+ cursors.put(new ReplicaOfflineCursor(replicaDBCursor, offlineCSN), null);
}
// recycle exhausted cursors,
// because client code will not manage the cursors itself
return new CompositeDBCursor<Void>(cursors, true);
}
+ private CSN getOfflineCSN(final ChangelogState state, DN baseDN, int serverId,
+ ServerState startAfterServerState)
+ {
+ final List<CSN> domain = state.getOfflineReplicas().get(baseDN);
+ if (domain != null)
+ {
+ for (CSN offlineCSN : domain)
+ {
+ if (serverId == offlineCSN.getServerId()
+ && !startAfterServerState.cover(offlineCSN))
+ {
+ return offlineCSN;
+ }
+ }
+ }
+ return null;
+ }
+
/** {@inheritDoc} */
@Override
public DBCursor<UpdateMsg> getCursorFrom(final DN baseDN, final int serverId, final CSN startAfterCSN)
--
Gitblit v1.10.0