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 409083ed20..b4e982c6f0 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 @@ -2783,8 +2783,9 @@ void replay(LDAPUpdateMsg msg, AtomicBoolean replayThreadShutdown) * bare, without the failure count that road did not reach, and restart the session * so that it is delivered again: this is the last resort, the throwable is rethrown * whatever happens here, and the thread this runs on may well be ending on it. A - * restart which can not run here leaves its request standing, and the state - * checkpointer of this domain runs it. + * restart which can not run here - it threw, or a total update is being processed + * over the session - leaves its request standing, and the state checkpointer of + * this domain runs it. */ if (owned != null) { @@ -3822,16 +3823,55 @@ private boolean recoverFromReplayFailure( */ sessionRestarts.request(replayThreadShutdown.get() || outOfMemory ? SessionRestart.NOW : SessionRestart.AFTER_BACKOFF); - runRequestedSessionRestarts(); + if (!runRequestedSessionRestarts() && !outOfMemory) + { + /* + * The line above said the session is being restarted for the change, and it is not + * yet: a total update is being processed over that session - an export from this + * replica, since a total update into it owns the session and is refused above - and + * the restart waits for it, for as long as the export takes. Said on its own, so that + * a change which is not delivered again for minutes is not a change nobody asked for; + * not on the road out of a JVM which has run out of memory, for the reason the line + * above is not built there. + */ + logger.info(NOTE_REPLAY_SESSION_RESTART_HELD_BY_TOTAL_UPDATE, csn, getBaseDN()); + } return true; } /** * Restarts the session as long as changes which could not be replayed are waiting to be - * delivered again. + * delivered again - unless a total update is being processed, in which case the requests + * are left standing for the state checkpointer to run once it is over. + *

+ * A restart stops the session the total update runs over, in either direction. An import + * into this replica reads its entries from that session and would end on the ones which + * had arrived - and {@code disabled} does not say an import is running, since + * {@code preBackendImport()} keeps the backend events this domain is the cause of from + * disabling it. An export from this replica publishes its entries over it, and + * {@code exportLDIFEntry()} gives the export up as + * {@code ERR_INIT_RS_DISCONNECTION_DURING_EXPORT} once the broker has been stopped under + * it, which leaves the replica it was initializing to be initialized again: minutes on a + * large backend, spent for a change which would have waited. So the change waits: the + * request stays standing, the state checkpointer comes for it once a second and runs it + * as soon as the total update is over ({@link #runPendingSessionRestart()}), and the + * replication server delivers the change again then. The ServerState waits with it, and + * the replay of this domain keeps running in the meantime. + *

+ * A total update which begins between this read and the stop of the session is not seen + * here, and is cut by it: the read and the claim of the import/export context share no + * lock, which is issue #1041 on the import side - a few statements wide, where the whole + * of the total update was. + * + * @return {@code false} when a total update is being processed and the requests were left + * standing for the state checkpointer, {@code true} otherwise */ - private void runRequestedSessionRestarts() + private boolean runRequestedSessionRestarts() { + if (ieRunning()) + { + return false; + } /* * The outer loop is what makes a request which was made while this thread was giving * up the recovery its own: the thread which made it found the recovery taken and left @@ -3871,6 +3911,7 @@ private void runRequestedSessionRestarts() replayFailureRecovery.set(false); } } + return true; } /** @@ -3886,22 +3927,15 @@ private void runRequestedSessionRestarts() * topology, with the changes it did not replay owned by the replication server and its * ServerState stopped behind them. *

- * Not run while a total update is being processed, in either direction: a restart stops - * the session the total update runs over. An import into this replica reads its entries - * from that session and would end on the ones which had arrived - and {@code disabled} - * does not say an import is running, since {@code preBackendImport()} keeps the backend - * events this domain is the cause of from disabling it. An export from this replica - * publishes its entries over it, and {@code exportLDIFEntry()} gives the export up as - * {@code ERR_INIT_RS_DISCONNECTION_DURING_EXPORT} once the broker has been stopped - * under it, which leaves the replica it was initializing to be initialized again. This - * thread is the one which can afford to wait: the request stays standing, and it comes - * back here once a second, so the restart is run as soon as the total update is over. - * The change the restart was asked for waits for as long as the total update takes, and - * the ServerState with it; the replay of this domain keeps running in the meantime. + * While a total update is being processed, in either direction, the restart is not run - + * no restart asked for by a released change is, see {@link #runRequestedSessionRestarts()} + * - and this thread is the one which can afford to wait for it: the request stays + * standing, and it comes back here once a second, so the restart is run as soon as the + * total update is over. */ private void runPendingSessionRestart() { - if (shutdown.get() || disabled || ieRunning() || !sessionRestarts.isPending()) + if (shutdown.get() || disabled || !sessionRestarts.isPending()) { return; } diff --git a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties index c6b8fc1718..2449cf0d59 100644 --- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties +++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties @@ -700,3 +700,7 @@ ERR_CHECKPOINTING_STATE_FAILED_323=Could not write the replication state of doma comes back with the last state it did write and replays the changes since ERR_STATE_CHECKPOINTER_NOT_STOPPED_324=The state checkpointer of domain "%s" has not stopped within \ %d ms : the shutdown of the domain goes on without it +NOTE_REPLAY_SESSION_RESTART_HELD_BY_TOTAL_UPDATE_331=The session restart asked for by change %s \ + in domain "%s" is held: a total update is being processed over that session, and stopping it \ + would end the total update. The restart runs once the total update is over, and the change is \ + sent again then; until then it is not recorded as replayed diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringExportTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringExportTest.java new file mode 100644 index 0000000000..db345a3937 --- /dev/null +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringExportTest.java @@ -0,0 +1,442 @@ +/* + * 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.nio.charset.StandardCharsets.*; +import static org.assertj.core.api.Assertions.*; +import static org.opends.messages.ReplicationMessages.*; +import static org.opends.server.TestCaseUtils.*; +import static org.opends.server.core.DirectoryServer.*; +import static org.testng.Assert.*; + +import java.net.SocketTimeoutException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.SortedSet; +import java.util.TreeSet; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.forgerock.opendj.ldap.DN; +import org.forgerock.opendj.ldap.ResultCode; +import org.forgerock.opendj.server.config.meta.ReplicationDomainCfgDefn.IsolationPolicy; +import org.opends.server.TestCaseUtils; +import org.opends.server.core.DirectoryServer; +import org.opends.server.plugins.ShortCircuitPlugin; +import org.opends.server.replication.ReplicationTestCase; +import org.opends.server.replication.common.CSN; +import org.opends.server.replication.common.CSNGenerator; +import org.opends.server.replication.common.ServerStatus; +import org.opends.server.replication.protocol.DoneMsg; +import org.opends.server.replication.protocol.EntryMsg; +import org.opends.server.replication.protocol.ErrorMsg; +import org.opends.server.replication.protocol.InitializeRcvAckMsg; +import org.opends.server.replication.protocol.InitializeRequestMsg; +import org.opends.server.replication.protocol.InitializeTargetMsg; +import org.opends.server.replication.protocol.LDAPUpdateMsg; +import org.opends.server.replication.protocol.ModifyMsg; +import org.opends.server.replication.protocol.ReplicationMsg; +import org.opends.server.replication.server.ReplServerFakeConfiguration; +import org.opends.server.replication.server.ReplicationServer; +import org.opends.server.replication.service.ReplicationBroker; +import org.opends.server.types.Entry; +import org.opends.server.types.OperationType; +import org.testng.Assert; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +/** + * Tests the replay of a change while this replica is the source of a total update. + *

+ * The export of a total update publishes its entries over the session of the domain, from a + * thread of the export pool, while the replay of the domain keeps running. A change which can + * not be replayed meanwhile is given back for the replication server to send again, and that + * takes a session restart. Run by the thread which released the change, the restart stops the + * broker the export publishes over, and {@code exportLDIFEntry()} gives the export up on it: the + * replica being initialized is left to be initialized again, for a change which would have + * waited (issue #1048). The restart has to wait for the export instead, and the state + * checkpointer runs it once the export is over. + *

+ * The importer is a broker of this test, so that the test says when the export moves: the + * exporter publishes no more than the initialization window ahead of the importer's + * acknowledgements, and the change is replayed while the export waits for one. + */ +@SuppressWarnings("javadoc") +public class ReplayDuringExportTest extends ReplicationTestCase +{ + /** + * A total update needs a backend which keeps its data across the export, and one which + * the exporter can lock: the {@code userRoot} backend, as for the import direction. + */ + private static final String EXAMPLE_DN = "dc=example,dc=com"; + private static final int RS_ID = 612; + private static final int DS_ID = 1; + private static final int IMPORTER_ID = 2; + /** How many entry messages the exporter publishes ahead of the importer's acknowledgements. */ + private static final int INIT_WINDOW = 2; + /** + * An entry message carries a buffer of the export stream rather than one entry, so the data + * has to outgrow the window by that much before the exporter waits for an acknowledgement. + */ + private static final int ENTRY_MSG_BYTES = 8192; + private static final int BULK_ENTRY_BYTES = 4096; + private static final int BULK_ENTRIES = 2 * (INIT_WINDOW + 2); + private static final AtomicBoolean SHUTDOWN = new AtomicBoolean(false); + + private DN baseDN; + private ReplicationServer replicationServer; + private LDAPReplicationDomain domain; + private TestSynchronousReplayQueue queue; + private ReplicationBroker importer; + private CSNGenerator gen; + + @BeforeMethod + public void setUpLocal() throws Exception + { + baseDN = DN.valueOf(EXAMPLE_DN); + TestCaseUtils.clearBackend("userRoot", EXAMPLE_DN); + + final int rsPort = TestCaseUtils.findFreePort(); + replicationServer = new ReplicationServer(new ReplServerFakeConfiguration( + rsPort, "replayDuringExportTestDb", 0, RS_ID, 0, 100, new TreeSet())); + + final SortedSet replServers = new TreeSet<>(); + replServers.add("localhost:" + rsPort); + final DomainFakeCfg conf = new DomainFakeCfg(baseDN, DS_ID, replServers); + conf.setIsolationPolicy(IsolationPolicy.ACCEPT_ALL_UPDATES); + queue = new TestSynchronousReplayQueue(); + domain = MultimasterReplication.createNewDomain(conf, queue); + domain.start(); + assertTrue(domain.isConnected(), "the domain did not connect to the replication server"); + + // A short socket timeout: the test bounds its own waits, and receive() returns to it on it. + importer = openReplicationSession(baseDN, IMPORTER_ID, 100, rsPort, 2000); + gen = new CSNGenerator(IMPORTER_ID, 0); + } + + @AfterMethod + public void tearDown() throws Exception + { + try + { + stop(importer); + MultimasterReplication.deleteDomain(baseDN); + } + finally + { + remove(replicationServer); + } + } + + /** + * A change which can not be replayed while the export streams must leave the session to + * the export, and be delivered again once the export is over. + *

+ * The attempts in place are spent - the backend is live, an export takes nothing away - and + * the change is given back and asked for again, as it is when nothing else is going on: what + * waits is the session restart that takes. The restart stands as a request for as long as + * the export runs, and the state checkpointer, which holds its own restarts back for the + * same reason, runs it when the export is over. Without the hold the replay thread stops the + * broker the exporter publishes over: the export ends on the entries which had been + * published, with {@code ERR_INIT_RS_DISCONNECTION_DURING_EXPORT}, the rest never reaches + * the importer, and the importer has to be initialized again. + */ + @Test(timeOut = 120_000) + public void aReplayWhichFailsDuringTheExportLeavesTheSessionToTheExport() throws Exception + { + final Entry entry = TestCaseUtils.addEntry( + "dn: cn=renamedSince," + EXAMPLE_DN, + "objectClass: top", + "objectClass: person", + "cn: renamedSince", + "sn: renamedSince"); + final String entryUUID = getEntryUUID(entry.getName()); + addEntriesWorthMoreThanTheWindow(); + final long exportedEntries = countEntriesOfTheDomain(); + + /* + * The change goes through the replication server, which is what has it to deliver again + * once the session has been restarted for it; the replay queue of the domain is the + * test's, so the change is replayed when the test says, which is during the export. + */ + final CSN csn = gen.newCSN(); + importer.publish(new ModifyMsg(csn, DN.valueOf("cn=movedAway," + EXAMPLE_DN), + generatemods("description", "replayed during the export"), entryUUID)); + final LDAPUpdateMsg delivered = awaitDelivery(csn, 30_000, "the change was not delivered"); + + startExport(); + final List held = receiveEntryMsgsWithoutAcknowledging(INIT_WINDOW); + assertTrue(domain.ieRunning(), "the export is not being processed"); + + // Replayed while the exporter waits for an acknowledgement: every attempt in place ends on + // an entryUUID search which does not run, and the change is given back. + ShortCircuitPlugin.registerShortCircuit( + OperationType.SEARCH, "PreParse", ResultCode.UNAVAILABLE.intValue()); + try + { + replay(delivered); + assertTrue(ShortCircuitPlugin.getShortCircuitCount(OperationType.SEARCH, "PreParse") + >= LDAPReplicationDomain.IN_PLACE_REPLAY_ATTEMPTS, + "every attempt in place must have made its search: the backend is live while the" + + " export runs, so nothing holds the replay off"); + } + finally + { + ShortCircuitPlugin.deregisterShortCircuit(OperationType.SEARCH, "PreParse"); + } + + assertThat(errorLogRecordsOf(WARN_REPLAY_RETRYING_CHANGE.ordinal(), csn)) + .as("the change was not asked for again: an export is not a total update into this" + + " replica, whose state would cover the change once it is loaded") + .isNotEmpty(); + + /* + * The export is held across a tick of the state checkpointer, which comes for every + * restart left standing once a second: the request is standing now, and whichever thread + * comes for it while the export runs has to leave it standing. + */ + Thread.sleep(1500); + + finishExport(held, exportedEntries); + + assertThat(errorLogRecordsOf(NOTE_REPLAY_SESSION_RESTART_HELD_BY_TOTAL_UPDATE.ordinal(), csn)) + .as("the restart the change was asked for again with was not reported as held") + .isNotEmpty(); + + // The change is delivered again once the export is over, and applied. + final LDAPUpdateMsg again = awaitDelivery(csn, 30_000, "the change was not delivered again" + + " once the export was over: the session restart it was asked for again with was" + + " not run"); + replay(again); + assertThat(DirectoryServer.getEntry(entry.getName()).getAllAttributes("description")) + .as("the change delivered again after the export was not applied").isNotEmpty(); + assertTrue(domain.getServerState().cover(csn), + "the change delivered again after the export was applied and not recorded"); + } + + /** Adds entries whose export outgrows the initialization window, so that the exporter waits. */ + private void addEntriesWorthMoreThanTheWindow() throws Exception + { + assertThat(BULK_ENTRIES * BULK_ENTRY_BYTES) + .as("the data must outgrow the window for the exporter to wait for an acknowledgement") + .isGreaterThan((INIT_WINDOW + 1) * ENTRY_MSG_BYTES); + final char[] padding = new char[BULK_ENTRY_BYTES]; + Arrays.fill(padding, 'x'); + for (int i = 0; i < BULK_ENTRIES; i++) + { + TestCaseUtils.addEntry( + "dn: cn=bulk" + i + "," + EXAMPLE_DN, + "objectClass: top", + "objectClass: person", + "cn: bulk" + i, + "sn: bulk" + i, + "description: " + new String(padding)); + } + } + + private long countEntriesOfTheDomain() throws Exception + { + return getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN) + .getNumberOfEntriesInBaseDN(baseDN); + } + + /** + * Has the importer ask this replica for a total update, and returns once the export has + * begun: the {@code InitializeTargetMsg} which starts it has arrived. + */ + private void startExport() throws Exception + { + // The export is refused while this replica does not see the importer in its topology. + final long deadline = System.currentTimeMillis() + 30_000; + while (!domain.getReplicaInfos().containsKey(IMPORTER_ID)) + { + assertTrue(System.currentTimeMillis() < deadline, + "the domain did not see the importer in its topology"); + Thread.sleep(20); + } + importer.publish(new InitializeRequestMsg(baseDN, IMPORTER_ID, DS_ID, INIT_WINDOW)); + // The exporter waits for the importer to be in the full update status before it streams. + importer.signalStatusChange(ServerStatus.FULL_UPDATE_STATUS); + final ReplicationMsg msg = receiveTotalUpdateMsg(30_000); + assertThat(msg).as("the total update did not begin").isInstanceOf(InitializeTargetMsg.class); + } + + /** + * Receives entry messages up to the window and acknowledges none of them: the exporter + * publishes no more than the window ahead of the last acknowledgement, so its next entry + * message waits for one from now on. + */ + private List receiveEntryMsgsWithoutAcknowledging(int window) throws Exception + { + final List received = new ArrayList<>(); + while (received.size() < window) + { + final ReplicationMsg msg = receiveTotalUpdateMsg(30_000); + assertThat(msg).as("the export did not stream up to the window").isInstanceOf(EntryMsg.class); + received.add((EntryMsg) msg); + } + return received; + } + + /** + * Acknowledges what arrived while the export was held and everything after it as it + * arrives, up to the {@code DoneMsg}, and checks that every entry of the domain arrived. A + * total update which was cut streams no further: the rest of its entries never arrives, or + * an {@code ErrorMsg} arrives in their place, and either fails here. The importer then + * leaves the full update status, which the exporter waits for before it releases its + * context - and it leaves it whatever happened, or the export never ends. + */ + private void finishExport(List held, long exportedEntries) throws Exception + { + try + { + final StringBuilder ldif = new StringBuilder(); + int lastMsgId = 0; + for (EntryMsg entryMsg : held) + { + ldif.append(new String(entryMsg.getEntryBytes(), UTF_8)); + lastMsgId = entryMsg.getMsgId(); + } + importer.publish(new InitializeRcvAckMsg(IMPORTER_ID, DS_ID, lastMsgId)); + final int heldAt = lastMsgId; + while (true) + { + final ReplicationMsg msg = receiveTotalUpdateMsg(60_000); + if (msg instanceof DoneMsg) + { + break; + } + assertThat(msg).as("the export was cut instead of streaming to its end") + .isInstanceOf(EntryMsg.class); + final EntryMsg entryMsg = (EntryMsg) msg; + ldif.append(new String(entryMsg.getEntryBytes(), UTF_8)); + lastMsgId = entryMsg.getMsgId(); + importer.publish(new InitializeRcvAckMsg(IMPORTER_ID, DS_ID, lastMsgId)); + } + assertThat(lastMsgId).as("the export did not stream past the window it was held at") + .isGreaterThan(heldAt); + assertThat(countEntries(ldif)).as("the export did not stream every entry of the domain") + .isEqualTo(exportedEntries); + } + finally + { + leaveTheFullUpdateStatus(); + final long deadline = System.currentTimeMillis() + 60_000; + while (domain.ieRunning()) + { + assertTrue(System.currentTimeMillis() < deadline, "the export did not end"); + Thread.sleep(50); + } + } + } + + /** Counts the entries of an LDIF stream by the blank line which separates them. */ + private static long countEntries(CharSequence ldif) + { + long count = 0; + for (int i = ldif.length() - 1; i > 0; i--) + { + if (ldif.charAt(i) == '\n' && ldif.charAt(i - 1) == '\n') + { + count++; + } + } + return count; + } + + /** + * The importer reconnects once its import is over - it comes back with the generation ID of + * the data it loaded, which is the one it was opened with here - and the exporter waits for + * the importer to leave the full update status before it releases its context. + */ + private void leaveTheFullUpdateStatus() + { + importer.reStart(true); + } + + /** + * Receives the next message of the total update on the importer: the updates of this + * replica's own and the topology are not it. + */ + private ReplicationMsg receiveTotalUpdateMsg(long timeoutMs) throws Exception + { + final long deadline = System.currentTimeMillis() + timeoutMs; + final List others = new ArrayList<>(); + while (System.currentTimeMillis() < deadline) + { + final ReplicationMsg msg; + try + { + msg = importer.receive(); + } + catch (SocketTimeoutException e) + { + continue; + } + if (msg instanceof InitializeTargetMsg || msg instanceof EntryMsg + || msg instanceof DoneMsg) + { + return msg; + } + if (msg instanceof ErrorMsg) + { + Assert.fail("the total update was given up: " + ((ErrorMsg) msg).getDetails()); + } + others.add(msg); + } + Assert.fail("nothing of the total update arrived within " + timeoutMs + " ms; received " + + others); + return null; + } + + /** + * Waits for the replication server to deliver the change to this replica: the listener + * thread of the domain puts it in the replay queue of the test, which takes it out. + */ + private LDAPUpdateMsg awaitDelivery(CSN csn, long timeoutMs, String orElse) throws Exception + { + final long deadline = System.currentTimeMillis() + timeoutMs; + while (queue.peek() == null) + { + assertTrue(System.currentTimeMillis() < deadline, orElse + " within " + timeoutMs + " ms"); + Thread.sleep(50); + } + final LDAPUpdateMsg msg = queue.take().getUpdateMessage(); + assertEquals(msg.getCSN(), csn, "another change than the one published was delivered"); + return msg; + } + + /** The records of the error log which carry the provided message id and the provided CSN. */ + private static List errorLogRecordsOf(int msgId, CSN csn) + { + final List records = new ArrayList<>(); + for (String record : TestCaseUtils.ERROR_TEXT_WRITER.getMessages()) + { + if (record.contains("msgID=" + msgId) && record.contains(csn.toString())) + { + records.add(record); + } + } + return records; + } + + private void replay(LDAPUpdateMsg ldapUpdate) + { + domain.markInProgress(ldapUpdate); + domain.replay(ldapUpdate, SHUTDOWN); + } +} diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/TestSynchronousReplayQueue.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/TestSynchronousReplayQueue.java index dd908c0623..a921f38b0d 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/TestSynchronousReplayQueue.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/TestSynchronousReplayQueue.java @@ -13,13 +13,15 @@ * * Copyright 2009 Sun Microsystems, Inc. * Portions copyright 2013-2015 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC. */ package org.opends.server.replication.plugin; import java.util.Collection; +import java.util.Deque; import java.util.Iterator; -import java.util.LinkedList; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentLinkedDeque; import java.util.concurrent.TimeUnit; /** @@ -30,7 +32,13 @@ */ public class TestSynchronousReplayQueue implements BlockingQueue { - private LinkedList list = new LinkedList<>(); + /** + * Written by the listener thread of the domain - a change the replication server delivers + * is offered here - and read by the thread of the test, which replays it: the two share no + * lock, so the deque has to be safe for that hand-off. {@code take()} is still synchronous - + * it throws when nothing was offered - which is what makes the queue a test one. + */ + private final Deque list = new ConcurrentLinkedDeque<>(); @Override public boolean add(UpdateToReplay e)