From c6650aedac5c1ade8aee9e7a76eb0dd100a3e179 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Mon, 02 Nov 2026 15:39:56 +0000
Subject: [PATCH] [#1246] Keep the messages of the last local changes, and have a recovery publish the changes done before it began from them, as they were done (#1251)
---
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java | 621 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 620 insertions(+), 1 deletions(-)
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java
index 0919df6..8f024b5 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java
@@ -12,6 +12,7 @@
* information: "Portions copyright [year] [name of copyright owner]".
*
* Copyright 2026 3A Systems, LLC.
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.replication.plugin;
@@ -24,7 +25,9 @@
import static org.testng.Assert.*;
import java.util.ArrayList;
+import java.lang.reflect.Field;
import java.util.List;
+import java.util.Map;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.CountDownLatch;
@@ -38,6 +41,7 @@
import org.forgerock.opendj.server.config.meta.ReplicationDomainCfgDefn.IsolationPolicy;
import org.opends.server.TestCaseUtils;
import org.opends.server.core.DirectoryServer;
+import org.opends.server.core.LockFileManager;
import org.opends.server.plugins.ShortCircuitPlugin;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.common.CSN;
@@ -57,6 +61,7 @@
import org.opends.server.types.DirectoryException;
import org.opends.server.types.Entry;
import org.opends.server.types.OperationType;
+import org.opends.server.tasks.TaskUtils;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
@@ -121,6 +126,620 @@
private ReplicationBroker exporter;
private CSNGenerator gen;
+ @Test(timeOut = 120_000)
+ public void aFailedImportUnlockForgetsMessagesOfTheReplacedData() throws Exception
+ {
+ TestCaseUtils.addEntry("dn: cn=before," + EXAMPLE_DN, "objectClass: top", "objectClass: person",
+ "cn: before", "sn: before");
+ final CSN change = domain.getLastLocalChange();
+ final CSN before = new CSN(change.getTime(), change.getSeqnum() - 1, DS_ID);
+ final Field pendingField = LDAPReplicationDomain.class.getDeclaredField("pendingChanges");
+ pendingField.setAccessible(true);
+ final PendingChanges pending = (PendingChanges) pendingField.get(domain);
+ assertThat(pending.recentChangesAfter(before, change).messages).hasSize(1);
+
+ final String lockFile = LockFileManager.getBackendLockFileName(
+ getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN));
+ final Field locksField = LockFileManager.class.getDeclaredField("exclusiveLocks");
+ final Field mapLockField = LockFileManager.class.getDeclaredField("mapLock");
+ locksField.setAccessible(true);
+ mapLockField.setAccessible(true);
+ final Map<?, ?> locks = (Map<?, ?>) locksField.get(null);
+ final Object mapLock = mapLockField.get(null);
+ try
+ {
+ final String[] exported = exportedEntries();
+ final CSN importedState = gen.newCSN();
+ exported[0] = exported[0].replace("dc: example\n", "dc: example\nds-sync-state: " + importedState + "\n");
+ startImportInto(exported.length);
+ waitUntil(() -> {
+ synchronized (mapLock)
+ {
+ return locks.containsKey(lockFile);
+ }
+ }, "the import did not acquire its exclusive lock");
+ assertTrue(LockFileManager.releaseLock(lockFile, new StringBuilder()));
+ finishImport(exported);
+ assertNull(getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN),
+ "the failed unlock must leave the backend disabled");
+ assertThat(errorLogRecordsOf(org.opends.messages.ToolMessages.WARN_LDIFIMPORT_CANNOT_UNLOCK_BACKEND.ordinal()))
+ .isNotEmpty();
+ assertNull(pending.recentChangesAfter(before, change).messages,
+ "the import replaced the data even though its lock could not be released");
+ assertFalse(domain.isConnected(), "a failed cleanup must stop the replication session");
+ TaskUtils.enableBackend("userRoot");
+ domain.enable();
+ domain.enableService();
+ assertFalse(domain.isConnected(), "enabling the backend or service must not revive a failed import");
+ assertEquals(org.opends.server.protocols.internal.InternalClientConnection.getRootConnection()
+ .processDelete(IMPORTED_ENTRY_DN).getResultCode(), ResultCode.UNAVAILABLE,
+ "writes must be rejected even when the backend is enabled again");
+ domain.backupStart();
+ domain.disable();
+ assertEquals(new PersistentServerState(baseDN, DS_ID,
+ new org.opends.server.replication.common.ServerState()).getMaxCSN(importedState.getServerId()),
+ importedState, "backup or disable must not overwrite the imported server state");
+ MultimasterReplication.deleteDomain(baseDN);
+ assertEquals(org.opends.server.protocols.internal.InternalClientConnection.getRootConnection()
+ .processDelete(IMPORTED_ENTRY_DN).getResultCode(), ResultCode.UNAVAILABLE,
+ "deleting the replication domain must not permit writes to quarantined data");
+ final DomainFakeCfg replacement = new DomainFakeCfg(baseDN, DS_ID, new TreeSet<String>());
+ assertNull(MultimasterReplication.createNewDomain(replacement),
+ "recreating the domain must not clear the failed-import quarantine");
+ assertEquals(new MultimasterReplication().applyConfigurationAdd(replacement).getResultCode(),
+ ResultCode.UNAVAILABLE, "configuration add must report failure without dereferencing a missing domain");
+ }
+ finally
+ {
+ // A process restart clears the production quarantine; this test shares its JVM with other cases.
+ clearWriteFailure();
+ final Field ignored = LDAPReplicationDomain.class.getDeclaredField("ignoreBackendInitializationEvent");
+ ignored.setAccessible(true);
+ ignored.setBoolean(domain, false);
+ if (getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN) == null)
+ {
+ TaskUtils.enableBackend("userRoot");
+ }
+ }
+ }
+
+ @Test(timeOut = 120_000)
+ public void anUnavailableStateReadAfterImportStopsTheDomain() throws Exception
+ {
+ final String[] exported = exportedEntries();
+ startImportInto(exported.length);
+ ShortCircuitPlugin.registerShortCircuit(OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue(),
+ 0, 1, operation -> {
+ final org.opends.server.types.operation.PreParseSearchOperation search =
+ (org.opends.server.types.operation.PreParseSearchOperation) operation;
+ return search.getRawBaseDN().toString().equals(EXAMPLE_DN)
+ && search.getAttributes().contains("ds-sync-state");
+ });
+ try
+ {
+ finishImport(exported);
+ assertTrue(ShortCircuitPlugin.getShortCircuitCount(OperationType.SEARCH, "PreParse") > 0,
+ "the state-read failure was not injected");
+ assertFalse(domain.isConnected(), "an unavailable state read must not fall back to stale configuration state");
+ assertNotNull(org.opends.server.api.LocalBackend.getWriteFailure(baseDN));
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse");
+ clearWriteFailure();
+ }
+ }
+
+ @Test(timeOut = 120_000)
+ public void anImportWhichCannotAcquireItsLockDoesNotReleaseAnotherOwnersLock() throws Exception
+ {
+ TestCaseUtils.addEntry("dn: cn=original," + EXAMPLE_DN,
+ "objectClass: person", "cn: original", "sn: original");
+ final String lockFile = LockFileManager.getBackendLockFileName(
+ getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN));
+ assertTrue(LockFileManager.acquireSharedLock(lockFile, new StringBuilder()));
+ try
+ {
+ final int completions = errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size();
+ exporter.publish(new InitializeTargetMsg(baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, 3, INIT_WINDOW));
+ waitUntil(() -> !domain.ieRunning()
+ && errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size() > completions,
+ "the import did not report its failed exclusive lock");
+ assertNotNull(getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN));
+ assertTrue(entryExists(DN.valueOf("cn=original," + EXAMPLE_DN)));
+ assertNull(org.opends.server.api.LocalBackend.getWriteFailure(baseDN));
+ final Field references = LockFileManager.class.getDeclaredField("sharedLockReferences");
+ final Field mapLock = LockFileManager.class.getDeclaredField("mapLock");
+ references.setAccessible(true);
+ mapLock.setAccessible(true);
+ synchronized (mapLock.get(null))
+ {
+ assertEquals(((Map<?, ?>) references.get(null)).get(lockFile), Integer.valueOf(2),
+ "the backend and the other owner must both retain their shared locks");
+ }
+ }
+ finally
+ {
+ assertTrue(LockFileManager.releaseLock(lockFile, new StringBuilder()));
+ }
+ }
+
+ @org.testng.annotations.DataProvider(name = "failedPreparation")
+ public Object[][] failedPreparation()
+ {
+ return new Object[][] { { false, false }, { true, false }, { true, true } };
+ }
+
+ /** A failed preparation must not quarantine data which the import never replaced. */
+ @Test(dataProvider = "failedPreparation", timeOut = 120_000)
+ public void aFailedPreparationCleanupLeavesTheOriginalDataRecoverable(boolean disabledBackend, boolean nestedBackend) throws Exception
+ {
+ if (nestedBackend)
+ {
+ TestCaseUtils.initializeMemoryBackend("importParent", "dc=com", true);
+ }
+ final Entry original = TestCaseUtils.addEntry("dn: cn=original," + EXAMPLE_DN,
+ "objectClass: person", "cn: original", "sn: original");
+ if (disabledBackend)
+ {
+ // Isolate the abandoned listing from legitimate redelivery of uncheckpointed local adds.
+ final Field stateField = LDAPReplicationDomain.class.getDeclaredField("state");
+ stateField.setAccessible(true);
+ ((PersistentServerState) stateField.get(domain)).save();
+ // A delivery taken from the test queue has no replay thread to finish it. The next
+ // session normally redelivers it; this fixture isolates the cleanup of the old listing.
+ domain.processUpdate(new ModifyMsg(gen.newCSN(), original.getName(),
+ generatemods("description", "before failed preparation"), getEntryUUID(original.getName())));
+ queue.take();
+ }
+ final int stoppedRecords = errorLogRecordsOf(ERR_INIT_DOMAIN_AWAITING_BACKEND.ordinal()).size();
+ final String lockFile = LockFileManager.getBackendLockFileName(
+ getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN));
+ boolean lockHeld = disabledBackend;
+ if (disabledBackend)
+ {
+ assertTrue(LockFileManager.acquireSharedLock(lockFile, new StringBuilder()));
+ }
+ ShortCircuitPlugin.registerShortCircuit(OperationType.MODIFY, "PreParse", ResultCode.UNAVAILABLE.intValue(),
+ disabledBackend ? 1 : 0, 2, operation -> {
+ final org.opends.server.types.operation.PreParseModifyOperation modify =
+ (org.opends.server.types.operation.PreParseModifyOperation) operation;
+ return DN.valueOf(modify.getRawEntryDN().toString())
+ .equals(DN.valueOf("ds-cfg-backend-id=userRoot,cn=Backends,cn=config"));
+ });
+ try
+ {
+ final int completions = errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size();
+ exporter.publish(new InitializeTargetMsg(baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, 3, INIT_WINDOW));
+ waitUntil(() -> !domain.ieRunning()
+ && errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size() > completions,
+ "the failed preparation did not complete");
+ assertTrue(ShortCircuitPlugin.getShortCircuitCount(OperationType.MODIFY, "PreParse") > 0);
+ assertNull(org.opends.server.api.LocalBackend.getWriteFailure(baseDN),
+ "an import which replaced no data must not quarantine the original suffix");
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.MODIFY, "PreParse");
+ if (disabledBackend)
+ {
+ assertFalse(domain.isConnected(), "a domain with an unavailable backend must be stopped reversibly");
+ assertThat(errorLogRecordsOf(ERR_INIT_DOMAIN_AWAITING_BACKEND.ordinal()).size()).isGreaterThan(stoppedRecords);
+ assertThat(errorLogRecordsOf(ERR_INIT_DOMAIN_AWAITING_BACKEND.ordinal()).get(stoppedRecords))
+ .contains("severity=ERROR", baseDN.toString(), "userRoot", "disable", "enable");
+ if (nestedBackend)
+ {
+ final org.opends.server.api.LocalBackend<?> parent =
+ getServerContext().getBackendConfigManager().getLocalBackendById("importParent");
+ domain.performBackendPreFinalizationProcessing(parent);
+ domain.performBackendPostInitializationProcessing(parent);
+ assertFalse(domain.isConnected(), "enabling the parent must not revive the missing child backend");
+ }
+ else
+ {
+ // A missing top-level backend must not break another backend's removal.
+ TestCaseUtils.initializeMemoryBackend("foreignImportFinalization", "o=foreignImportFinalization", true);
+ final org.opends.server.api.LocalBackend<?> foreign =
+ getServerContext().getBackendConfigManager().getLocalBackendById("foreignImportFinalization");
+ try
+ {
+ domain.performBackendPreFinalizationProcessing(foreign);
+ assertFalse(domain.isConnected(), "a foreign backend must not restore this domain");
+ }
+ finally
+ {
+ foreign.finalizeBackend();
+ getServerContext().getBackendConfigManager().deregisterLocalBackend(foreign);
+ }
+ }
+ TaskUtils.enableBackend("userRoot");
+ assertTrue(LockFileManager.releaseLock(lockFile, new StringBuilder()));
+ lockHeld = false;
+ }
+ waitUntil(domain::isConnected, "the backend enable must restore replication without a process restart");
+ assertTrue(entryExists(original.getName()));
+ final ModifyMsg after = new ModifyMsg(gen.newCSN(), original.getName(),
+ generatemods("description", "after failed preparation"), getEntryUUID(original.getName()));
+ replayMsg(after);
+ assertTrue(domain.getServerState().cover(after.getCSN()),
+ "a delivery abandoned by the old session must not hold back the new session's state");
+ TestCaseUtils.addEntry("dn: cn=after," + EXAMPLE_DN,
+ "objectClass: person", "cn: after", "sn: after");
+ // The restored listener must consume new traffic, not just keep a connected broker.
+ startImportInto(exportedEntries().length);
+ finishImport(exportedEntries());
+ assertTrue(entryExists(DN.valueOf(IMPORTED_ENTRY_DN)));
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.MODIFY, "PreParse");
+ clearWriteFailure();
+ final Field ignored = LDAPReplicationDomain.class.getDeclaredField("ignoreBackendInitializationEvent");
+ ignored.setAccessible(true);
+ ignored.setBoolean(domain, false);
+ if (lockHeld)
+ {
+ assertTrue(LockFileManager.releaseLock(lockFile, new StringBuilder()));
+ }
+ if (!getServerContext().getBackendConfigManager().hasLocalBackend("userRoot"))
+ {
+ TaskUtils.enableBackend("userRoot");
+ }
+ if (nestedBackend)
+ {
+ final org.opends.server.api.LocalBackend<?> parent =
+ getServerContext().getBackendConfigManager().getLocalBackendById("importParent");
+ parent.finalizeBackend();
+ getServerContext().getBackendConfigManager().deregisterLocalBackend(parent);
+ }
+ }
+ }
+
+ /** Backend callbacks run under the configuration lock and must skip foreign domain locks. */
+ @Test
+ public void aForeignBackendCallbackDoesNotWaitForThisDomainsServiceLock() throws Exception
+ {
+ TestCaseUtils.initializeMemoryBackend("foreignImportCallback", "o=foreignImportCallback", true);
+ final org.opends.server.api.LocalBackend<?> foreign =
+ getServerContext().getBackendConfigManager().getLocalBackendById("foreignImportCallback");
+ final Field serviceLock = org.opends.server.replication.service.ReplicationDomain.class
+ .getDeclaredField("serviceStateLock");
+ serviceLock.setAccessible(true);
+ final CountDownLatch completed = new CountDownLatch(1);
+ final Thread callback = new Thread(() -> {
+ domain.performBackendPostInitializationProcessing(foreign);
+ completed.countDown();
+ }, "foreign backend initialization callback");
+ try
+ {
+ synchronized (serviceLock.get(domain))
+ {
+ callback.start();
+ assertTrue(completed.await(5, TimeUnit.SECONDS), "a foreign callback waited for the domain's service lock");
+ }
+ }
+ finally
+ {
+ callback.join(10000);
+ assertFalse(callback.isAlive());
+ foreign.finalizeBackend();
+ getServerContext().getBackendConfigManager().deregisterLocalBackend(foreign);
+ }
+ }
+
+ @org.testng.annotations.DataProvider(name = "abortedImportStop")
+ public Object[][] abortedImportStop()
+ {
+ return new Object[][] { { true }, { false } };
+ }
+
+ /** An enable during cleanup is retained, and stopping on the listener never needs window credit. */
+ @Test(dataProvider = "abortedImportStop", timeOut = 120_000)
+ public void anAbortedImportStopsWithoutCreditAndRetainsABackendEnable(boolean enableDuringStop) throws Exception
+ {
+ final String lockFile = LockFileManager.getBackendLockFileName(
+ getServerContext().getBackendConfigManager().getLocalBackendById("userRoot"));
+ assertTrue(LockFileManager.acquireSharedLock(lockFile, new StringBuilder()));
+ boolean lockHeld = true;
+ final Field brokerField = org.opends.server.replication.service.ReplicationDomain.class.getDeclaredField("broker");
+ brokerField.setAccessible(true);
+ final ReplicationBroker broker = (ReplicationBroker) brokerField.get(domain);
+ final Field stopLockField = ReplicationBroker.class.getDeclaredField("startStopLock");
+ stopLockField.setAccessible(true);
+ final Field windowField = ReplicationBroker.class.getDeclaredField("sendWindow");
+ windowField.setAccessible(true);
+ final java.util.concurrent.Semaphore window = (java.util.concurrent.Semaphore) windowField.get(broker);
+ final Field disabledField = LDAPReplicationDomain.class.getDeclaredField("disabled");
+ disabledField.setAccessible(true);
+ final long timeout = MultimasterReplication.getConnectionTimeoutMS();
+ final DN provider = DN.valueOf("cn=Multimaster Synchronization,cn=Synchronization Providers,cn=config");
+ ShortCircuitPlugin.registerShortCircuit(OperationType.MODIFY, "PreParse", ResultCode.UNAVAILABLE.intValue(),
+ 1, 2, operation -> DN.valueOf(((org.opends.server.types.operation.PreParseModifyOperation) operation)
+ .getRawEntryDN().toString()).equals(DN.valueOf("ds-cfg-backend-id=userRoot,cn=Backends,cn=config")));
+ try
+ {
+ assertEquals(org.opends.server.protocols.internal.InternalClientConnection.getRootConnection().processModify(
+ modifyRequest(provider,
+ org.forgerock.opendj.ldap.ModificationType.REPLACE, "ds-cfg-connection-timeout", "0 ms"))
+ .getResultCode(), ResultCode.SUCCESS);
+ final int completions = errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size();
+ synchronized (stopLockField.get(broker))
+ {
+ exporter.publish(new InitializeTargetMsg(baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, 3, INIT_WINDOW));
+ new org.opends.server.util.TestTimer.Builder().maxSleep(10, TimeUnit.SECONDS)
+ .sleepTimes(10, TimeUnit.MILLISECONDS).toTimer()
+ .repeatUntilSuccess(() -> assertTrue(disabledField.getBoolean(domain)));
+ if (enableDuringStop)
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.MODIFY, "PreParse");
+ TaskUtils.enableBackend("userRoot");
+ }
+ else
+ {
+ // No update is needed: hold the actual window shut at the instant stop starts.
+ window.drainPermits();
+ }
+ }
+ new org.opends.server.util.TestTimer.Builder().maxSleep(5, TimeUnit.SECONDS)
+ .sleepTimes(10, TimeUnit.MILLISECONDS).toTimer()
+ .repeatUntilSuccess(() -> assertTrue(!domain.ieRunning()
+ && errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size() > completions,
+ "the listener must finish import cleanup without receiving window credit"));
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.MODIFY, "PreParse");
+ if (!enableDuringStop)
+ {
+ TaskUtils.enableBackend("userRoot");
+ }
+ new org.opends.server.util.TestTimer.Builder().maxSleep(10, TimeUnit.SECONDS)
+ .sleepTimes(10, TimeUnit.MILLISECONDS).toTimer()
+ .repeatUntilSuccess(() -> assertTrue(domain.isConnected(), "the backend enable must restore the domain"));
+ assertTrue(LockFileManager.releaseLock(lockFile, new StringBuilder()));
+ lockHeld = false;
+ startImportInto(exportedEntries().length);
+ finishImport(exportedEntries());
+ assertTrue(entryExists(DN.valueOf(IMPORTED_ENTRY_DN)));
+ }
+ finally
+ {
+ // Release a broken implementation too, so the negative run cannot wedge the test JVM.
+ window.release(100000);
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.MODIFY, "PreParse");
+ waitUntil(() -> !domain.ieRunning(), "import cleanup did not finish after credit was restored");
+ final Field ignored = LDAPReplicationDomain.class.getDeclaredField("ignoreBackendInitializationEvent");
+ ignored.setAccessible(true);
+ ignored.setBoolean(domain, false);
+ if (!getServerContext().getBackendConfigManager().hasLocalBackend("userRoot"))
+ {
+ TaskUtils.enableBackend("userRoot");
+ }
+ if (lockHeld)
+ {
+ assertTrue(LockFileManager.releaseLock(lockFile, new StringBuilder()));
+ }
+ assertEquals(org.opends.server.protocols.internal.InternalClientConnection.getRootConnection().processModify(
+ modifyRequest(provider,
+ org.forgerock.opendj.ldap.ModificationType.REPLACE, "ds-cfg-connection-timeout", timeout + " ms"))
+ .getResultCode(), ResultCode.SUCCESS);
+ }
+ }
+
+ /** The reconnect after an abandoned import already satisfies replay's outstanding restart. */
+ @Test(timeOut = 120_000)
+ public void anAbandonedImportConsumesTheReplayRestart() throws Exception
+ {
+ final Entry original = TestCaseUtils.addEntry("dn: cn=original," + EXAMPLE_DN,
+ "objectClass: person", "cn: original", "sn: original");
+ final Field stateField = LDAPReplicationDomain.class.getDeclaredField("state");
+ stateField.setAccessible(true);
+ final Object state = stateField.get(domain);
+ final Field saveLockField = PersistentServerState.class.getDeclaredField("saveLock");
+ saveLockField.setAccessible(true);
+ final java.util.concurrent.locks.Lock saveLock = (java.util.concurrent.locks.Lock) saveLockField.get(state);
+ final Field importing = LDAPReplicationDomain.class.getDeclaredField("importingData");
+ importing.setAccessible(true);
+ final CSN csn = gen.newCSN();
+ final ModifyMsg msg = new ModifyMsg(csn, original.getName(), generatemods("description", "replayed"),
+ getEntryUUID(original.getName()));
+ final ModifyMsg queued = new ModifyMsg(gen.newCSN(), original.getName(),
+ generatemods("description", "queued before the import"), getEntryUUID(original.getName()));
+ domain.processUpdate(queued);
+ final LDAPUpdateMsg queuedDelivery = queue.take().getUpdateMessage();
+ final int completions = errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size();
+ domain.setReplayDrainTimeout(3000);
+ saveLock.lock();
+ try
+ {
+ exporter.publish(new InitializeTargetMsg(baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, 3, INIT_WINDOW));
+ new org.opends.server.util.TestTimer.Builder().maxSleep(30, TimeUnit.SECONDS)
+ .sleepTimes(10, TimeUnit.MILLISECONDS).toTimer()
+ .repeatUntilSuccess(() -> assertTrue(importing.getBoolean(domain)));
+ replayMsg(msg);
+ assertFalse(domain.getServerState().cover(csn), "the replay must be deferred by the import");
+ domain.failNextSessionRestarts(1);
+ waitUntil(() -> !domain.ieRunning()
+ && errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size() > completions,
+ "the import did not abandon its wait for the state save");
+ Thread.sleep(2000);
+ assertEquals(domain.getSessionRestartFailuresLeft(), 1,
+ "the checkpointer restarted the session which import completion already restarted");
+ assertTrue(domain.isConnected());
+ replayMsg(msg);
+ // This delivery was already queued before the import: it must still be listed, without
+ // processUpdate inserting it again. Clearing all pending changes would discard its replay.
+ replay(queuedDelivery);
+ assertTrue(domain.getServerState().cover(queued.getCSN()));
+ assertThat(DirectoryServer.getEntry(original.getName()).getAllAttributes("description")).isNotEmpty();
+ }
+ finally
+ {
+ saveLock.unlock();
+ domain.failNextSessionRestarts(0);
+ domain.setReplayDrainTimeout(30_000);
+ }
+ }
+
+ @org.testng.annotations.DataProvider(name = "importDrains")
+ public Object[][] importDrains()
+ {
+ return new Object[][] { { false }, { true } };
+ }
+
+ @Test(dataProvider = "importDrains", timeOut = 120_000)
+ public void aTimedOutDrainLeavesTheOriginalDataUsable(boolean stateSave) throws Exception
+ {
+ final Entry original = TestCaseUtils.makeEntry("dn: cn=original," + EXAMPLE_DN,
+ "objectClass: top", "objectClass: person", "cn: original", "sn: original");
+ TestCaseUtils.addEntry(original);
+ final Object owner;
+ final Field lockField;
+ if (stateSave)
+ {
+ final Field stateField = LDAPReplicationDomain.class.getDeclaredField("state");
+ stateField.setAccessible(true);
+ owner = stateField.get(domain);
+ lockField = PersistentServerState.class.getDeclaredField("saveLock");
+ }
+ else
+ {
+ owner = domain;
+ lockField = LDAPReplicationDomain.class.getDeclaredField("replayReadLock");
+ }
+ lockField.setAccessible(true);
+ final java.util.concurrent.locks.Lock lock = (java.util.concurrent.locks.Lock) lockField.get(owner);
+ final Object backend = getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN);
+ final int completions = errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size();
+ final int shutdownWarnings = errorLogRecordsOf(WARN_REPLAY_NOT_DRAINED.ordinal()).size();
+ final int drainErrors = errorLogRecordsOf(ERR_INIT_OPERATIONS_NOT_DRAINED.ordinal()).size();
+ domain.setReplayDrainTimeout(100);
+ lock.lock();
+ try
+ {
+ exporter.publish(new InitializeTargetMsg(baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, 3, INIT_WINDOW));
+ waitUntil(() -> !domain.ieRunning()
+ && errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size() > completions,
+ "the import did not abandon the drain before replacing data");
+ assertThat(errorLogRecordsOf(ERR_INIT_OPERATIONS_NOT_DRAINED.ordinal()).size()).isGreaterThan(drainErrors);
+ assertThat(errorLogRecordsOf(ERR_INIT_OPERATIONS_NOT_DRAINED.ordinal()).get(drainErrors))
+ .contains("severity=ERROR");
+ assertSame(getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN), backend);
+ assertTrue(entryExists(original.getName()), "a rejected preflight must leave the original data intact");
+ assertNull(org.opends.server.api.LocalBackend.getWriteFailure(baseDN));
+ assertThat(errorLogRecordsOf(WARN_REPLAY_NOT_DRAINED.ordinal())).hasSize(shutdownWarnings);
+ final ErrorMsg failure = waitForSpecificMsg(exporter, ErrorMsg.class);
+ assertEquals(failure.getDetails().toString(), ERR_INIT_OPERATIONS_NOT_DRAINED.get(baseDN).toString());
+ }
+ finally
+ {
+ lock.unlock();
+ domain.setReplayDrainTimeout(30_000);
+ }
+ waitUntil(domain::isConnected, "a preflight failure must leave the domain recoverable");
+ final String[] exported = exportedEntries();
+ startImportInto(exported.length);
+ finishImport(exported);
+ assertTrue(entryExists(DN.valueOf(IMPORTED_ENTRY_DN)), "a later import must still work");
+ }
+
+ @org.testng.annotations.DataProvider(name = "interruptedPreparation")
+ public Object[][] interruptedPreparation()
+ {
+ return new Object[][] { { false }, { true } };
+ }
+
+ @Test(dataProvider = "interruptedPreparation", timeOut = 120_000)
+ public void anInterruptedImportDrainReportsTheInterrupt(boolean waitingForSave) throws Exception
+ {
+ final Field stateField = LDAPReplicationDomain.class.getDeclaredField("state");
+ stateField.setAccessible(true);
+ final Object state = stateField.get(domain);
+ final Field suspended = PersistentServerState.class.getDeclaredField("savesSuspended");
+ suspended.setAccessible(true);
+ final Field lockField = waitingForSave
+ ? PersistentServerState.class.getDeclaredField("saveLock")
+ : LDAPReplicationDomain.class.getDeclaredField("replayReadLock");
+ lockField.setAccessible(true);
+ final java.util.concurrent.locks.Lock lock =
+ (java.util.concurrent.locks.Lock) lockField.get(waitingForSave ? state : domain);
+ final Field listenerField = org.opends.server.replication.service.ReplicationDomain.class
+ .getDeclaredField("listenerThread");
+ listenerField.setAccessible(true);
+ final Thread listener = (Thread) listenerField.get(domain);
+ final Field importing = LDAPReplicationDomain.class.getDeclaredField("importingData");
+ importing.setAccessible(true);
+ final int completions = errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size();
+ final int interrupts = errorLogRecordsOf(ERR_INIT_PREPARATION_INTERRUPTED.ordinal()).size();
+ lock.lock();
+ try
+ {
+ exporter.publish(new InitializeTargetMsg(baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, 3, INIT_WINDOW));
+ new org.opends.server.util.TestTimer.Builder().maxSleep(10, TimeUnit.SECONDS)
+ .sleepTimes(10, TimeUnit.MILLISECONDS).toTimer()
+ .repeatUntilSuccess(() -> {
+ assertTrue(importing.getBoolean(domain));
+ if (waitingForSave)
+ {
+ assertTrue(suspended.getBoolean(state), "the replay drain must finish before interruption");
+ assertEquals(listener.getState(), Thread.State.TIMED_WAITING);
+ }
+ });
+ listener.interrupt();
+ waitUntil(() -> !domain.ieRunning()
+ && errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()).size() > completions,
+ "the interrupted import did not finish");
+ final List<String> records = errorLogRecordsOf(ERR_INIT_PREPARATION_INTERRUPTED.ordinal());
+ assertThat(records.subList(interrupts, records.size()))
+ .isNotEmpty().allSatisfy(record -> assertThat(record).contains("severity=ERROR", "interrupted"));
+ assertNull(org.opends.server.api.LocalBackend.getWriteFailure(baseDN));
+ }
+ finally
+ {
+ lock.unlock();
+ }
+ waitUntil(domain::isConnected, "an interrupted preparation must leave the domain recoverable");
+ startImportInto(exportedEntries().length);
+ finishImport(exportedEntries());
+ assertTrue(entryExists(DN.valueOf(IMPORTED_ENTRY_DN)));
+ }
+
+ @Test
+ public void writeQuarantineSurvivesProviderRemovalAndRejectsMovesIntoTheSuffix() throws Exception
+ {
+ TestCaseUtils.addEntry("dn: ou=People," + EXAMPLE_DN, "objectClass: organizationalUnit", "ou: People");
+ TestCaseUtils.addEntry("dn: cn=outside," + EXAMPLE_DN,
+ "objectClass: person", "cn: outside", "sn: outside");
+ final DN quarantined = DN.valueOf("ou=People," + EXAMPLE_DN);
+ final java.util.List<org.opends.server.api.SynchronizationProvider<
+ org.forgerock.opendj.server.config.server.SynchronizationProviderCfg>> providers =
+ new ArrayList<>(getSynchronizationProviders());
+ org.opends.server.api.LocalBackend.quarantineWrites(quarantined,
+ org.forgerock.i18n.LocalizableMessage.raw("failed import test"));
+ try
+ {
+ for (org.opends.server.api.SynchronizationProvider<?> provider : providers)
+ {
+ deregisterSynchronizationProvider(provider);
+ }
+ final org.opends.server.protocols.internal.InternalClientConnection connection =
+ org.opends.server.protocols.internal.InternalClientConnection.getRootConnection();
+ assertEquals(connection.processDelete(quarantined).getResultCode(), ResultCode.UNAVAILABLE);
+ assertEquals(connection.processModifyDN("cn=outside," + EXAMPLE_DN, "cn=outside", true,
+ quarantined.toString()).getResultCode(), ResultCode.UNAVAILABLE,
+ "a move into quarantined data must be refused even when its source is healthy");
+ assertTrue(entryExists(DN.valueOf("cn=outside," + EXAMPLE_DN)));
+ }
+ finally
+ {
+ getSynchronizationProviders().addAll(providers);
+ final Field failures = org.opends.server.api.LocalBackend.class.getDeclaredField("writeFailures");
+ failures.setAccessible(true);
+ ((Map<?, ?>) failures.get(null)).remove(quarantined);
+ }
+ }
+
+ private void clearWriteFailure() throws Exception
+ {
+ final Field failures = org.opends.server.api.LocalBackend.class.getDeclaredField("writeFailures");
+ failures.setAccessible(true);
+ ((Map<?, ?>) failures.get(null)).remove(baseDN);
+ }
+
@BeforeMethod
public void setUpLocal() throws Exception
{
@@ -1120,7 +1739,7 @@
exporter.publish(new InitializeTargetMsg(
baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, entryCount, INIT_WINDOW));
final long deadline = System.currentTimeMillis() + 30_000;
- while (getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN) != null)
+ while (getServerContext().getBackendConfigManager().hasLocalBackend("userRoot"))
{
assertTrue(System.currentTimeMillis() < deadline,
"the import did not deregister the backend of the domain");
--
Gitblit v1.10.0