From b8c3c195995500ba0351a3736ad43147a6036dd0 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Fri, 25 Sep 2026 06:26:29 +0000
Subject: [PATCH] [#1055] Report a ReplicaOfflineMsg forwarded once it is written to the peer, not once it is queued (#1057)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java | 136 ++++++++++++++++++++++++++++++++++++++------
1 files changed, 116 insertions(+), 20 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java
index bd867c0..f92154f 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java
@@ -40,6 +40,7 @@
import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.i18n.slf4j.LocalizedLogger;
+import org.forgerock.util.annotations.VisibleForTesting;
import org.opends.server.api.DirectoryThread;
import org.opends.server.types.HostPort;
import org.opends.server.util.StaticUtils;
@@ -112,8 +113,27 @@
*/
private BufferedOutputStream output;
- private final LinkedBlockingQueue<byte[]> sendQueue = new LinkedBlockingQueue<>(4000);
+ /** A message queued for the thread of this session, and what to run once it is written. */
+ private static final class Outgoing
+ {
+ private final byte[] buffer;
+ private final Runnable whenWritten;
+
+ private Outgoing(byte[] buffer, Runnable whenWritten)
+ {
+ this.buffer = buffer;
+ this.whenWritten = whenWritten;
+ }
+ }
+
+ private final LinkedBlockingQueue<Outgoing> sendQueue = new LinkedBlockingQueue<>(4000);
private AtomicBoolean isRunning = new AtomicBoolean(false);
+ /**
+ * What {@link #publish(ReplicationMsg, Runnable)} runs between its check that no close has
+ * begun and the offer of the message to {@code sendQueue}, or null. Only the tests set it - see
+ * {@link #beforeQueueing(Runnable)}.
+ */
+ private volatile Runnable beforeQueueing;
private final CountDownLatch latch = new CountDownLatch(1);
/**
@@ -166,8 +186,10 @@
* This object won't be used anymore after this method is called.
* <p>
* A message which was published on this session but which its publisher thread had not sent yet
- * is sent here rather than dropped, within the budget of {@link #DRAIN_BUDGET_MS}. See {@link
- * #sendWhatThePublisherLeftQueued()}.
+ * is sent here rather than dropped, within the budget of {@link #DRAIN_BUDGET_MS}, and its
+ * callback runs here once it is written. See {@link #sendWhatThePublisherLeftQueued()}. What the
+ * close gives up on is not written, and the callbacks of those messages never run - see
+ * {@link #publish(ReplicationMsg, Runnable)}.
*/
@Override
public void close()
@@ -341,8 +363,8 @@
private void sendWhatThePublisherLeftQueued()
{
final long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(DRAIN_BUDGET_MS);
- byte[] buffer;
- while ((buffer = sendQueue.poll()) != null)
+ Outgoing outgoing;
+ while ((outgoing = sendQueue.poll()) != null)
{
if (System.nanoTime() - deadline >= 0)
{
@@ -352,7 +374,7 @@
}
try
{
- send(buffer);
+ send(outgoing.buffer);
}
catch (final IOException e)
{
@@ -367,6 +389,7 @@
"the write failed with " + e.getClass().getName() + ": " + e.getMessage());
return;
}
+ written(outgoing.whenWritten);
}
}
@@ -496,26 +519,57 @@
*/
public void publish(final ReplicationMsg msg) throws IOException
{
+ publish(msg, null);
+ }
+
+ /**
+ * Sends a replication message to the remote peer, and runs the provided callback once the
+ * message has been written to the socket.
+ * <p>
+ * While the thread of this session runs, a message published is queued for it and written
+ * later, so the return of this method says only that the message is queued. The callback is
+ * the only word that the message has left this server: it runs once, on the thread which wrote
+ * the message, after the write returned - the thread of the session, or the one closing it for
+ * a message the close sends out of the queue - and never for a message which was not written,
+ * which is what becomes of a message the write of which fails, and of what a close gives up on
+ * (see {@link #close()}). It must be short and must not block: the session writes nothing else
+ * until it returns.
+ *
+ * @param msg
+ * The message to be sent.
+ * @param whenWritten
+ * What to run once the message has been written, or null.
+ * @return whether the message was written or queued to be written; false when it was neither,
+ * because it has no encoding for the protocol version of the peer or because the
+ * session is being closed - the callback then never runs. A message queued after a
+ * close drained the queue is taken back, and counts as neither.
+ * @throws IOException
+ * If an IO error occurred.
+ */
+ public boolean publish(final ReplicationMsg msg, final Runnable whenWritten) throws IOException
+ {
final byte[] buffer = msg.getBytes(protocolVersion);
if (buffer == null)
{
// skip anything that cannot be encoded for this peer.
- return;
+ return false;
}
if (isRunning.get())
{
+ final Outgoing outgoing = new Outgoing(buffer, whenWritten);
while (!closeInitiated)
{
+ final Runnable hook = beforeQueueing;
+ if (hook != null)
+ {
+ hook.run();
+ }
try
{
// Avoid blocking forever so that we can check for session closure.
- if (sendQueue.offer(buffer, 100, TimeUnit.MILLISECONDS))
+ if (sendQueue.offer(outgoing, 100, TimeUnit.MILLISECONDS))
{
- if (!isRunning.get())
- {
- takeBackWhatWasQueuedTooLate(buffer);
- }
- return;
+ return isRunning.get() || !takeBackWhatWasQueuedTooLate(outgoing);
}
}
catch (final InterruptedException e)
@@ -524,10 +578,27 @@
throw new IOException(e.getMessage());
}
}
+ return false;
}
- else
+ send(buffer);
+ written(whenWritten);
+ return true;
+ }
+
+ /** Runs what was to run once a message is written; a callback which fails takes nothing down. */
+ private void written(final Runnable whenWritten)
+ {
+ if (whenWritten != null)
{
- send(buffer);
+ try
+ {
+ whenWritten.run();
+ }
+ catch (final RuntimeException e)
+ {
+ logger.error(LocalizableMessage.raw("The callback of a message written to %s failed: %s",
+ readableRemoteAddress, stackTraceToSingleLineString(e)));
+ }
}
}
@@ -542,16 +613,21 @@
* what nothing sends. A buffer queued before the session came off the queueing branch is left
* to the drain - it cannot have seen the flag cleared - and a buffer the drain or the close
* already took is not found here, so nothing is reported twice.
+ *
+ * @return whether the message was taken back - it is then never written, and its callback never
+ * runs
*/
- private void takeBackWhatWasQueuedTooLate(final byte[] buffer)
+ private boolean takeBackWhatWasQueuedTooLate(final Outgoing outgoing)
{
publishLock.lock();
try
{
- if (sendQueue.remove(buffer))
+ if (sendQueue.remove(outgoing))
{
reportQueueNotSent(1, "it was queued after the publisher of the session had stopped");
+ return true;
}
+ return false;
}
finally
{
@@ -559,6 +635,24 @@
}
}
+ /**
+ * Sets what {@link #publish(ReplicationMsg, Runnable)} runs between its check that no close has
+ * begun and the offer of the message to the queue.
+ * <p>
+ * Only there for the tests of {@link #takeBackWhatWasQueuedTooLate(Outgoing)}: a
+ * {@code publish()} descheduled at that spot is the only one which can queue a message after a
+ * close has drained the queue, and nothing else holds a thread there on cue while the close
+ * runs to its end.
+ *
+ * @param hook
+ * What to run there, on the publishing thread, or null for nothing.
+ */
+ @VisibleForTesting
+ void beforeQueueing(final Runnable hook)
+ {
+ beforeQueueing = hook;
+ }
+
/** Sends a replication message already encoded to the socket.
*
* @param buffer
@@ -769,10 +863,10 @@
boolean needClosing = false;
while (!closeInitiated)
{
- byte[] buffer;
+ Outgoing outgoing;
try
{
- buffer = sendQueue.take();
+ outgoing = sendQueue.take();
}
catch (InterruptedException ie)
{
@@ -780,14 +874,16 @@
}
try
{
- send(buffer);
+ send(outgoing.buffer);
}
catch (IOException e)
{
setSessionError(e);
publisherFailedWrites.incrementAndGet();
needClosing = true;
+ continue;
}
+ written(outgoing.whenWritten);
}
/*
* A close clears the flag itself, under publishLock, once it has joined this thread - see
--
Gitblit v1.10.0