domainMap =
changelogDB.getDomainToReplicaDBs().get(TEST_ROOT_DN);
changelogDB.failNextReplicaDBCreation();
try
{
changelogDB.getOrCreateReplicaDB(TEST_ROOT_DN, FAILING_SERVER_ID, replicationServer);
failBecauseExceptionWasNotThrown(ChangelogException.class);
}
catch (ChangelogException expected)
{
assertThat(expected).hasMessage(CREATION_FAILURE.toString());
}
assertThat(changelogDB.getDomainToReplicaDBs().get(TEST_ROOT_DN))
.as("the domain map holding the replica DB created before the failure")
.isSameAs(domainMap)
.containsOnlyKeys(EXISTING_SERVER_ID);
}
finally
{
try
{
if (changelogDB != null)
{
changelogDB.shutdownDB();
}
}
finally
{
remove(replicationServer);
TestCaseUtils.deleteDirectory(testRoot);
}
}
}
/**
* The cleanup of a failed creation leaves the domain announced to every multi domain cursor
* which was live at the time, and the next successful creation of the domain announces it to
* them again: the second announcement must not open a second cursor over the same domain. Such
* a cursor would either leak unclosed - the cursor tree of {@code CompositeDBCursor} collapses
* cursors comparing equal - or deliver every change twice, which kills the
* {@code ChangeNumberIndexer} thread with the {@code IllegalStateException} its cookie update
* throws on a replayed change.
*
* This covers the drop path only: the {@code removeDomain()} path never re-announces a domain
* to a cursor which still holds it - the cursor drops the domain, through
* {@code indexer.clear()}, before the domain is unmapped.
*/
@Test
public void announcingADomainTwiceToALiveCursorMustNotOpenASecondDomainCursor() throws Exception
{
TestCaseUtils.startServer();
ReplicationServer replicationServer = null;
RaceableChangelogDB changelogDB = null;
File testRoot = null;
try
{
replicationServer = configureReplicationServer(100, 5000);
testRoot = createCleanDir("FileChangelogDB");
changelogDB = new RaceableChangelogDB(replicationServer, testRoot.getPath(), createCryptoSuite(false));
changelogDB.initializeDB();
// the live cursor both announcements reach
final MultiDomainDBCursor cursor = changelogDB.getCursorFrom(
new MultiDomainServerState(), new CursorOptions(GREATER_THAN_OR_EQUAL_TO_KEY, ON_MATCHING_KEY));
try
{
// first announcement: the failed creation announces the domain before dropping its map
changelogDB.failNextReplicaDBCreation();
try
{
changelogDB.getOrCreateReplicaDB(TEST_ROOT_DN, FAILING_SERVER_ID, replicationServer);
failBecauseExceptionWasNotThrown(ChangelogException.class);
}
catch (ChangelogException expected)
{
assertThat(expected).hasMessage(CREATION_FAILURE.toString());
}
cursor.next();
assertThat(changelogDB.walkedDomains)
.as("the domains the live cursor iterates over after the first announcement")
.containsExactly(TEST_ROOT_DN);
// second announcement: the next creation of the domain announces it to the cursor again
final FileReplicaDB replicaDB =
changelogDB.getOrCreateReplicaDB(TEST_ROOT_DN, FAILING_SERVER_ID, replicationServer).getFirst();
final CSN csn = new CSN(System.currentTimeMillis(), 1, FAILING_SERVER_ID);
replicaDB.add(new DeleteMsg(TEST_ROOT_DN, csn, "uid"));
waitChangesArePersisted(replicaDB, 1);
assertThat(cursor.next()).as("the change published after the second announcement").isTrue();
assertThat(cursor.getRecord().getCSN()).isEqualTo(csn);
assertThat(changelogDB.walkedDomains)
.as("announcing an already incorporated domain again must not open a second cursor over it")
.containsExactly(TEST_ROOT_DN);
assertThat(cursor.next()).as("the single published change is delivered more than once").isFalse();
}
finally
{
cursor.close();
}
}
finally
{
try
{
if (changelogDB != null)
{
changelogDB.shutdownDB();
}
}
finally
{
remove(replicationServer);
TestCaseUtils.deleteDirectory(testRoot);
}
}
}
/**
* A creation which bails out on the identity check must not drop the domain map another creation
* has freshly inserted: the drop is equality based and two empty maps are equal, so an identity
* unaware cleanup would unmap the fresh map, and the replica DB about to be published into it
* would no longer be reachable from {@code domainToReplicaDBs} - nothing would ever shut it
* down, which is the leak of #813 all over again.
*
* The interleaving is driven step by step:
*
* - the stale creator obtains its domain map and is held before entering its monitor;
* - {@code removeDomain()} unmaps that domain map;
* - a fresh creator inserts a new, still empty domain map and is held inside
* {@code newReplicaDB()}, under the fresh map's monitor;
* - the stale creator is released: its identity check fails and it must bail out without
* touching the fresh map, then retry and block on the fresh map's monitor;
* - the fresh creator is released: both creations complete into that same map.
*
*/
@Test
public void bailOutMustNotUnmapAnotherThreadsFreshDomainMap() throws Exception
{
TestCaseUtils.startServer();
ReplicationServer replicationServer = null;
RaceableChangelogDB changelogDB = null;
File testRoot = null;
Thread staleCreator = null;
Thread freshCreator = null;
final AtomicReference staleCreationFailure = new AtomicReference<>();
final AtomicReference freshCreationFailure = new AtomicReference<>();
try
{
replicationServer = configureReplicationServer(100, 5000);
testRoot = createCleanDir("FileChangelogDB");
changelogDB = new RaceableChangelogDB(replicationServer, testRoot.getPath(), createCryptoSuite(false));
changelogDB.initializeDB();
final RaceableChangelogDB racedChangelogDB = changelogDB;
final ReplicationServer racedReplicationServer = replicationServer;
// 1- the stale creator obtains the domain map about to be unmapped, and is parked there
changelogDB.holdNextCreationAfterItsDomainMapIsObtained();
staleCreator = new Thread("FileChangelogDBTest stale replica DB creator")
{
@Override
public void run()
{
try
{
racedChangelogDB.getOrCreateReplicaDB(TEST_ROOT_DN, STALE_SERVER_ID, racedReplicationServer);
}
catch (Throwable t)
{
staleCreationFailure.set(t);
}
}
};
staleCreator.start();
changelogDB.awaitCreatorHoldingItsDomainMap();
// 2- the domain map the stale creator holds is unmapped
changelogDB.removeDomain(TEST_ROOT_DN);
// 3- the fresh creator inserts a new, still empty domain map, and is parked inside
// newReplicaDB(), under the monitor of that fresh map
changelogDB.holdNextReplicaDBOnceCreated();
freshCreator = new Thread("FileChangelogDBTest fresh replica DB creator")
{
@Override
public void run()
{
try
{
racedChangelogDB.getOrCreateReplicaDB(TEST_ROOT_DN, FRESH_SERVER_ID, racedReplicationServer);
}
catch (Throwable t)
{
freshCreationFailure.set(t);
}
}
};
freshCreator.start();
changelogDB.awaitCreatorHoldingItsCreatedReplicaDB();
final ConcurrentMap freshDomainMap =
changelogDB.getDomainToReplicaDBs().get(TEST_ROOT_DN);
assertThat(freshDomainMap).as("the fresh domain map, not published into yet").isNotNull().isEmpty();
// 4- the stale creator bails out on its identity check, retries, and blocks on the monitor
// of the fresh domain map - without the identity check its cleanup would have unmapped the
// fresh map, and it would have completed into a third map instead of blocking
changelogDB.releaseCreatorHoldingItsDomainMap();
waitUntilBlockedOnOrCompleted(staleCreator, freshDomainMap);
assertThat(changelogDB.getDomainToReplicaDBs().get(TEST_ROOT_DN))
.as("the domain map the fresh creator is about to publish its replica DB into")
.isSameAs(freshDomainMap);
// 5- both creations complete into that same map
changelogDB.releaseCreatedReplicaDB();
staleCreator.join(TIMEOUT_MS);
freshCreator.join(TIMEOUT_MS);
assertThat(staleCreator.isAlive()).as("the stale creator thread did not complete").isFalse();
assertThat(freshCreator.isAlive()).as("the fresh creator thread did not complete").isFalse();
assertThat(staleCreationFailure.get()).isNull();
assertThat(freshCreationFailure.get()).isNull();
assertThat(changelogDB.getDomainToReplicaDBs().get(TEST_ROOT_DN))
.isSameAs(freshDomainMap)
.containsOnlyKeys(STALE_SERVER_ID, FRESH_SERVER_ID);
}
finally
{
try
{
if (changelogDB != null)
{
changelogDB.releaseAllHeldThreads();
changelogDB.shutdownDB();
}
}
finally
{
join(staleCreator);
join(freshCreator);
deregisterLeakedReplicaDBMonitors(replicationServer, STALE_SERVER_ID, FRESH_SERVER_ID);
remove(replicationServer);
TestCaseUtils.deleteDirectory(testRoot);
}
}
}
/**
* The concurrent remover unmapped the domain and shut its replica DBs down, exactly like
* the {@code shutdownDB()} drain does: {@code removeDomain()} must complete without
* throwing a {@link NullPointerException}.
*/
@Test
public void removeDomainRacingConcurrentRemovalMustNotThrowNPE() throws Exception
{
ReplicationServer replicationServer = null;
try
{
TestCaseUtils.startServer();
replicationServer = configureReplicationServer(100, 100);
final FileChangelogDB changelogDB = (FileChangelogDB) replicationServer.getChangelogDB();
final FileReplicaDB replicaDB =
changelogDB.getOrCreateReplicaDB(TEST_ROOT_DN, SERVER_ID, replicationServer).getFirst();
final ConcurrentMap> domainToReplicaDBs =
changelogDB.getDomainToReplicaDBs();
final ConcurrentMap domainMap = domainToReplicaDBs.get(TEST_ROOT_DN);
assertThat(domainMap).isNotNull();
final AtomicReference thrown = new AtomicReference<>();
final Thread remover = newRemoverThread(changelogDB, thrown);
synchronized (domainMap)
{
remover.start();
// removeDomain() read the domain entry and is now blocked on the monitor held here
waitUntilBlockedOn(remover, domainMap);
domainToReplicaDBs.remove(TEST_ROOT_DN);
replicaDB.shutdown();
}
remover.join(TIMEOUT_MS);
assertFalse(remover.isAlive(), "removeDomain() did not complete");
assertThat(thrown.get()).isNull();
}
finally
{
remove(replicationServer);
}
}
/**
* The concurrent remover unmapped the domain and {@code getOrCreateReplicaDB()} then
* recreated it: {@code removeDomain()} must only unmap the domainMap instance it holds the
* monitor on, never the recreated one.
*/
@Test
public void removeDomainMustNotUnmapConcurrentlyRecreatedDomain() throws Exception
{
ReplicationServer replicationServer = null;
try
{
TestCaseUtils.startServer();
replicationServer = configureReplicationServer(100, 100);
final FileChangelogDB changelogDB = (FileChangelogDB) replicationServer.getChangelogDB();
final FileReplicaDB replicaDB =
changelogDB.getOrCreateReplicaDB(TEST_ROOT_DN, SERVER_ID, replicationServer).getFirst();
final ConcurrentMap> domainToReplicaDBs =
changelogDB.getDomainToReplicaDBs();
final ConcurrentMap domainMap = domainToReplicaDBs.get(TEST_ROOT_DN);
assertThat(domainMap).isNotNull();
final ConcurrentMap recreatedDomainMap = new ConcurrentHashMap<>();
final AtomicReference thrown = new AtomicReference<>();
final Thread remover = newRemoverThread(changelogDB, thrown);
synchronized (domainMap)
{
remover.start();
waitUntilBlockedOn(remover, domainMap);
domainToReplicaDBs.remove(TEST_ROOT_DN);
replicaDB.shutdown();
domainToReplicaDBs.put(TEST_ROOT_DN, recreatedDomainMap);
}
remover.join(TIMEOUT_MS);
assertFalse(remover.isAlive(), "removeDomain() did not complete");
assertThat(thrown.get()).isNull();
assertThat(domainToReplicaDBs.get(TEST_ROOT_DN)).isSameAs(recreatedDomainMap);
}
finally
{
remove(replicationServer);
}
}
private Thread newRemoverThread(final FileChangelogDB changelogDB, final AtomicReference thrown)
{
return new Thread(new Runnable()
{
@Override
public void run()
{
try
{
changelogDB.removeDomain(TEST_ROOT_DN);
}
catch (Throwable t)
{
thrown.set(t);
}
}
}, "removeDomain() under test");
}
/** Waits until the provided replica DB has persisted the provided number of records. */
private void waitChangesArePersisted(FileReplicaDB replicaDB, int recordCount) throws Exception
{
final long deadline = System.currentTimeMillis() + TIMEOUT_MS;
while (replicaDB.getNumberRecords() < recordCount)
{
if (System.currentTimeMillis() > deadline)
{
throw new AssertionError("Timed out waiting for " + recordCount + " records to be persisted");
}
Thread.sleep(10);
}
}
/** Waits until the provided thread is blocked acquiring the monitor of the provided object. */
private void waitUntilBlockedOn(Thread thread, Object monitor) throws Exception
{
final long deadline = System.currentTimeMillis() + TIMEOUT_MS;
while (System.currentTimeMillis() < deadline)
{
if (isBlockedOn(thread, monitor))
{
return;
}
Thread.sleep(1);
}
throw new AssertionError(
"Timed out waiting for " + thread.getName() + " to block on the domainMap monitor");
}
/**
* Waits until the provided thread is blocked acquiring the monitor of the provided object, or
* has completed: completion is left for the caller's assertions to diagnose.
*/
private void waitUntilBlockedOnOrCompleted(Thread thread, Object monitor) throws Exception
{
final long deadline = System.currentTimeMillis() + TIMEOUT_MS;
while (System.currentTimeMillis() < deadline)
{
if (!thread.isAlive() || isBlockedOn(thread, monitor))
{
return;
}
Thread.sleep(1);
}
throw new AssertionError("Timed out waiting for " + thread.getName()
+ " to block on the domainMap monitor or complete");
}
private static boolean isBlockedOn(Thread thread, Object monitor)
{
final ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean();
final ThreadInfo threadInfo = threadMXBean.getThreadInfo(thread.getId());
final LockInfo lockInfo = threadInfo != null ? threadInfo.getLockInfo() : null;
return lockInfo != null
&& threadInfo.getThreadState() == Thread.State.BLOCKED
&& lockInfo.getIdentityHashCode() == System.identityHashCode(monitor);
}
/** Joins the provided thread, leaving a signal behind when it did not die within the timeout. */
private void join(final Thread thread) throws InterruptedException
{
if (thread != null)
{
thread.join(TIMEOUT_MS);
if (thread.isAlive())
{
final IllegalStateException hung = new IllegalStateException("Test thread " + thread.getName()
+ " is still alive after " + TIMEOUT_MS + " ms: it may leak a live changelog into later tests");
hung.setStackTrace(thread.getStackTrace());
hung.printStackTrace();
thread.interrupt();
}
}
}
/**
* Returns the name the monitor provider of the provided replica DB is registered under, i.e. the
* name built by {@code FileReplicaDB.DbMonitorProvider.getMonitorInstanceName()}, lower-cased
* the way {@code DirectoryServer.registerMonitorProvider()} stores it.
*/
private String replicaDBMonitorName(final ReplicationServer replicationServer, final int serverId)
{
final ReplicationServerDomain domain = replicationServer.getReplicationServerDomain(TEST_ROOT_DN);
assertThat(domain).as("the domain scoping the monitor name of DS(" + serverId + ")").isNotNull();
return toLowerCase("Changelog for DS(" + serverId + "),cn=" + domain.getMonitorInstanceName());
}
/** Releases the monitor providers a regression leaks, so that they do not outlive this test. */
private void deregisterLeakedReplicaDBMonitors(final ReplicationServer replicationServer, final int... serverIds)
{
if (replicationServer == null || replicationServer.getReplicationServerDomain(TEST_ROOT_DN) == null)
{
return; // no replica DB was ever created, hence no monitor provider was ever registered
}
for (final int serverId : serverIds)
{
// deregister the provider instead of removing the map entry, so that the JMX MBean
// registered alongside it is released as well
final MonitorProvider provider =
DirectoryServer.getMonitorProviders().get(replicaDBMonitorName(replicationServer, serverId));
if (provider != null)
{
DirectoryServer.deregisterMonitorProvider(provider);
}
}
}
/**
* A changelog DB which lets a test hold a thread creating a replica DB right after it has read
* the shutdown flag, hold it after it has obtained its domain map but before it enters the
* monitor, hold it again once the replica DB is created but not yet published into the domain
* map, hold the shutdown inside the drain of {@code domainToReplicaDBs}, make the next replica
* DB creation fail, and record the domains cursors are opened for.
*/
private static final class RaceableChangelogDB extends FileChangelogDB
{
private final AtomicBoolean holdNextCreation = new AtomicBoolean();
private final AtomicBoolean holdNextDomainMapObtained = new AtomicBoolean();
private final AtomicBoolean holdNextReplicaDBShutdown = new AtomicBoolean();
private final AtomicBoolean holdNextCreatedReplicaDB = new AtomicBoolean();
private final AtomicBoolean failNextCreation = new AtomicBoolean();
private final CountDownLatch creatorIsInWindow = new CountDownLatch(1);
private final CountDownLatch creatorIsReleased = new CountDownLatch(1);
private final CountDownLatch creatorHoldsItsDomainMap = new CountDownLatch(1);
private final CountDownLatch domainMapIsReleased = new CountDownLatch(1);
private final CountDownLatch creatorHoldsItsCreatedReplicaDB = new CountDownLatch(1);
private final CountDownLatch createdReplicaDBIsReleased = new CountDownLatch(1);
private final CountDownLatch drainIsInReplicaDBShutdown = new CountDownLatch(1);
private final CountDownLatch drainIsReleased = new CountDownLatch(1);
/** The baseDNs of the domains any cursor was opened for, one element per opening. */
private final List walkedDomains = new CopyOnWriteArrayList<>();
RaceableChangelogDB(final ReplicationServer replicationServer, final String dbDirectoryPath,
final CryptoSuite cryptoSuite) throws ConfigException
{
super(replicationServer, dbDirectoryPath, cryptoSuite);
}
@Override
ConcurrentMap getExistingOrNewDomainMap(final DN baseDN)
{
if (holdNextCreation.compareAndSet(true, false))
{
creatorIsInWindow.countDown();
await(creatorIsReleased);
}
final ConcurrentMap domainMap = super.getExistingOrNewDomainMap(baseDN);
if (holdNextDomainMapObtained.compareAndSet(true, false))
{
creatorHoldsItsDomainMap.countDown();
await(domainMapIsReleased);
}
return domainMap;
}
@Override
FileReplicaDB newReplicaDB(final int serverId, final DN baseDN, final ReplicationServer server,
final CryptoSuite cryptoSuite, final ReplicationEnvironment replicationEnv) throws ChangelogException
{
if (failNextCreation.compareAndSet(true, false))
{
throw new ChangelogException(CREATION_FAILURE);
}
if (holdNextReplicaDBShutdown.compareAndSet(true, false))
{
return new HeldOnShutdownReplicaDB(serverId, baseDN, server, cryptoSuite, replicationEnv);
}
final FileReplicaDB replicaDB = super.newReplicaDB(serverId, baseDN, server, cryptoSuite, replicationEnv);
if (holdNextCreatedReplicaDB.compareAndSet(true, false))
{
// the replica DB exists and its monitor provider is registered, but it is not published
// into the domain map yet: hold the creator there, under the domain map monitor
creatorHoldsItsCreatedReplicaDB.countDown();
await(createdReplicaDBIsReleased);
}
return replicaDB;
}
@Override
public DBCursor getCursorFrom(final DN baseDN, final ServerState startState,
final CursorOptions options) throws ChangelogException
{
walkedDomains.add(baseDN);
return super.getCursorFrom(baseDN, startState, options);
}
void holdNextReplicaDBCreationBeforeItsDomainMapIsInserted()
{
holdNextCreation.set(true);
}
void holdNextCreationAfterItsDomainMapIsObtained()
{
holdNextDomainMapObtained.set(true);
}
void holdNextReplicaDBInItsShutdown()
{
holdNextReplicaDBShutdown.set(true);
}
void holdNextReplicaDBOnceCreated()
{
holdNextCreatedReplicaDB.set(true);
}
void failNextReplicaDBCreation()
{
failNextCreation.set(true);
}
void awaitCreatorInWindow()
{
await(creatorIsInWindow);
}
void awaitCreatorHoldingItsDomainMap()
{
await(creatorHoldsItsDomainMap);
}
void awaitCreatorHoldingItsCreatedReplicaDB()
{
await(creatorHoldsItsCreatedReplicaDB);
}
void awaitDrainInReplicaDBShutdown()
{
await(drainIsInReplicaDBShutdown);
}
void releaseCreator()
{
creatorIsReleased.countDown();
}
void releaseCreatorHoldingItsDomainMap()
{
domainMapIsReleased.countDown();
}
void releaseCreatedReplicaDB()
{
createdReplicaDBIsReleased.countDown();
}
void releaseDrain()
{
drainIsReleased.countDown();
}
void releaseAllHeldThreads()
{
releaseCreator();
releaseCreatorHoldingItsDomainMap();
releaseCreatedReplicaDB();
releaseDrain();
}
/** A replica DB which holds the thread shutting it down until the test releases it. */
private final class HeldOnShutdownReplicaDB extends FileReplicaDB
{
HeldOnShutdownReplicaDB(final int serverId, final DN baseDN, final ReplicationServer server,
final CryptoSuite cryptoSuite, final ReplicationEnvironment replicationEnv) throws ChangelogException
{
super(serverId, baseDN, server, cryptoSuite, replicationEnv);
}
@Override
void shutdown()
{
drainIsInReplicaDBShutdown.countDown();
await(drainIsReleased);
super.shutdown();
}
}
private static void await(final CountDownLatch latch)
{
try
{
if (!latch.await(TIMEOUT_MS, TimeUnit.MILLISECONDS))
{
throw new IllegalStateException("timed out waiting for the replica DB creation race");
}
}
catch (InterruptedException e)
{
Thread.currentThread().interrupt();
throw new IllegalStateException(e);
}
}
}
}