From 13d57e063cbd3e884f787744a9eaedbfb98b5a51 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Fri, 11 Sep 2026 13:19:34 +0000
Subject: [PATCH] [#949] Report a ReplicaOfflineMsg the broker refused as not sent (#976)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java | 37 +++++++++---
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java | 5 +
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PendingChangesTest.java | 61 +++++++++++++++++--
opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java | 7 ++
opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java | 14 ++++
5 files changed, 101 insertions(+), 23 deletions(-)
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 25b6418..64eded8 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
@@ -2221,8 +2221,9 @@
else if (logger.isTraceEnabled())
{
logger.trace("Replica " + getServerId() + " of domain baseDN=" + getBaseDN()
- + " could not announce itself offline: a change which is still in flight holds"
- + " the message back, and " + pendingChanges.size() + " change(s) are pending");
+ + " could not announce itself offline: the message was not published - a change which"
+ + " is still in flight holds it back, or the broker had no session to write it to,"
+ + " and " + pendingChanges.size() + " change(s) are pending");
}
}
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java
index d718937..4f88b62 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java
@@ -132,8 +132,8 @@
* Such a message is given up on rather than left queued, so that it is neither reported as
* sent nor published later on the session which follows.
*
- * @return the CSN of the message which was published, or {@code null} if it could not be
- * published
+ * @return the CSN of the message which was published, or {@code null} if it could not be:
+ * a change which is still in flight holds it back, or the broker refused it
*/
public synchronized CSN putReplicaOfflineMsg()
{
@@ -143,23 +143,36 @@
pendingChange.setCommitted(true);
pendingChanges.put(offlineCSN, pendingChange);
- pushCommittedChanges();
- // pushCommittedChanges() removes whatever it published, so the message is still listed
- // here if and only if a change before it held it back.
- final boolean heldBack = pendingChanges.remove(offlineCSN) != null;
- return heldBack ? null : offlineCSN;
+ /*
+ * The message is the last change of the queue, so a push which did not reach it, or which
+ * the broker refused, reports another CSN or none.
+ */
+ final boolean published = offlineCSN.equals(pushCommittedChanges());
+ // pushCommittedChanges() removes whatever it reached, so the message is still listed here
+ // if and only if a change before it held it back - it is dropped rather than left queued.
+ pendingChanges.remove(offlineCSN);
+ return published ? offlineCSN : null;
}
/**
* Push all committed local changes to the replicationServer service.
+ *
+ * @return the CSN of the last {@link ReplicaOfflineMsg} the replication service accepted, or
+ * {@code null} if none was pushed or the broker refused it. The announcement that a
+ * replica goes offline is the one message whose delivery the caller must know about:
+ * it is stored nowhere, so nothing publishes it again, while a change the broker
+ * refuses is republished from the historical information of its entry on the next
+ * session.
*/
- synchronized void pushCommittedChanges()
+ synchronized CSN pushCommittedChanges()
{
+ CSN publishedOfflineCSN = null;
+
// peek the oldest change
Entry<CSN, PendingChange> firstEntry = pendingChanges.firstEntry();
if (firstEntry == null)
{
- return;
+ return null;
}
PendingChange firstChange = firstEntry.getValue();
@@ -185,7 +198,10 @@
}
else if (msg instanceof ReplicaOfflineMsg)
{
- domain.publish(msg);
+ if (domain.publish(msg))
+ {
+ publishedOfflineCSN = msg.getCSN();
+ }
}
// false warning: firstEntry will not be null if firstChange is not null
@@ -195,6 +211,7 @@
firstEntry = pendingChanges.firstEntry();
firstChange = firstEntry != null ? firstEntry.getValue() : null;
}
+ return publishedOfflineCSN;
}
/**
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java
index b99c5bb..91c68af 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java
@@ -2436,7 +2436,12 @@
}
}
}
- return true;
+ /*
+ * The loop also ends when the broker is stopped, and then nothing was written to any
+ * session: a caller told the message was published would report as sent a message which
+ * never reached the wire.
+ */
+ return done;
}
/**
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java
index 879bfb4..e2babe2 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java
@@ -3696,15 +3696,25 @@
* {@link #processUpdate(UpdateMsg)} message.
*
* @param msg The UpdateMsg that should be published.
+ * @return {@code true} if the message was written to the session of the
+ * Replication Service, {@code false} if the broker could not write
+ * it: it has no usable session, the changes which come before this
+ * one still have to be republished, or it was stopped in between.
*/
- public void publish(UpdateMsg msg)
+ public boolean publish(UpdateMsg msg)
{
- broker.publish(msg);
+ final boolean published = broker.publish(msg, true);
+ /*
+ * The domain state is updated whether or not the message was written: it says which changes
+ * this replica has done, and it is by finding it ahead of the state its replication server
+ * reports that the next session knows which changes to republish.
+ */
if (msg.contributesToDomainState())
{
state.update(msg.getCSN());
}
numSentUpdates.incrementAndGet();
+ return published;
}
/**
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PendingChangesTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PendingChangesTest.java
index 04f95ed..23c026f 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PendingChangesTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/PendingChangesTest.java
@@ -36,7 +36,7 @@
/**
* Tests the bookkeeping a replica does on its own changes: they are published in the order of
* their CSNs, and the announcement that the replica goes offline is only reported as sent when
- * it really was.
+ * it really was published.
* <p>
* These tests need no server: the changes are built by a CSNGenerator, which reads the time
* service, and the time service is up as soon as its class is loaded.
@@ -48,9 +48,9 @@
private static final int SERVER_ID = 42;
@Test
- public void replicaOfflineMsgIsSentWhenNoChangeIsPending() throws Exception
+ public void replicaOfflineMsgTheBrokerPublishedIsReportedAsSent() throws Exception
{
- final ReplicationDomain domain = mock(ReplicationDomain.class);
+ final ReplicationDomain domain = domainWhichPublishes(true);
final PendingChanges pendingChanges = newPendingChanges(domain);
final CSN offlineCSN = pendingChanges.putReplicaOfflineMsg();
@@ -62,15 +62,31 @@
}
/**
+ * The broker writes nothing when it has no usable session, when the changes which come before
+ * this one still have to be republished by the recovery, or when it is stopped in between - and
+ * what was not written must not be reported as sent: the shutdown of a collocated replication
+ * server waits out the whole grace period of a message it was told about and which never
+ * reached the wire.
+ */
+ @Test
+ public void replicaOfflineMsgTheBrokerRefusedIsNotReportedAsSent() throws Exception
+ {
+ final ReplicationDomain domain = domainWhichPublishes(false);
+ final PendingChanges pendingChanges = newPendingChanges(domain);
+
+ assertNull(pendingChanges.putReplicaOfflineMsg(), "the broker refused the message");
+
+ assertTrue(onlyMsgPublishedBy(domain) instanceof ReplicaOfflineMsg, "it was attempted");
+ }
+
+ /**
* The message carries the newest CSN of the replica, so a change which is still in flight
- * holds it back - and what was never published must not be reported as sent: the shutdown of
- * a collocated replication server waits out the whole grace period of a message it was told
- * about and which never reaches the wire.
+ * holds it back, and the broker is never even asked to publish it.
*/
@Test
public void replicaOfflineMsgQueuedBehindAnUncommittedChangeIsNotReportedAsSent() throws Exception
{
- final ReplicationDomain domain = mock(ReplicationDomain.class);
+ final ReplicationDomain domain = domainWhichPublishes(true);
final PendingChanges pendingChanges = newPendingChanges(domain);
pendingChanges.putLocalOperation(newLocalOperation());
@@ -88,7 +104,7 @@
@Test
public void replicaOfflineMsgWhichCouldNotBeSentIsNotPublishedLater() throws Exception
{
- final ReplicationDomain domain = mock(ReplicationDomain.class);
+ final ReplicationDomain domain = domainWhichPublishes(true);
final PendingChanges pendingChanges = newPendingChanges(domain);
final CSN changeCSN = pendingChanges.putLocalOperation(newLocalOperation());
assertNull(pendingChanges.putReplicaOfflineMsg(), "nothing was published");
@@ -100,11 +116,40 @@
assertTrue(published instanceof LDAPUpdateMsg, "published " + published);
}
+ /**
+ * A change the broker refused leaves the pending changes all the same: the replica has done
+ * it, its ServerState says so, and it is by finding that state ahead of the one its
+ * replication server reports that the next session republishes the change from the historical
+ * information of its entry. Only the offline announcement, which is stored nowhere, needs the
+ * answer of the broker.
+ */
+ @Test
+ public void changeTheBrokerRefusedStillLeavesThePendingChanges() throws Exception
+ {
+ final ReplicationDomain domain = domainWhichPublishes(false);
+ final PendingChanges pendingChanges = newPendingChanges(domain);
+ final CSN changeCSN = pendingChanges.putLocalOperation(newLocalOperation());
+ assertEquals(pendingChanges.size(), 1);
+
+ pendingChanges.commitAndPushCommittedChanges(changeCSN, mock(LDAPUpdateMsg.class));
+
+ assertTrue(onlyMsgPublishedBy(domain) instanceof LDAPUpdateMsg, "the change was published");
+ assertEquals(pendingChanges.size(), 0, "and is not queued for a second attempt");
+ }
+
private PendingChanges newPendingChanges(ReplicationDomain domain)
{
return new PendingChanges(new CSNGenerator(SERVER_ID, 0), domain);
}
+ /** A domain whose broker accepts, or refuses, whatever it is given to publish. */
+ private ReplicationDomain domainWhichPublishes(boolean accepted)
+ {
+ final ReplicationDomain domain = mock(ReplicationDomain.class);
+ when(domain.publish(any(UpdateMsg.class))).thenReturn(accepted);
+ return domain;
+ }
+
/** A local operation, i.e. one this replica must publish to the other replicas. */
private PluginOperation newLocalOperation()
{
--
Gitblit v1.10.0