From 5154f34503981b2f96755b1b6da9b425bb20ca7f Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Thu, 05 Nov 2026 14:32:12 +0000
Subject: [PATCH] [#1269] Preserve replay order across duplicate replication deliveries (#1270)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java | 186 +++++-
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/DuplicateReplayTest.java | 746 +++++++++++++++++++++++++++++
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/NamingConflictTest.java | 11
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChange.java | 52 ++
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java | 62 +
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java | 87 ++-
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/UpdateToReplay.java | 37 +
opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java | 6
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java | 272 ++++++++++
9 files changed, 1,351 insertions(+), 108 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java
index f14c25e..44d7bc9 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java
@@ -1147,7 +1147,7 @@
*/
pendingChanges = new PendingChanges(getGenerator(), this,
new ShutdownSyncAnnouncer(dsrsShutdownSync, getBaseDN()));
- remotePendingChanges = new RemotePendingChanges(getServerState());
+ remotePendingChanges = new RemotePendingChanges(getServerState(), getServerId());
// listen for changes on the configuration
configuration.addChangeListener(this);
@@ -2986,9 +2986,8 @@
* Marks the specified message as the one currently processed by a replay thread.
*
* @param msg the message being processed
- * @return {@code false} if the change is not pending anymore, which happens when the
- * session was restarted after a failed replay while this message was waiting
- * in the replay queue: it is sent again, so this copy must not be replayed
+ * @return {@code false} if the change is already committed, the domain has forgotten
+ * it, the delivery was cancelled, or a retry has replaced this stale delivery
*/
boolean markInProgress(LDAPUpdateMsg msg)
{
@@ -4842,13 +4841,9 @@
* Asks for a session restart the way a replay thread on its way out does, without
* running it.
* <p>
- * Only there for the tests, which have no other way to leave a request standing at a
- * time of their choosing without a change released for it: the one a replay thread makes
- * on a domain whose session has no owner is made and run in one go, and one made under
- * an owner stands for a change which is listed, and which the owner forgets or the
- * checkpointer has delivered again.
+ * Also used when a queue hand-off fails: the listener or replay thread must not
+ * synchronously restart the session from its failure cleanup.
*/
- @VisibleForTesting
public void requestSessionRestart()
{
sessionRestarts.request(SessionRestart.NOW);
@@ -5831,6 +5826,7 @@
state.clearInMemory();
state.loadState(strict);
bringAStateWithNoChangeOfThisReplicaUpToTheData(strict);
+ remotePendingChanges.resetReplayState();
getGenerator().adjust(state.getMaxCSN(getServerId()));
forgetTheLocalChangesTheDataLacks(strict);
@@ -8189,24 +8185,30 @@
{
LDAPUpdateMsg msg = (LDAPUpdateMsg) updateMsg;
- // Put the UpdateMsg in the RemotePendingChanges list.
- if (!remotePendingChanges.putRemoteUpdate(msg))
+ final UpdateToReplay updateToReplay = new UpdateToReplay(msg, this);
+ final RemotePendingChanges.Admission admission = remotePendingChanges.putRemoteUpdate(updateToReplay);
+ if (admission == RemotePendingChanges.Admission.QUEUED_DUPLICATE)
+ {
+ // The queued copy completes the current session's delivery after replay, including
+ // its error. An immediate completion here would grant extra credit and an early ack.
+ incProcessedUpdates();
+ return false;
+ }
+ if (admission == RemotePendingChanges.Admission.DUPLICATE)
{
/*
- * A replay thread already owns this change, so this delivery is a duplicate:
- * ignore it. This happens when there are uncommitted changes in the queue and
- * session failover occurs causing a recovery of all changes since the current
- * committed server state. See OPENDJ-1115.
+ * This change is already applied, owned by a replay thread, or committed
+ * behind another pending change. Session failover can deliver any of these again
+ * (OPENDJ-1115 and issue #1269).
*
- * The copy which is already listed owns the change: it is the one which records
- * it in the ServerState once it really has been replayed. Report this delivery
+ * The accepted copy is the one which records the change in the ServerState once
+ * it really has been replayed, if it has not already done so. Report this delivery
* as done - the window and the ack are per delivery - but as handled
* asynchronously, so that the listener does not push the CSN to the ServerState
* over a change which is still being replayed or is failing (issue #889).
*
- * A change whose replay failed is not owned by anyone anymore, so this is not the
- * path it takes: putRemoteUpdate() takes this delivery over from the one which
- * failed and has it replayed again.
+ * A released change has neither an owner nor a queued delivery, so this is not
+ * the path it takes: putRemoteUpdate() accepts its next delivery for replay.
*/
if (logger.isTraceEnabled())
{
@@ -8220,22 +8222,45 @@
// Put update message into the replay queue
// (block until some place in the queue is available)
- final UpdateToReplay updateToReplay = new UpdateToReplay(msg, this);
- while (!isListenerShuttingDown())
+ boolean queued = false;
+ boolean handoffFailed = true;
+ try
{
- // loop until we can offer to the queue or shutdown was initiated
- try
+ while (!isListenerShuttingDown())
{
- if (updateToReplayQueue.offer(updateToReplay, 1, TimeUnit.SECONDS))
+ // loop until we can offer to the queue or shutdown was initiated
+ try
{
- // successful offer to the queue, let's exit the loop
- break;
+ if (updateToReplayQueue.offer(updateToReplay, 1, TimeUnit.SECONDS))
+ {
+ queued = true;
+ synchronized (updateToReplayQueue)
+ {
+ updateToReplayQueue.notifyAll();
+ }
+ break;
+ }
+ }
+ catch (InterruptedException e)
+ {
+ // Thread interrupted: check for shutdown.
+ Thread.currentThread().interrupt();
}
}
- catch (InterruptedException e)
+ handoffFailed = false;
+ }
+ finally
+ {
+ if (!queued)
{
- // Thread interrupted: check for shutdown.
- Thread.currentThread().interrupt();
+ // Cancel before comparing listener identity. If join completes, the next session
+ // starts after this cancellation and can redeliver without an extra restart. If
+ // join was interrupted, a newer session may already have refused the delivery.
+ updateToReplay.cancel(false);
+ if (handoffFailed || !isCurrentListenerThread())
+ {
+ requestSessionRestart();
+ }
}
}
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChange.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChange.java
index fa96a19..4a77a40 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChange.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChange.java
@@ -38,9 +38,20 @@
*/
private volatile UpdateMsg msg;
/**
+ * A remote delivery has been accepted and is waiting to be claimed from the replay
+ * queue. It has no owner yet, but replacing it would move it behind newer changes.
+ * Local pending changes do not use this flag.
+ */
+ @GuardedBy("RemotePendingChanges.pendingChangesLock")
+ private boolean queuedForReplay = true;
+ @GuardedBy("RemotePendingChanges.pendingChangesLock")
+ private boolean replayStarted;
+ @GuardedBy("RemotePendingChanges.pendingChangesLock")
+ private UpdateToReplay delivery;
+ /**
* The replay thread which owns this change: it is being replayed, or it waits for the
- * change it depends on. A remote change which no thread owns is one whose replay
- * failed and which the replication server is expected to deliver again.
+ * change it depends on. A remote change which no thread owns is either queued for
+ * replay or was released and may be taken over by another delivery.
* <p>
* The owner is kept rather than the bare fact that there is one, so that a change is
* given back by the thread it was handed to and by nobody else: a release which arrives
@@ -137,6 +148,43 @@
public void setMsg(LDAPUpdateMsg msg)
{
this.msg = msg;
+ queuedForReplay = true;
+ }
+
+ /**
+ * Returns whether a remote delivery is waiting to be claimed from the replay queue.
+ *
+ * @return {@code true} if the accepted delivery has not been dequeued or cancelled
+ */
+ public boolean isQueuedForReplay()
+ {
+ return queuedForReplay && !isDeliveryCancelled();
+ }
+
+ /** Installs a retry without changing this change's ordering or failure history. */
+ void setDelivery(UpdateToReplay delivery)
+ {
+ setMsg(delivery.getUpdateMessage());
+ this.delivery = delivery;
+ }
+
+ /** A failed queue hand-off can be cancelled without acquiring the pending lock. */
+ boolean isDeliveryCancelled()
+ {
+ return delivery != null && delivery.isCancelled();
+ }
+
+ /** Marks a remote delivery as dequeued while its replay claim is being established. */
+ public void markDequeued()
+ {
+ queuedForReplay = false;
+ replayStarted = true;
+ }
+
+ /** Keeps the in-progress monitor stable while a failed replay waits in the retry queue. */
+ boolean hasStartedReplay()
+ {
+ return replayStarted;
}
/**
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java
index 085d147..5e801bc 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/RemotePendingChanges.java
@@ -74,10 +74,10 @@
@GuardedBy("dependentChangesLock")
private final SortedSet<PendingChange> dependentChanges = new TreeSet<>();
/**
- * {@code activeAndDependentChanges} also contains changes discovered to be dependent
- * on currently in progress changes.
+ * Uncommitted changes, registered before enqueue. A queued change whose hand-off
+ * fails must remain a predecessor: otherwise its DELETE can run before its ADD retry.
*/
- private final ConcurrentSkipListSet<PendingChange> activeAndDependentChanges = new ConcurrentSkipListSet<>();
+ private final ConcurrentSkipListSet<PendingChange> replayDependencies;
/**
* The change each replay thread is replaying right now, read by the give-back on the way
@@ -106,7 +106,7 @@
* races nothing for it. {@link #clear()} is the one writer of every entry, and it holds
* both locks.
*/
- private final ConcurrentMap<Thread, CSN> changeBeingReplayed = new ConcurrentHashMap<>();
+ private final ConcurrentMap<Thread, CSN> changeBeingReplayed;
private final ReentrantReadWriteLock pendingChangesLock = new ReentrantReadWriteLock(true);
private final ReentrantReadWriteLock.ReadLock pendingChangesReadLock = pendingChangesLock.readLock();
@@ -115,6 +115,16 @@
/** The ServerState that will be updated when LDAPUpdateMsg are fully replayed. */
private final ServerState state;
+ private final int localServerId;
+ /** Own-origin replay progress; local publication must not cover restore catch-up. */
+ @GuardedBy("pendingChangesLock")
+ private CSN localReplayCSN;
+
+ /** Whether a delivery needs replay or how its duplicate should be completed. */
+ enum Admission
+ {
+ ACCEPTED, QUEUED_DUPLICATE, DUPLICATE
+ }
/**
* How many of the changes listed here have a replay failure recorded against them.
@@ -137,7 +147,43 @@
*/
public RemotePendingChanges(ServerState state)
{
+ this(state, -1);
+ }
+
+ RemotePendingChanges(ServerState state, int localServerId)
+ {
+ this(state, localServerId, new ConcurrentHashMap<Thread, CSN>());
+ }
+
+ RemotePendingChanges(ServerState state, int localServerId,
+ ConcurrentMap<Thread, CSN> changeBeingReplayed)
+ {
+ this(state, localServerId, changeBeingReplayed, new ConcurrentSkipListSet<PendingChange>());
+ }
+
+ RemotePendingChanges(ServerState state, int localServerId,
+ ConcurrentMap<Thread, CSN> changeBeingReplayed,
+ ConcurrentSkipListSet<PendingChange> replayDependencies)
+ {
+ this.replayDependencies = replayDependencies;
this.state = state;
+ this.localServerId = localServerId;
+ this.changeBeingReplayed = changeBeingReplayed;
+ resetReplayState();
+ }
+
+ /** Reloads the own-origin watermark only when the backend state is loaded. */
+ void resetReplayState()
+ {
+ pendingChangesWriteLock.lock();
+ try
+ {
+ localReplayCSN = state.getCSN(localServerId);
+ }
+ finally
+ {
+ pendingChangesWriteLock.unlock();
+ }
}
/**
@@ -169,7 +215,23 @@
*/
public int changesInProgressSize()
{
- return activeAndDependentChanges.size();
+ pendingChangesReadLock.lock();
+ try
+ {
+ int count = 0;
+ for (PendingChange change : replayDependencies)
+ {
+ if (change.hasStartedReplay())
+ {
+ count++;
+ }
+ }
+ return count;
+ }
+ finally
+ {
+ pendingChangesReadLock.unlock();
+ }
}
/**
@@ -194,47 +256,82 @@
* Add a new LDAPUpdateMsg that was received from the replication server
* to the pendingList.
* <p>
- * A change which is already listed and which a replay thread owns - it is being
- * replayed, it waits for the change it depends on, or it has been replayed and waits
- * for the changes before it - is left alone: that copy knows what this one does not,
- * and replaying it again is exactly what the duplicate check is there to prevent
- * (OPENDJ-1115).
+ * Changes covered by the applied ServerState are rejected even after they have left
+ * the pending list. For this replica's own origin, only the state loaded from the backend
+ * and subsequent replays count: a newer local publication does not prove that an older
+ * change sent back after restore is in the data. A listed change is also left alone while its delivery is queued,
+ * owned by a replay thread, or committed and waiting for earlier changes. Replacing a
+ * queued delivery would move it behind newer changes: a DELETE could then be replayed
+ * before its ADD, leaving the entry present once the ADD is replayed (issue #1269).
* <p>
- * A change which is listed but which no replay thread owns is taken over by this
- * delivery: it is the one to replay, and the copy which may still wait in the replay
- * queue is dropped by {@link #markInProgress(LDAPUpdateMsg)}. That is a change whose
- * replay failed - it is deliberately left out of the ServerState, so the replication
- * server sends it again over the session which was restarted (issue #889) - and it is
- * also, harmlessly, a change which was listed a moment ago by a delivery no replay
- * thread has picked up yet: the two deliveries carry the same change, and the last one
- * listed is the one replayed.
+ * Only a delivery which was released after failed or abandoned replay, or cancelled
+ * before being enqueued, may be replaced. It is deliberately left out of the
+ * ServerState so the replication server sends it again over the restarted session
+ * (issue #889). Neither an owner nor a queued delivery remains to finish it.
* <p>
* The failures the change went through are kept, whichever delivery replays it: they
* are what has this replica eventually give up on a change it can not apply.
*
* @param update The LDAPUpdateMsg that was received from the replication
* server and that will be added to the pending list.
- * @return {@code false} if the update was already registered in the pending
- * changes and is owned by a replay thread.
+ * @return {@code false} if the change is already applied or has a delivery which is
+ * still queued, owned, or committed
*/
public boolean putRemoteUpdate(LDAPUpdateMsg update)
{
+ return putRemoteUpdate(new UpdateToReplay(update, null)) == Admission.ACCEPTED;
+ }
+
+ Admission putRemoteUpdate(UpdateToReplay delivery)
+ {
+ final LDAPUpdateMsg update = delivery.getUpdateMessage();
pendingChangesWriteLock.lock();
try
{
final CSN csn = update.getCSN();
+ if (csn.getServerId() == localServerId
+ ? localReplayCSN != null && localReplayCSN.compareTo(csn) >= 0 : state.cover(csn))
+ {
+ return Admission.DUPLICATE;
+ }
final PendingChange listed = pendingChanges.get(csn);
if (listed == null)
{
- pendingChanges.put(csn, new PendingChange(csn, null, update));
- return true;
+ final PendingChange change = new PendingChange(csn, null, update);
+ change.setDelivery(delivery);
+ pendingChanges.put(csn, change);
+ try
+ {
+ replayDependencies.add(change);
+ }
+ catch (RuntimeException | Error e)
+ {
+ // No other admission sees this partial registration while the write lock is held.
+ // Request recovery before rollback: removing a skip-list node can also fail.
+ delivery.cancel();
+ pendingChanges.remove(csn);
+ replayDependencies.remove(change);
+ throw e;
+ }
+ return Admission.ACCEPTED;
}
if (listed.isCommitted() || listed.isOwned())
{
- return false;
+ return Admission.DUPLICATE;
}
- listed.setMsg(update);
- return true;
+ if (listed.isQueuedForReplay())
+ {
+ // Keep the queue position, but complete for the latest delivery's assured mode.
+ // This changes in-memory ack metadata only; the received message's cached wire
+ // encoding is stale and must not be reused to publish this message.
+ final LDAPUpdateMsg accepted = listed.getLDAPUpdateMsg();
+ accepted.setAssured(update.isAssured());
+ accepted.setAssuredMode(update.getAssuredMode());
+ accepted.setSafeDataLevel(update.getSafeDataLevel());
+ return Admission.QUEUED_DUPLICATE;
+ }
+ listed.setDelivery(delivery);
+ return Admission.ACCEPTED;
}
finally
{
@@ -278,7 +375,7 @@
curChange.setCommitted(true);
curChange.setOwner(null);
changeBeingReplayed.remove(Thread.currentThread(), csn);
- activeAndDependentChanges.remove(curChange);
+ replayDependencies.remove(curChange);
final Iterator<PendingChange> it = pendingChanges.values().iterator();
while (it.hasNext())
@@ -291,6 +388,10 @@
if (pendingChange.getMsg().contributesToDomainState())
{
state.update(pendingChange.getCSN());
+ if (pendingChange.getCSN().getServerId() == localServerId)
+ {
+ localReplayCSN = pendingChange.getCSN();
+ }
}
if (pendingChange.getReplayFailures() > 0)
{
@@ -504,7 +605,7 @@
{
pendingChanges.clear();
dependentChanges.clear();
- activeAndDependentChanges.clear();
+ replayDependencies.clear();
changeBeingReplayed.clear();
failingChanges = 0;
}
@@ -520,12 +621,8 @@
*
* @param msg
* the message whose change is being replayed
- * @return {@code false} if this message is not the delivery which is listed as
- * pending, which happens when the session was restarted after a failed replay
- * while this message was still waiting in the replay queue: the replication
- * server delivered the change again and that delivery took over from this one,
- * or the domain was disabled and forgot the change, so this copy must not be
- * replayed.
+ * @return {@code false} if the change is already committed, the domain has forgotten
+ * it, the delivery was cancelled, or this is a stale delivery which a retry has replaced
*/
public boolean markInProgress(LDAPUpdateMsg msg)
{
@@ -533,18 +630,19 @@
try
{
final PendingChange change = pendingChanges.get(msg.getCSN());
- if (change == null || change.isCommitted() || change.getLDAPUpdateMsg() != msg)
+ if (change == null || change.isCommitted() || change.isDeliveryCancelled()
+ || change.getLDAPUpdateMsg() != msg)
{
return false;
}
+ // This delivery has left the queue. If an allocation below fails before it gets
+ // an owner, a later delivery must be able to take it over.
+ change.markDequeued();
/*
- * Listed as being replayed before it is owned, and not the other way round: the
- * caller enters the replay - where the give-back on the way out lives - once this
- * returns, so nothing which allocates must run between the owner being stamped and
- * that. A change listed here without an owner is the state a failed replay leaves
- * behind, and the dependency checks which read this set do not read the owner.
+ * The dependency was registered before enqueue, so a failed claim still holds back
+ * its successors. Allocate the thread index before stamping the owner: the caller
+ * reaches replay's give-back only once this method returns.
*/
- activeAndDependentChanges.add(change);
changeBeingReplayed.put(Thread.currentThread(), change.getCSN());
change.setOwner(Thread.currentThread());
return true;
@@ -836,7 +934,7 @@
boolean hasDependencies = false;
final DN targetDN = op.getEntryDN();
- for (PendingChange pendingChange : activeAndDependentChanges)
+ for (PendingChange pendingChange : replayDependencies)
{
if (pendingChange.getCSN().isNewerThanOrEqualTo(csn))
{
@@ -910,7 +1008,7 @@
boolean hasDependencies = false;
final DN targetDN = change.getLDAPUpdateMsg().getDN();
- for (PendingChange pendingChange : activeAndDependentChanges)
+ for (PendingChange pendingChange : replayDependencies)
{
if (pendingChange.getCSN().isNewerThanOrEqualTo(csn))
{
@@ -977,7 +1075,7 @@
boolean hasDependencies = false;
final DN targetDN = change.getLDAPUpdateMsg().getDN();
- for (PendingChange pendingChange : activeAndDependentChanges)
+ for (PendingChange pendingChange : replayDependencies)
{
if (pendingChange.getCSN().isNewerThanOrEqualTo(csn))
{
@@ -1052,7 +1150,7 @@
boolean hasDependencies = false;
final DN targetDN = op.getEntryDN();
- for (PendingChange pendingChange : activeAndDependentChanges)
+ for (PendingChange pendingChange : replayDependencies)
{
if (pendingChange.getCSN().isNewerThanOrEqualTo(csn))
{
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java
index 6c5e357..12bfa5d 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/ReplayThread.java
@@ -74,6 +74,10 @@
public void shutdown()
{
shutdown.set(true);
+ synchronized (updateToReplayQueue)
+ {
+ updateToReplayQueue.notifyAll();
+ }
}
/**
@@ -143,29 +147,53 @@
{
LDAPReplicationDomain domain;
LDAPUpdateMsg updateMsg;
+ UpdateToReplay delivery = null;
try
{
- if (shutdown.get())
+ synchronized (updateToReplayQueue)
{
- break;
+ if (shutdown.get())
+ {
+ break;
+ }
+ if (updateToReplayQueue.isEmpty())
+ {
+ updateToReplayQueue.wait(1000L);
+ }
+ if (shutdown.get())
+ {
+ break;
+ }
+ // Retain the head before removing it. LinkedBlockingQueue.poll() can throw
+ // OOME after unlinking it (signalNotFull takes another lock on JDK 11/17).
+ // Every consumer holds this monitor, so peek and poll refer to the same item.
+ delivery = updateToReplayQueue.peek();
+ if (delivery == null)
+ {
+ continue;
+ }
+ updateToReplayQueue.poll();
+ updateMsg = delivery.getUpdateMessage();
+ domain = delivery.getReplicationDomain();
+ if (!domain.markInProgress(updateMsg))
+ {
+ /*
+ * This delivery was cancelled, replaced, committed or forgotten. The
+ * domain counted the stale copy; only its accepted retry may replay it.
+ */
+ continue;
+ }
}
- UpdateToReplay updateToreplay = updateToReplayQueue.poll(1L, TimeUnit.SECONDS);
- if (updateToreplay == null)
+ }
+ catch (Throwable t)
+ {
+ if (delivery != null)
{
- continue;
+ // No pending-changes lock is needed: the failed claim may itself have
+ // run out of memory while trying to acquire that lock.
+ delivery.cancel();
}
- // Find replication domain for that update message and mark it as "in progress"
- updateMsg = updateToreplay.getUpdateMessage();
- domain = updateToreplay.getReplicationDomain();
- if (!domain.markInProgress(updateMsg))
- {
- /*
- * The domain restarted its session after a failed replay while this
- * message was waiting here, so it does not know about this change
- * anymore: the replication server sends it again over the new session.
- */
- continue;
- }
+ throw t;
}
finally
{
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/UpdateToReplay.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/UpdateToReplay.java
index 3919dd6..6fdf659 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/UpdateToReplay.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/UpdateToReplay.java
@@ -13,21 +13,22 @@
*
* Copyright 2006-2008 Sun Microsystems, Inc.
* Portions Copyright 2015 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.replication.plugin;
import org.opends.server.replication.protocol.LDAPUpdateMsg;
/**
- * This is a bag class to hold an update to replay in the queue of updates to
- * be replayed by the replay threads.
- * It associates an update message to replay with the matching
- * ReplicationDomain.
+ * A delivery in the shared replay queue, associating an update with its domain.
+ * Cancellation belongs to this delivery, so an enqueue or claim failure can release
+ * it without acquiring the pending-changes lock or cancelling a newer retry.
*/
public class UpdateToReplay
{
- private LDAPUpdateMsg updateMessage;
- private LDAPReplicationDomain replicationDomain;
+ private final LDAPUpdateMsg updateMessage;
+ private final LDAPReplicationDomain replicationDomain;
+ private volatile boolean cancelled;
/**
* Construct the object associating the update message with the replication
@@ -44,6 +45,30 @@
}
/**
+ * Releases a failed hand-off without a lock or allocation, even on an OOME road.
+ * A stale cancellation affects only this delivery, never the retry which replaced it.
+ */
+ void cancel()
+ {
+ cancel(true);
+ }
+
+ /** Cancels this delivery, optionally requesting another session. */
+ void cancel(boolean requestRestart)
+ {
+ cancelled = true;
+ if (requestRestart && replicationDomain != null)
+ {
+ replicationDomain.requestSessionRestart();
+ }
+ }
+
+ boolean isCancelled()
+ {
+ return cancelled;
+ }
+
+ /**
* Getter for update message.
* @return The update message
*/
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java
index f5fef33..fd04a25 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java
@@ -3599,6 +3599,12 @@
return tmp == null || tmp.isShutdownInitiated();
}
+ /** Returns whether the calling thread is still the registered listener for this session. */
+ protected final boolean isCurrentListenerThread()
+ {
+ return Thread.currentThread() == listenerThread;
+ }
+
/**
* Returns the generation of the session of this domain: bumped by
* {@link #disableService()} and {@link #enableService()} under
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/DuplicateReplayTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/DuplicateReplayTest.java
new file mode 100644
index 0000000..4d8d710
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/DuplicateReplayTest.java
@@ -0,0 +1,746 @@
+/*
+ * The contents of this file are subject to the terms of the Common Development and
+ * Distribution License (the License). You may not use this file except in compliance with the
+ * License.
+ *
+ * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
+ * specific language governing permission and limitations under the License.
+ *
+ * When distributing Covered Software, include this CDDL Header Notice in each file and include
+ * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
+ * Header, with the fields enclosed by brackets [] replaced by your own identifying
+ * information: "Portions copyright [year] [name of copyright owner]".
+ *
+ * Copyright 2026 3A Systems, LLC.
+ */
+package org.opends.server.replication.plugin;
+
+import static org.opends.server.TestCaseUtils.*;
+import static org.opends.server.core.DirectoryServer.*;
+import static org.testng.Assert.*;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.TreeSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.forgerock.opendj.ldap.DN;
+import org.forgerock.opendj.server.config.meta.ReplicationDomainCfgDefn.IsolationPolicy;
+import org.opends.server.api.DirectoryThread;
+import org.opends.server.core.AddOperation;
+import org.opends.server.protocols.internal.InternalClientConnection;
+import org.opends.server.replication.ReplicationTestCase;
+import org.opends.server.replication.common.AssuredMode;
+import org.opends.server.replication.common.CSN;
+import org.opends.server.replication.common.CSNGenerator;
+import org.opends.server.replication.protocol.AckMsg;
+import org.opends.server.replication.protocol.AddMsg;
+import org.opends.server.replication.protocol.DeleteMsg;
+import org.opends.server.replication.protocol.LDAPUpdateMsg;
+import org.opends.server.replication.protocol.ReplSessionSecurity;
+import org.opends.server.replication.protocol.ReplicationMsg;
+import org.opends.server.replication.service.ReplicationBroker;
+import org.opends.server.replication.service.ReplicationDomain;
+import org.opends.server.types.Entry;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+/** Tests duplicate delivery across reconnects against the real replay and backend (issue #1269). */
+@SuppressWarnings("javadoc")
+public class DuplicateReplayTest extends ReplicationTestCase
+{
+ private static final String ENTRY_UUID = "c9cb8c3c-615a-4122-865d-50323aaaed48";
+
+ private DN baseDN;
+ private LDAPReplicationDomain domain;
+ private TestSynchronousReplayQueue queue;
+ private Entry entry;
+ private String parentUUID;
+ private CSN addCSN;
+ private CSN deleteCSN;
+ private OutOfMemoryError enqueueFailure;
+ private OutOfMemoryError pollFailure;
+ private boolean failBeforePoll;
+ private Runnable beforeEnqueue;
+ private boolean refuseEnqueue;
+ private Runnable onEmpty;
+ private Runnable afterPoll;
+
+ @BeforeMethod
+ public void setUpLocal() throws Exception
+ {
+ baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ initializeTestBackend(true);
+ enqueueFailure = null;
+ pollFailure = null;
+ failBeforePoll = false;
+ beforeEnqueue = null;
+ refuseEnqueue = false;
+ onEmpty = null;
+ afterPoll = null;
+ queue = new TestSynchronousReplayQueue()
+ {
+ @Override
+ public boolean offer(UpdateToReplay update, long timeout, TimeUnit unit) throws InterruptedException
+ {
+ if (beforeEnqueue != null)
+ {
+ beforeEnqueue.run();
+ }
+ if (enqueueFailure != null)
+ {
+ throw enqueueFailure;
+ }
+ return !refuseEnqueue && super.offer(update, timeout, unit);
+ }
+
+ @Override
+ public UpdateToReplay poll(long timeout, TimeUnit unit)
+ {
+ return poll();
+ }
+
+ @Override
+ public UpdateToReplay poll()
+ {
+ if (pollFailure != null && failBeforePoll)
+ {
+ throw pollFailure;
+ }
+ final UpdateToReplay result = super.poll();
+ if (pollFailure != null)
+ {
+ throw pollFailure;
+ }
+ if (afterPoll != null)
+ {
+ afterPoll.run();
+ }
+ return result;
+ }
+
+ @Override
+ public boolean isEmpty()
+ {
+ final boolean empty = super.isEmpty();
+ if (empty && onEmpty != null)
+ {
+ onEmpty.run();
+ }
+ return empty;
+ }
+ };
+ final DomainFakeCfg conf = new DomainFakeCfg(baseDN, 1, new TreeSet<String>());
+ conf.setIsolationPolicy(IsolationPolicy.ACCEPT_ALL_UPDATES);
+ domain = MultimasterReplication.createNewDomain(conf, queue);
+ domain.start();
+
+ entry = makeEntry("dn: cn=duplicate," + baseDN,
+ "objectClass: top", "objectClass: person", "cn: duplicate", "sn: duplicate",
+ "entryUUID: " + ENTRY_UUID);
+ parentUUID = getEntryUUID(baseDN);
+ final CSNGenerator generator = new CSNGenerator(201, 0);
+ addCSN = generator.newCSN();
+ deleteCSN = generator.newCSN();
+ }
+
+ @AfterMethod
+ public void tearDown() throws Exception
+ {
+ MultimasterReplication.deleteDomain(baseDN);
+ }
+
+ @Test
+ public void appliedAddDeliveredAgainAfterDeleteDoesNotResurrectEntry() throws Exception
+ {
+ domain.processUpdate(addMsg());
+ replayQueuedMessages();
+ assertTrue(entryExists(entry.getName()));
+ assertEquals(getEntryUUID(entry.getName()), ENTRY_UUID);
+ domain.processUpdate(deleteMsg());
+ replayQueuedMessages();
+ assertFalse(entryExists(entry.getName()));
+ assertTrue(domain.getServerState().cover(deleteCSN));
+
+ // A new session can deliver an old change after it has left the pending list.
+ assertFalse(domain.processUpdate(addMsg()));
+ replayQueuedMessages();
+
+ assertFalse(entryExists(entry.getName()), "an applied ADD was replayed after its DELETE");
+ assertEquals(processedUpdates(), 3, "each delivery must be counted, including the duplicate");
+ }
+
+ @Test
+ public void queuedDuplicateAddDoesNotMoveTheOriginalBehindDelete() throws Exception
+ {
+ domain.processUpdate(addMsg());
+ domain.processUpdate(deleteMsg());
+ final int completions = completedUpdates();
+ // Pause replay until the duplicate has arrived: no timing or reconnect race is needed.
+ assertFalse(domain.processUpdate(addMsg()));
+ assertEquals(completedUpdates(), completions, "a queued duplicate must not return credit before replay");
+ assertFalse(domain.getServerState().cover(addCSN), "discarding a duplicate must not advance state");
+ assertFalse(domain.getServerState().cover(deleteCSN));
+
+ replayQueuedMessages();
+
+ assertEquals(completedUpdates(), completions + 2, "one completion per replayed change");
+ assertFalse(entryExists(entry.getName()), "the queued ADD moved behind its DELETE");
+ assertTrue(domain.getServerState().cover(deleteCSN));
+ assertEquals(processedUpdates(), 3, "each delivery must be counted, including the duplicate");
+ }
+
+ @Test
+ public void deliveryAbandonedBeforeEnqueueCanBeRetried() throws Exception
+ {
+ final AtomicBoolean recovering = (AtomicBoolean) domainField("replayFailureRecovery");
+ final SessionRestartRequests restarts = (SessionRestartRequests) domainField("sessionRestarts");
+ recovering.set(true);
+ try
+ {
+ domain.disableService();
+ domain.processUpdate(addMsg());
+ assertTrue(queue.isEmpty());
+ assertFalse(domain.getServerState().cover(addCSN));
+
+ domain.enableService();
+ domain.processUpdate(addMsg());
+ replayQueuedMessages();
+
+ assertTrue(entryExists(entry.getName()), "a delivery abandoned before enqueue was refused on retry");
+ assertTrue(domain.getServerState().cover(addCSN));
+ }
+ finally
+ {
+ restarts.clear();
+ recovering.set(false);
+ }
+ }
+
+ @Test
+ public void failedEnqueueCanBeRetried() throws Exception
+ {
+ final AtomicBoolean recovering = (AtomicBoolean) domainField("replayFailureRecovery");
+ final SessionRestartRequests restarts = (SessionRestartRequests) domainField("sessionRestarts");
+ recovering.set(true);
+ try
+ {
+ enqueueFailure = new OutOfMemoryError("enqueue allocation failed");
+ try
+ {
+ domain.processUpdate(addMsg());
+ fail("the injected allocation failure did not reach enqueue");
+ }
+ catch (OutOfMemoryError e)
+ {
+ assertSame(e, enqueueFailure);
+ }
+ finally
+ {
+ enqueueFailure = null;
+ }
+ assertTrue(queue.isEmpty());
+ assertFalse(domain.getServerState().cover(addCSN));
+
+ assertTrue(restarts.isPending(), "a failed enqueue must request redelivery");
+ domain.processUpdate(addMsg());
+ replayQueuedMessages();
+
+ assertTrue(entryExists(entry.getName()), "a failed enqueue was refused on retry");
+ assertTrue(domain.getServerState().cover(addCSN));
+ }
+ finally
+ {
+ restarts.clear();
+ recovering.set(false);
+ }
+ }
+
+ @Test
+ public void ownChangeSentBeforeANewerLocalPublishIsStillReplayed() throws Exception
+ {
+ // Model the incoming backlog after restoring a backup that predates these changes.
+ final CSNGenerator generator = new CSNGenerator(1, 0);
+ addCSN = generator.newCSN();
+ deleteCSN = generator.newCSN();
+ domain.publish(deleteMsg());
+ assertTrue(domain.getServerState().cover(addCSN));
+
+ domain.processUpdate(addMsg());
+ replayQueuedMessages();
+ assertTrue(entryExists(entry.getName()), "local publication must not hide a missing own-origin change");
+ domain.processUpdate(deleteMsg());
+ replayQueuedMessages();
+ assertFalse(entryExists(entry.getName()));
+ domain.processUpdate(addMsg());
+ replayQueuedMessages();
+ assertFalse(entryExists(entry.getName()), "own-origin replay also needs duplicate suppression");
+ }
+
+ @Test
+ public void backendReloadResetsTheOwnReplayWatermark() throws Exception
+ {
+ final CSNGenerator generator = new CSNGenerator(1, 0);
+ addCSN = generator.newCSN();
+ deleteCSN = generator.newCSN();
+ domain.processUpdate(addMsg());
+ domain.processUpdate(deleteMsg());
+ replayQueuedMessages();
+ assertFalse(entryExists(entry.getName()));
+ domain.disable();
+ // Restore the persisted state of a backup taken before the ADD. Its entry is absent,
+ // as it is in this backend after DELETE; no history on that entry can cover the ADD.
+ ((PersistentServerState) domainField("state")).clear();
+ domain.enable();
+ assertFalse(domain.getServerState().cover(addCSN));
+ domain.processUpdate(addMsg());
+ replayQueuedMessages();
+ assertTrue(entryExists(entry.getName()), "backend reload retained the pre-restore replay watermark");
+ }
+
+ private int completedUpdates() throws Exception
+ {
+ final Field brokerField = ReplicationDomain.class.getDeclaredField("broker");
+ brokerField.setAccessible(true);
+ final Object broker = brokerField.get(domain);
+ final Field counter = ReplicationBroker.class.getDeclaredField("updateDoneCount");
+ counter.setAccessible(true);
+ return counter.getInt(broker);
+ }
+
+ @DataProvider
+ public Object[][] queueFailurePositions()
+ {
+ return new Object[][] { { false }, { true } };
+ }
+
+ @Test(dataProvider = "queueFailurePositions")
+ public void failureAroundQueueRemovalDoesNotLoseTheChange(boolean beforeRemoval) throws Exception
+ {
+ final AtomicBoolean recovering = (AtomicBoolean) domainField("replayFailureRecovery");
+ final SessionRestartRequests restarts = (SessionRestartRequests) domainField("sessionRestarts");
+ recovering.set(true); // Keep the checkpointer from consuming the request under test.
+ try
+ {
+ domain.processUpdate(addMsg());
+ domain.processUpdate(deleteMsg());
+ failBeforePoll = beforeRemoval;
+ pollFailure = new OutOfMemoryError("dequeue failed");
+ try
+ {
+ new ReplayThread(queue, new ReentrantLock()).run();
+ fail("the injected dequeue failure did not escape");
+ }
+ catch (OutOfMemoryError e)
+ {
+ assertSame(e, pollFailure);
+ }
+ finally
+ {
+ pollFailure = null;
+ }
+ assertEquals(queue.size(), beforeRemoval ? 2 : 1);
+ assertTrue(restarts.isPending(), "a lost queue delivery must be requested again");
+ replayQueuedMessages(); // DELETE must wait for the cancelled ADD, not commit NO_SUCH_OBJECT.
+ domain.processUpdate(addMsg());
+ replayQueuedMessages();
+ assertFalse(entryExists(entry.getName()), "DELETE overtook the retry of the removed ADD");
+ assertTrue(domain.getServerState().cover(deleteCSN));
+ }
+ finally
+ {
+ restarts.clear();
+ recovering.set(false);
+ }
+ }
+
+ @Test
+ public void cancellationAfterANewerDeliveryWasRefusedRequestsAnotherSession() throws Exception
+ {
+ final AtomicBoolean recovering = (AtomicBoolean) domainField("replayFailureRecovery");
+ final SessionRestartRequests restarts = (SessionRestartRequests) domainField("sessionRestarts");
+ recovering.set(true);
+ try
+ {
+ enqueueFailure = new OutOfMemoryError("old listener cannot finish enqueue");
+ beforeEnqueue = () ->
+ {
+ // A newer session redelivers while the old listener is still in offer().
+ assertFalse(domain.processUpdate(addMsg()));
+ assertTrue(queue.isEmpty());
+ beforeEnqueue = null;
+ final OutOfMemoryError oldFailure = enqueueFailure;
+ enqueueFailure = null;
+ try
+ {
+ domain.processUpdate(deleteMsg());
+ }
+ finally
+ {
+ enqueueFailure = oldFailure;
+ }
+ };
+ try
+ {
+ domain.processUpdate(addMsg());
+ fail("enqueue must fail");
+ }
+ catch (OutOfMemoryError e)
+ {
+ assertSame(e, enqueueFailure);
+ }
+ finally
+ {
+ beforeEnqueue = null;
+ enqueueFailure = null;
+ }
+ assertTrue(restarts.isPending(), "the newer session already missed its chance to retry");
+ replayQueuedMessages();
+ domain.processUpdate(addMsg());
+ replayQueuedMessages();
+ assertFalse(entryExists(entry.getName()), "DELETE overtook the cancelled ADD's retry");
+ assertTrue(domain.getServerState().cover(deleteCSN));
+ }
+ finally
+ {
+ restarts.clear();
+ recovering.set(false);
+ }
+ }
+
+ @DataProvider
+ public Object[][] listenerStops()
+ {
+ return new Object[][] { { false }, { true } };
+ }
+
+ @Test(dataProvider = "listenerStops")
+ public void shutdownCancellationRestartsOnlyAfterAnIncompleteJoin(boolean interruptJoin) throws Exception
+ {
+ final AtomicBoolean recovering = (AtomicBoolean) domainField("replayFailureRecovery");
+ final SessionRestartRequests restarts = (SessionRestartRequests) domainField("sessionRestarts");
+ final CountDownLatch offering = new CountDownLatch(1);
+ final CountDownLatch releaseOffer = new CountDownLatch(1);
+ final CountDownLatch stopping = new CountDownLatch(1);
+ final AtomicReference<Throwable> listenerFailure = new AtomicReference<>();
+ final Field listenerField = ReplicationDomain.class.getDeclaredField("listenerThread");
+ listenerField.setAccessible(true);
+ recovering.set(true);
+ domain.disableService();
+ refuseEnqueue = true;
+ beforeEnqueue = () ->
+ {
+ offering.countDown();
+ await(releaseOffer);
+ };
+ // Use the real listener slot and disableService/join, but supply the incoming change
+ // directly so the interleaving does not need a replication server or a full 10000-slot queue.
+ final DirectoryThread listener = new DirectoryThread(() ->
+ {
+ try
+ {
+ domain.processUpdate(addMsg());
+ }
+ catch (Throwable t)
+ {
+ listenerFailure.set(t);
+ }
+ }, "held duplicate-replay listener");
+ final FutureTask<Void> stop = new FutureTask<>(() -> { domain.disableService(); return null; });
+ final Thread stopper = new Thread(stop, "duplicate-replay service stop");
+ try
+ {
+ listenerField.set(domain, listener);
+ listener.start();
+ assertTrue(offering.await(10, TimeUnit.SECONDS));
+ domain.setServiceStopHook(stopping::countDown);
+ stopper.start();
+ assertTrue(stopping.await(10, TimeUnit.SECONDS));
+ if (interruptJoin)
+ {
+ stopper.interrupt();
+ stop.get(10, TimeUnit.SECONDS);
+ domain.setServiceStopHook(null);
+ domain.enableService();
+ domain.processUpdate(addMsg()); // The new session refuses the still-live old delivery.
+ assertTrue(queue.isEmpty());
+ domain.disableService();
+ restarts.clear(); // A restart before cancellation cannot be enough.
+ }
+ else
+ {
+ // Taking sessionLock would deadlock with join; wait on the thread's own flag.
+ final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+ while (!listener.isShutdownInitiated() && System.nanoTime() < deadline)
+ {
+ Thread.yield();
+ }
+ assertTrue(listener.isShutdownInitiated());
+ }
+ releaseOffer.countDown();
+ stop.get(10, TimeUnit.SECONDS);
+ listener.join(10000);
+ assertFalse(listener.isAlive());
+ assertNull(listenerFailure.get());
+ assertEquals(restarts.isPending(), interruptJoin,
+ "a completed join already precedes redelivery; a detached listener needs a fresh restart");
+ refuseEnqueue = false;
+ beforeEnqueue = null;
+ domain.enableService();
+ domain.processUpdate(addMsg());
+ replayQueuedMessages();
+ assertTrue(entryExists(entry.getName()));
+ }
+ finally
+ {
+ releaseOffer.countDown();
+ stopper.join(10000);
+ listener.join(10000);
+ domain.setServiceStopHook(null);
+ restarts.clear();
+ recovering.set(false);
+ }
+ }
+
+ @Test
+ public void failedHandoffByRegisteredListenerRequestsRestart() throws Exception
+ {
+ final AtomicBoolean recovering = (AtomicBoolean) domainField("replayFailureRecovery");
+ final SessionRestartRequests restarts = (SessionRestartRequests) domainField("sessionRestarts");
+ final AtomicReference<Throwable> listenerFailure = new AtomicReference<>();
+ final Field listenerField = ReplicationDomain.class.getDeclaredField("listenerThread");
+ listenerField.setAccessible(true);
+ recovering.set(true);
+ domain.disableService();
+ enqueueFailure = new OutOfMemoryError("registered listener enqueue failed");
+ final DirectoryThread listener = new DirectoryThread(() ->
+ {
+ try
+ {
+ domain.processUpdate(addMsg());
+ }
+ catch (Throwable t)
+ {
+ listenerFailure.set(t);
+ }
+ }, "failed duplicate-replay listener");
+ try
+ {
+ listenerField.set(domain, listener);
+ listener.start();
+ listener.join(10000);
+ assertFalse(listener.isAlive());
+ assertSame(listenerFailure.get(), enqueueFailure);
+ assertTrue(restarts.isPending(), "failure needs recovery even while the listener is registered");
+ enqueueFailure = null;
+ domain.disableService();
+ domain.enableService();
+ domain.processUpdate(addMsg());
+ replayQueuedMessages();
+ assertTrue(entryExists(entry.getName()));
+ }
+ finally
+ {
+ listener.join(10000);
+ enqueueFailure = null;
+ restarts.clear();
+ recovering.set(false);
+ }
+ }
+
+ @Test
+ public void replayConsumerWakesForAnEmptyQueueAndDrainsItsBacklog() throws Exception
+ {
+ final CountDownLatch empty = new CountDownLatch(1);
+ final CountDownLatch removed = new CountDownLatch(2);
+ final CountDownLatch replayed = new CountDownLatch(1);
+ final ReplayThread consumer = new ReplayThread(queue, new ReentrantLock());
+ onEmpty = () ->
+ {
+ empty.countDown();
+ if (domain.getServerState().cover(deleteCSN))
+ {
+ replayed.countDown();
+ }
+ };
+ afterPoll = removed::countDown;
+ try
+ {
+ consumer.start();
+ assertTrue(empty.await(10, TimeUnit.SECONDS));
+ synchronized (queue)
+ {
+ // The consumer observed an empty queue while holding this monitor. Acquiring it
+ // proves it has now entered wait(). Deliver both changes before it can wake up.
+ domain.processUpdate(addMsg());
+ domain.processUpdate(deleteMsg());
+ }
+ assertTrue(removed.await(750, TimeUnit.MILLISECONDS),
+ "notification must wake the consumer and a non-empty queue must not wait again");
+ // poll() runs before replay. Stopping at the second removal could abandon DELETE.
+ assertTrue(replayed.await(10, TimeUnit.SECONDS), "the consumer did not commit its backlog");
+ }
+ finally
+ {
+ consumer.shutdown();
+ consumer.join(10000);
+ onEmpty = null;
+ afterPoll = null;
+ }
+ assertFalse(consumer.isAlive());
+ assertFalse(entryExists(entry.getName()));
+ assertTrue(domain.getServerState().cover(deleteCSN));
+ }
+
+ private static void await(CountDownLatch latch)
+ {
+ try
+ {
+ assertTrue(latch.await(10, TimeUnit.SECONDS), "controlled hand-off timed out");
+ }
+ catch (InterruptedException e)
+ {
+ Thread.currentThread().interrupt();
+ throw new AssertionError(e);
+ }
+ }
+
+ private Object domainField(String name) throws Exception
+ {
+ final Field field = LDAPReplicationDomain.class.getDeclaredField(name);
+ field.setAccessible(true);
+ return field.get(domain);
+ }
+
+ @DataProvider
+ public Object[][] assuredDeliveries()
+ {
+ return new Object[][] {
+ { false, true, false }, { true, true, false }, { true, false, false },
+ { false, true, true }, { true, true, true }, { true, false, true }
+ };
+ }
+
+ @Test(dataProvider = "assuredDeliveries")
+ public void queuedRedeliveryAcknowledgesOnlyTheReplayResult(
+ boolean firstAssured, boolean lastAssured, boolean failReplay) throws Exception
+ {
+ final Field brokerField = ReplicationDomain.class.getDeclaredField("broker");
+ brokerField.setAccessible(true);
+ final ReplicationBroker originalBroker = (ReplicationBroker) brokerField.get(domain);
+ final List<AckMsg> acks = new ArrayList<>();
+ // A non-started broker retains real completion accounting but records the wire ack.
+ // No domain is passed to its constructor, so it registers no second monitor.
+ final ReplicationBroker recordingBroker = new ReplicationBroker(null, domain.getServerState(),
+ new DomainFakeCfg(baseDN, 1, new TreeSet<String>()), new ReplSessionSecurity())
+ {
+ @Override
+ public byte getRsGroupId()
+ {
+ return domain.getGroupId();
+ }
+
+ @Override
+ public void publish(ReplicationMsg msg)
+ {
+ if (msg instanceof AckMsg)
+ {
+ acks.add((AckMsg) msg);
+ }
+ }
+ };
+ brokerField.set(domain, recordingBroker);
+ final AtomicBoolean recovering = (AtomicBoolean) domainField("replayFailureRecovery");
+ final SessionRestartRequests restarts = (SessionRestartRequests) domainField("sessionRestarts");
+ recovering.set(true);
+ final OutOfMemoryError failure = new OutOfMemoryError("cannot construct replay operation");
+ try
+ {
+ final AddMsg original = failReplay ? new AddMsg(addCSN, entry.getName(), ENTRY_UUID, parentUUID,
+ entry.getObjectClasses(), entry.getUserAttributes(), entry.getOperationalAttributes())
+ {
+ @Override
+ public AddOperation createOperation(InternalClientConnection connection, DN newDN)
+ {
+ throw failure;
+ }
+ } : addMsg();
+ original.setAssured(firstAssured);
+ original.setAssuredMode(AssuredMode.SAFE_DATA_MODE);
+ domain.processUpdate(original);
+ for (int i = 0; i < 2; i++)
+ {
+ final AddMsg duplicate = addMsg();
+ duplicate.setAssured(lastAssured);
+ duplicate.setAssuredMode(AssuredMode.SAFE_READ_MODE);
+ domain.processUpdate(duplicate);
+ assertTrue(acks.isEmpty(), "queued redelivery must not acknowledge unapplied data");
+ assertEquals(completedUpdates(), 0);
+ }
+ try
+ {
+ replayQueuedMessages();
+ assertFalse(failReplay, "the injected replay failure must escape");
+ }
+ catch (OutOfMemoryError e)
+ {
+ assertTrue(failReplay);
+ assertSame(e, failure);
+ }
+ assertEquals(completedUpdates(), 1, "only the accepted replay returns window credit");
+ assertEquals(acks.size(), lastAssured ? 1 : 0);
+ if (lastAssured)
+ {
+ assertEquals(acks.get(0).getCSN(), addCSN);
+ assertEquals(acks.get(0).hasReplayError(), failReplay);
+ }
+ assertEquals(entryExists(entry.getName()), !failReplay);
+ }
+ finally
+ {
+ brokerField.set(domain, originalBroker);
+ restarts.clear();
+ recovering.set(false);
+ }
+ }
+
+ private AddMsg addMsg()
+ {
+ return new AddMsg(addCSN, entry.getName(), ENTRY_UUID, parentUUID,
+ entry.getObjectClasses(), entry.getUserAttributes(), entry.getOperationalAttributes());
+ }
+
+ private DeleteMsg deleteMsg()
+ {
+ return new DeleteMsg(entry.getName(), deleteCSN, ENTRY_UUID);
+ }
+
+ private void replayQueuedMessages() throws Exception
+ {
+ while (!queue.isEmpty())
+ {
+ final LDAPUpdateMsg msg = queue.take().getUpdateMessage();
+ if (domain.markInProgress(msg))
+ {
+ domain.replay(msg, new AtomicBoolean(false));
+ }
+ }
+ }
+
+ private int processedUpdates() throws Exception
+ {
+ // The monitor counter is package-private in service; this domain has no RS to register a monitor.
+ final Method getter = ReplicationDomain.class.getDeclaredMethod("getNumProcessedUpdates");
+ getter.setAccessible(true);
+ return (Integer) getter.invoke(domain);
+ }
+}
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/NamingConflictTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/NamingConflictTest.java
index 94855bc..fc0117e 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/NamingConflictTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/NamingConflictTest.java
@@ -147,11 +147,13 @@
Entry entry = createAndAddEntry("simultaneousModrdnConflict");
String entryUUID = getEntryUUID(entry.getName());
- // generate two consecutive CSN that will be used in backward order
+ // Changes from different origins can arrive in reverse CSN order. Older changes
+ // from the same origin would already be covered by its applied ServerState.
CSN csn1 = gen.newCSN();
- CSN csn2 = gen.newCSN();
+ CSN csn2 = new CSN(csn1.getTime() + 1, 0, 202);
replayMsg(modDnMsg(entry, entryUUID, parentUUID, csn2, "uid=simultaneous2"));
+ assertFalse(domain.getServerState().cover(csn1));
// This MODIFY DN uses an older DN and should therefore be cancelled at replay time.
replayMsg(modDnMsg(entry, entryUUID, parentUUID, csn1, "uid=simulatneouswrong"));
@@ -287,10 +289,11 @@
final String entryUUID = getEntryUUID(entry.getName());
final int givenUpBefore = failedReplayedUpdates();
- // Two consecutive CSNs, replayed in the reverse order.
+ // Reverse order across origins must still reach naming-conflict resolution.
final CSN older = gen.newCSN();
- final CSN newer = gen.newCSN();
+ final CSN newer = new CSN(older.getTime() + 1, 0, 202);
replayMsg(modDnMsg(entry, entryUUID, parentUUID, newer, "cn=renamedAfter"));
+ assertFalse(domain.getServerState().cover(older));
replayMsg(modDnMsg(entry, entryUUID, parentUUID, older, "cn=renamedBefore"));
assertTrue(entryExists(DN.valueOf("cn=renamedAfter," + TEST_ROOT_DN_STRING)),
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java
index 6c1ed1c..b013fbd 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/RemotePendingChangesTest.java
@@ -19,9 +19,14 @@
import static org.testng.Assert.*;
import java.util.NoSuchElementException;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReentrantLock;
import org.forgerock.opendj.ldap.DN;
import org.opends.server.DirectoryServerTestCase;
@@ -33,6 +38,7 @@
import org.opends.server.replication.protocol.LDAPUpdateMsg;
import org.opends.server.replication.protocol.ModifyDNMsg;
import org.testng.annotations.BeforeClass;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
/**
@@ -68,6 +74,264 @@
assertEquals(pendingChanges.getQueueSize(), 0);
}
+ @Test
+ public void appliedChangesAreRejectedAfterLeavingThePendingList() throws Exception
+ {
+ final ServerState state = new ServerState();
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(state);
+ final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0);
+ final CSN first = generator.newCSN();
+ final CSN last = generator.newCSN();
+
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(first, "uuid-1")));
+ pendingChanges.commit(first);
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(last, "uuid-2")));
+ pendingChanges.commit(last);
+
+ assertFalse(pendingChanges.putRemoteUpdate(deleteMsg(first, "uuid-1")));
+ assertFalse(pendingChanges.putRemoteUpdate(deleteMsg(last, "uuid-2")));
+ assertEquals(pendingChanges.getQueueSize(), 0);
+
+ // The watermark belongs to the originating server, not to all servers at once.
+ final CSN anotherOrigin = new CSN(first.getTime(), first.getSeqnum(), SERVER_ID + 1);
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(anotherOrigin, "uuid-3")));
+ }
+
+ @Test
+ public void duplicateDoesNotReplaceAQueuedDelivery() throws Exception
+ {
+ final ServerState state = new ServerState();
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(state);
+ final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
+ final DeleteMsg original = deleteMsg(csn, "uuid-1");
+ final DeleteMsg duplicate = deleteMsg(csn, "uuid-1");
+
+ assertTrue(pendingChanges.putRemoteUpdate(original));
+ assertFalse(pendingChanges.putRemoteUpdate(duplicate));
+ assertFalse(pendingChanges.markInProgress(duplicate));
+ assertTrue(pendingChanges.markInProgress(original));
+ assertTrue(state.isEmpty(), "discarding a duplicate must not record the queued change as replayed");
+ pendingChanges.commit(csn);
+ assertTrue(state.cover(csn));
+ }
+
+ @Test
+ public void duplicateDoesNotReplaceAQueuedRetry() throws Exception
+ {
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(new ServerState());
+ final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
+ final DeleteMsg original = deleteMsg(csn, "uuid-1");
+ final DeleteMsg retry = deleteMsg(csn, "uuid-1");
+ final DeleteMsg duplicate = deleteMsg(csn, "uuid-1");
+
+ assertTrue(pendingChanges.putRemoteUpdate(original));
+ assertTrue(pendingChanges.markInProgress(original));
+ assertTrue(pendingChanges.replayFailed(csn));
+ assertTrue(pendingChanges.putRemoteUpdate(retry));
+ assertEquals(pendingChanges.changesInProgressSize(), 1, "a queued retry retains its in-progress count");
+ assertFalse(pendingChanges.putRemoteUpdate(duplicate));
+ assertFalse(pendingChanges.markInProgress(original));
+ assertFalse(pendingChanges.markInProgress(duplicate));
+ assertTrue(pendingChanges.markInProgress(retry));
+ }
+
+ @Test
+ public void interruptedClaimDoesNotLeaveTheDeliveryMarkedAsQueued() throws Exception
+ {
+ final OutOfMemoryError failure = new OutOfMemoryError("claim allocation failed");
+ final AtomicBoolean failClaim = new AtomicBoolean(true);
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(new ServerState(), -1,
+ new ConcurrentHashMap<Thread, CSN>()
+ {
+ @Override
+ public CSN put(Thread thread, CSN csn)
+ {
+ if (failClaim.getAndSet(false))
+ {
+ throw failure;
+ }
+ return super.put(thread, csn);
+ }
+ });
+ final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
+ final DeleteMsg original = deleteMsg(csn, "uuid-1");
+ assertTrue(pendingChanges.putRemoteUpdate(original));
+
+ try
+ {
+ pendingChanges.markInProgress(original);
+ fail("the injected allocation failure did not reach the claim");
+ }
+ catch (OutOfMemoryError e)
+ {
+ assertSame(e, failure);
+ }
+
+ final DeleteMsg retry = deleteMsg(csn, "uuid-1");
+ assertTrue(pendingChanges.putRemoteUpdate(retry), "the dequeued delivery has no owner to finish it");
+ assertTrue(pendingChanges.markInProgress(retry));
+ }
+
+ @DataProvider
+ public Object[][] registrationFailures()
+ {
+ return new Object[][] { { false }, { true } };
+ }
+
+ @Test(dataProvider = "registrationFailures")
+ public void failedDependencyRegistrationCancelsDeliveryBeforeRollback(boolean failRollback) throws Exception
+ {
+ final OutOfMemoryError registrationFailure = new OutOfMemoryError("dependency registration failed");
+ final OutOfMemoryError rollbackFailure = new OutOfMemoryError("dependency rollback failed");
+ final AtomicBoolean failRegistration = new AtomicBoolean(true);
+ final ServerState state = new ServerState();
+ final RemotePendingChanges pending = new RemotePendingChanges(state, -1, new ConcurrentHashMap<>(),
+ new ConcurrentSkipListSet<PendingChange>()
+ {
+ @Override
+ public boolean add(PendingChange change)
+ {
+ if (failRegistration.getAndSet(false))
+ {
+ // Also cover a set insertion that throws after linking its element.
+ super.add(change);
+ throw registrationFailure;
+ }
+ return super.add(change);
+ }
+
+ @Override
+ public boolean remove(Object change)
+ {
+ if (failRollback)
+ {
+ throw rollbackFailure;
+ }
+ return super.remove(change);
+ }
+ });
+ final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
+ final UpdateToReplay delivery = new UpdateToReplay(deleteMsg(csn, "uuid-1"), null);
+ try
+ {
+ pending.putRemoteUpdate(delivery);
+ fail("registration failure must escape admission");
+ }
+ catch (OutOfMemoryError e)
+ {
+ assertSame(e, failRollback ? rollbackFailure : registrationFailure);
+ }
+ assertTrue(delivery.isCancelled(), "rollback failure must not skip delivery cancellation");
+ assertFalse(state.cover(csn));
+ if (!failRollback)
+ {
+ assertEquals(pending.getQueueSize(), 0);
+ final DeleteMsg retry = deleteMsg(csn, "uuid-1");
+ assertTrue(pending.putRemoteUpdate(retry), "failed registration must not wedge the next delivery");
+ assertTrue(pending.markInProgress(retry));
+ pending.commit(csn);
+ assertTrue(state.cover(csn));
+ }
+ }
+
+ @Test
+ public void lateEnqueueCancellationDoesNotReleaseAnotherDeliveryOrItsOwner() throws Exception
+ {
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(new ServerState());
+ final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
+ final DeleteMsg original = deleteMsg(csn, "uuid-1");
+ final DeleteMsg retry = deleteMsg(csn, "uuid-1");
+ final UpdateToReplay oldDelivery = new UpdateToReplay(original, null);
+ final UpdateToReplay newDelivery = new UpdateToReplay(retry, null);
+ assertEquals(pendingChanges.putRemoteUpdate(oldDelivery), RemotePendingChanges.Admission.ACCEPTED);
+ oldDelivery.cancel();
+ assertEquals(pendingChanges.putRemoteUpdate(newDelivery), RemotePendingChanges.Admission.ACCEPTED);
+
+ oldDelivery.cancel();
+ assertFalse(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")),
+ "cancelling the previous delivery must not release its queued replacement");
+ assertTrue(pendingChanges.markInProgress(retry));
+
+ newDelivery.cancel();
+ assertFalse(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")),
+ "enqueue cancellation must not take a delivery away from its replay owner");
+ assertTrue(pendingChanges.replayFailed(csn));
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")));
+ }
+
+ @Test
+ public void resettingStateAllowsPreviouslyAppliedChangesToBeReplayed() throws Exception
+ {
+ final ServerState state = new ServerState();
+ final RemotePendingChanges pendingChanges = new RemotePendingChanges(state);
+ final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
+ assertTrue(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")));
+ pendingChanges.commit(csn);
+
+ // Disabling a domain clears pending changes, but retains its applied state.
+ pendingChanges.clear();
+ assertFalse(pendingChanges.putRemoteUpdate(deleteMsg(csn, "uuid-1")));
+
+ // A restore/import may reset that state so the same changes are needed again.
+ state.clear();
+ final DeleteMsg redelivered = deleteMsg(csn, "uuid-1");
+ assertTrue(pendingChanges.putRemoteUpdate(redelivered));
+ assertTrue(pendingChanges.markInProgress(redelivered));
+ }
+
+ @Test
+ public void failedClaimBeforeTakingThePendingLockReleasesItsDelivery() throws Exception
+ {
+ final RemotePendingChanges pending = new RemotePendingChanges(new ServerState());
+ final CSN csn = new CSNGenerator(SERVER_ID, 0).newCSN();
+ final DeleteMsg original = deleteMsg(csn, "uuid-1");
+ final OutOfMemoryError failure = new OutOfMemoryError("pending lock allocation failed");
+ // Inject between dequeue and claim, before RemotePendingChanges can acquire its lock.
+ final UpdateToReplay delivery = new UpdateToReplay(original, null)
+ {
+ @Override
+ public LDAPReplicationDomain getReplicationDomain()
+ {
+ throw failure;
+ }
+ };
+ assertEquals(pending.putRemoteUpdate(delivery), RemotePendingChanges.Admission.ACCEPTED);
+ final LinkedBlockingQueue<UpdateToReplay> queue = new LinkedBlockingQueue<>();
+ queue.add(delivery);
+ try
+ {
+ new ReplayThread(queue, new ReentrantLock()).run();
+ fail("the failed claim must end the replay thread");
+ }
+ catch (OutOfMemoryError e)
+ {
+ assertSame(e, failure);
+ }
+ assertTrue(queue.isEmpty());
+ assertFalse(pending.markInProgress(original), "a cancelled delivery must not acquire ownership");
+ final DeleteMsg retry = deleteMsg(csn, "uuid-1");
+ assertTrue(pending.putRemoteUpdate(retry), "the failed claim must not block redelivery");
+ assertTrue(pending.markInProgress(retry));
+ }
+
+ @Test
+ public void ownReplayWatermarkSurvivesClearAndResetsWithBackendState() throws Exception
+ {
+ final ServerState state = new ServerState();
+ final CSNGenerator generator = new CSNGenerator(SERVER_ID, 0);
+ final CSN first = generator.newCSN();
+ final CSN later = generator.newCSN();
+ final RemotePendingChanges pending = new RemotePendingChanges(state, SERVER_ID);
+ state.update(later); // A local publication does not prove that first was replayed.
+ assertTrue(pending.putRemoteUpdate(deleteMsg(first, "uuid-1")));
+ pending.commit(first);
+ pending.clear();
+ assertFalse(pending.putRemoteUpdate(deleteMsg(first, "uuid-1")));
+ state.clear();
+ pending.resetReplayState();
+ assertTrue(pending.putRemoteUpdate(deleteMsg(first, "uuid-1")));
+ }
+
/**
* A change which is not committed must hold back the ServerState, even when the
* changes which follow it have been replayed: the replication server resumes from the
@@ -180,9 +444,8 @@
}
/**
- * The replication server sends the change again over the new session while the message
- * of the previous delivery may still be waiting in the replay queue: only the delivery
- * which is listed as pending is replayed, or the same change would be applied twice.
+ * A failed delivery must not claim the change after the replication server has sent
+ * a replacement over the new session: only the replacement may replay it.
*/
@Test
public void markInProgressRejectsThePreviousDeliveryOfAChange() throws Exception
@@ -193,7 +456,8 @@
final DeleteMsg newDelivery = deleteMsg(csn, "uuid-1");
assertTrue(pendingChanges.putRemoteUpdate(previousDelivery));
- pendingChanges.replayFailed(csn);
+ assertTrue(pendingChanges.markInProgress(previousDelivery));
+ assertTrue(pendingChanges.replayFailed(csn));
assertTrue(pendingChanges.putRemoteUpdate(newDelivery), "the next delivery must be replayed");
assertFalse(pendingChanges.markInProgress(previousDelivery),
--
Gitblit v1.10.0