/*
|
* 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 java.util.concurrent.TimeUnit.*;
|
|
import static org.opends.server.util.CollectionUtils.*;
|
import static org.opends.server.util.StaticUtils.*;
|
import static org.testng.Assert.*;
|
|
import java.util.SortedSet;
|
import java.util.concurrent.CountDownLatch;
|
import java.util.concurrent.atomic.AtomicInteger;
|
import java.util.concurrent.atomic.AtomicReference;
|
|
import org.forgerock.opendj.ldap.DN;
|
import org.opends.server.TestCaseUtils;
|
import org.opends.server.api.DirectoryThread;
|
import org.opends.server.backends.MemoryBackend;
|
import org.opends.server.core.ModifyOperation;
|
import org.opends.server.replication.ReplicationTestCase;
|
import org.opends.server.replication.common.CSN;
|
import org.opends.server.replication.server.ReplServerFakeConfiguration;
|
import org.opends.server.replication.server.ReplicationServer;
|
import org.opends.server.types.DirectoryException;
|
import org.opends.server.types.Entry;
|
import org.opends.server.util.TestTimer;
|
import org.opends.server.util.TestTimer.CallableVoid;
|
import org.testng.annotations.AfterMethod;
|
import org.testng.annotations.BeforeMethod;
|
import org.testng.annotations.Test;
|
|
/**
|
* Tests the state checkpointer of a {@link LDAPReplicationDomain} against a backend which
|
* fails the write of the replication state.
|
*/
|
@SuppressWarnings("javadoc")
|
public class ServerStateFlushTest extends ReplicationTestCase
|
{
|
private static final String BACKEND_ID = "serverStateFlushTest";
|
private static final String BASE_DN_STRING = "o=serverStateFlushTest";
|
private static final int DS_ID = 1;
|
private static final int RS_ID = 601;
|
|
/** How long a test waits for something the checkpointer does on its own, in seconds. */
|
private static final int CHECKPOINT_TIMEOUT_IN_SECS = 20;
|
|
private DN baseDN;
|
private StateWriteFailureBackend backend;
|
private ReplicationServer replicationServer;
|
private LDAPReplicationDomain domain;
|
private Thread checkpointer;
|
private boolean domainDeleted;
|
|
/** What the backend does with a write of the replication state instead of writing it. */
|
private enum StateWrite
|
{
|
/** Write it, the way the backend normally would. */
|
SUCCEEDS,
|
/** Throw a {@link RuntimeException} the way an unwrapped driver failure would. */
|
THROWS_RUNTIME_EXCEPTION,
|
/** Throw an {@link Error}, which no {@code catch} of the checkpointer can be expected to hold. */
|
THROWS_ERROR,
|
/** Block until the test releases it, the way a write to an unresponsive storage would. */
|
BLOCKS;
|
}
|
|
/** A memory backend whose write of the replication state fails on demand. */
|
private static final class StateWriteFailureBackend extends MemoryBackend
|
{
|
private final DN stateEntryDN;
|
private volatile StateWrite stateWrite = StateWrite.SUCCEEDS;
|
private final AtomicInteger failedStateWrites = new AtomicInteger();
|
/**
|
* Those of {@link #failedStateWrites} the checkpointer made once it was asked to stop:
|
* the write its wake-up triggers, and the last one it runs on its way out.
|
*/
|
private final AtomicInteger failedStateWritesAfterShutdown = new AtomicInteger();
|
private final CountDownLatch blockedWriteStarted = new CountDownLatch(1);
|
private final CountDownLatch blockedWriteReleased = new CountDownLatch(1);
|
|
private StateWriteFailureBackend(DN stateEntryDN)
|
{
|
this.stateEntryDN = stateEntryDN;
|
}
|
|
/**
|
* Not synchronized, unlike the method it overrides: a write which blocks must not hold
|
* the monitor of the backend, or every other operation on it would block with it.
|
*/
|
@Override
|
public void replaceEntry(Entry oldEntry, Entry newEntry, ModifyOperation modifyOperation)
|
throws DirectoryException
|
{
|
if (stateEntryDN.equals(newEntry.getName()))
|
{
|
switch (stateWrite)
|
{
|
case THROWS_RUNTIME_EXCEPTION:
|
recordFailedStateWrite();
|
throw new IllegalStateException("injected failure of a replication state write");
|
case THROWS_ERROR:
|
recordFailedStateWrite();
|
throw new OutOfMemoryError("injected failure of a replication state write");
|
case BLOCKS:
|
blockedWriteStarted.countDown();
|
try
|
{
|
blockedWriteReleased.await();
|
}
|
catch (InterruptedException e)
|
{
|
Thread.currentThread().interrupt();
|
return;
|
}
|
break;
|
default:
|
break;
|
}
|
}
|
super.replaceEntry(oldEntry, newEntry, modifyOperation);
|
}
|
|
private void recordFailedStateWrite()
|
{
|
failedStateWrites.incrementAndGet();
|
final Thread writer = Thread.currentThread();
|
if (writer instanceof DirectoryThread && ((DirectoryThread) writer).isShutdownInitiated())
|
{
|
failedStateWritesAfterShutdown.incrementAndGet();
|
}
|
}
|
}
|
|
@BeforeMethod
|
public void setUpDomain() throws Exception
|
{
|
baseDN = DN.valueOf(BASE_DN_STRING);
|
|
backend = new StateWriteFailureBackend(baseDN);
|
backend.setBackendID(BACKEND_ID);
|
backend.setBaseDNs(baseDN);
|
backend.configureBackend(null, TestCaseUtils.getServerContext());
|
backend.openBackend();
|
TestCaseUtils.getServerContext().getBackendConfigManager().registerLocalBackend(backend);
|
backend.addEntry(createEntry(baseDN), null);
|
|
replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
|
TestCaseUtils.findFreePort(), "serverStateFlushTestDb", 0, RS_ID, 0, 100, null));
|
|
final SortedSet<String> replServers = newTreeSet("localhost:" + replicationServer.getReplicationPort());
|
domain = MultimasterReplication.createNewDomain(new DomainFakeCfg(baseDN, DS_ID, replServers));
|
domainDeleted = false;
|
domain.start();
|
|
checkpointer = checkpointerOf(domain);
|
assertNotNull(checkpointer, "the state checkpointer of the domain was not started");
|
assertTrue(checkpointer.isAlive(), "the state checkpointer of the domain is not running");
|
}
|
|
/**
|
* Takes back what {@link #setUpDomain()} put in place, one resource at a time: a setup
|
* which failed halfway through must not leave a backend or a domain behind for the next
|
* test of the class to trip over.
|
* <p>
|
* Bounded on its own: the {@code timeOut} of a test covers the test alone, and for a test
|
* which leaves the domain to this method, the shutdown under test runs here - a shutdown
|
* which hangs must fail the test rather than hold the whole run. The bound has room for
|
* the budget the shutdown gives the checkpointer and for the wait for the thread below.
|
*/
|
@AfterMethod(timeOut = 90000)
|
public void tearDownDomain() throws Exception
|
{
|
if (backend != null)
|
{
|
// Let go of a write the test left blocked, and of the checkpointer waiting on it,
|
// before the backend it is writing to is taken away.
|
backend.stateWrite = StateWrite.SUCCEEDS;
|
backend.blockedWriteReleased.countDown();
|
}
|
if (domain != null && !domainDeleted)
|
{
|
// Stops the checkpointer within a tick: waiting for it before this would spend the
|
// whole wait on a thread which is running by design.
|
deleteDomain();
|
}
|
domain = null;
|
if (checkpointer != null)
|
{
|
checkpointer.join(SECONDS.toMillis(CHECKPOINT_TIMEOUT_IN_SECS));
|
checkpointer = null;
|
}
|
if (replicationServer != null)
|
{
|
remove(replicationServer);
|
replicationServer = null;
|
}
|
if (backend != null)
|
{
|
backend.finalizeBackend();
|
TestCaseUtils.getServerContext().getBackendConfigManager().deregisterLocalBackend(backend);
|
backend = null;
|
}
|
}
|
|
/**
|
* A write of the state which throws must not stop the checkpointer: the state stays unsaved,
|
* and the next checkpoint writes it.
|
*/
|
@Test(timeOut = 120000)
|
public void checkpointerKeepsCheckpointingAfterAStateWriteThatThrows() throws Exception
|
{
|
backend.stateWrite = StateWrite.THROWS_RUNTIME_EXCEPTION;
|
final CSN csn = newCSN();
|
domain.getServerState().update(csn);
|
|
waitForFailedStateWrites(1);
|
|
backend.stateWrite = StateWrite.SUCCEEDS;
|
checkEntryHasAttributeValue(baseDN, "ds-sync-state", csn.toString(), CHECKPOINT_TIMEOUT_IN_SECS,
|
"the checkpointer did not write the state after a write which threw");
|
}
|
|
/**
|
* Shutting the domain down must not wait forever for a checkpointer whose writes all
|
* throw, and the checkpointer must hold the exception of the last write it runs on its
|
* way out as it does the others: that write used to end the thread before it reported
|
* that it was done.
|
*/
|
@Test(timeOut = 120000)
|
public void shutdownCompletesWhenEveryStateWriteThrows() throws Exception
|
{
|
backend.stateWrite = StateWrite.THROWS_RUNTIME_EXCEPTION;
|
domain.getServerState().update(newCSN());
|
|
waitForFailedStateWrites(1);
|
|
// A thread which lets an exception out is dead too, so isAlive() alone can not tell the
|
// last write being held from the thread dying of it. The handler of the thread outranks
|
// the one of its group, which would log the exception and raise an alert instead.
|
final AtomicReference<Throwable> uncaught = new AtomicReference<>();
|
checkpointer.setUncaughtExceptionHandler((t, e) -> uncaught.set(e));
|
|
deleteDomain();
|
|
assertFalse(checkpointer.isAlive(), "the state checkpointer is still running after the shutdown");
|
assertNull(uncaught.get(), "the last state write let an exception out of the checkpointer");
|
assertTrue(backend.failedStateWritesAfterShutdown.get() > 0,
|
"the checkpointer did not write the state on its way out");
|
}
|
|
/**
|
* Shutting the domain down must not wait forever for a checkpointer which is gone: an
|
* {@link Error} kills the thread whatever it catches.
|
*/
|
@Test(timeOut = 120000)
|
public void shutdownCompletesWhenTheCheckpointerDiedOfAnError() throws Exception
|
{
|
backend.stateWrite = StateWrite.THROWS_ERROR;
|
domain.getServerState().update(newCSN());
|
|
waitForCheckpointerToDie();
|
|
deleteDomain();
|
}
|
|
/**
|
* Shutting the domain down must not wait forever for a checkpointer whose write does not
|
* come back.
|
*/
|
@Test(timeOut = 120000)
|
public void shutdownCompletesWhileAStateWriteIsStuck() throws Exception
|
{
|
backend.stateWrite = StateWrite.BLOCKS;
|
domain.getServerState().update(newCSN());
|
|
assertTrue(backend.blockedWriteStarted.await(CHECKPOINT_TIMEOUT_IN_SECS, SECONDS),
|
"the checkpointer did not start the state write the test blocks on");
|
|
deleteDomain();
|
}
|
|
private CSN newCSN()
|
{
|
return new CSN(System.currentTimeMillis(), 1, DS_ID);
|
}
|
|
private void deleteDomain()
|
{
|
domainDeleted = true;
|
MultimasterReplication.deleteDomain(baseDN);
|
}
|
|
private void waitForFailedStateWrites(final int count) throws Exception
|
{
|
newTimer().repeatUntilSuccess(new CallableVoid()
|
{
|
@Override
|
public void call() throws Exception
|
{
|
assertTrue(backend.failedStateWrites.get() >= count,
|
"the state write of the checkpointer did not fail " + count + " time(s)");
|
}
|
});
|
}
|
|
private void waitForCheckpointerToDie() throws Exception
|
{
|
newTimer().repeatUntilSuccess(new CallableVoid()
|
{
|
@Override
|
public void call() throws Exception
|
{
|
assertFalse(checkpointer.isAlive(), "the state checkpointer did not die of the injected error");
|
}
|
});
|
}
|
|
private TestTimer newTimer()
|
{
|
return new TestTimer.Builder()
|
.maxSleep(CHECKPOINT_TIMEOUT_IN_SECS, SECONDS)
|
.sleepTimes(100, MILLISECONDS)
|
.toTimer();
|
}
|
|
private static Thread checkpointerOf(LDAPReplicationDomain domain)
|
{
|
final String name =
|
"Replica DS(" + domain.getServerId() + ") state checkpointer for domain \"" + domain.getBaseDN() + "\"";
|
final ThreadGroup group = DirectoryThread.DIRECTORY_THREAD_GROUP;
|
final Thread[] threads = new Thread[group.activeCount() * 2 + 10];
|
final int count = group.enumerate(threads, true);
|
for (int i = 0; i < count; i++)
|
{
|
if (name.equals(threads[i].getName()))
|
{
|
return threads[i];
|
}
|
}
|
return null;
|
}
|
}
|
