+ * The message was introduced by
+ * {@link ProtocolVersion#REPLICATION_PROTOCOL_V8} and has no encoding below
+ * it: an older peer is not told that the replica went offline, and cannot be.
+ * This answers for the message type, for a caller which has no instance at
+ * hand - see {@link #isEncodableFor(short)} for the one which has.
+ *
+ * @param protocolVersion
+ * The protocol version negotiated with the peer.
+ * @return true if that version carries this message,
+ * false otherwise.
+ */
+ public static boolean isSupportedBy(short protocolVersion)
+ {
+ return protocolVersion >= ProtocolVersion.REPLICATION_PROTOCOL_V8;
+ }
+
/** {@inheritDoc} */
@Override
public int size()
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/ReplicationMsg.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/ReplicationMsg.java
index bb0f2bd786..ce75fe6a19 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/ReplicationMsg.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/ReplicationMsg.java
@@ -13,6 +13,7 @@
*
* Copyright 2006-2010 Sun Microsystems, Inc.
* Portions copyright 2013-2016 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.replication.protocol;
@@ -108,6 +109,25 @@ protected ReplicationMsg()
*/
public abstract byte[] getBytes(short protocolVersion);
+ /**
+ * Whether this message has an encoding for a peer which negotiated the
+ * provided replication protocol version.
+ *
+ * A message which has none is dropped by {@link Session#publish(ReplicationMsg)}
+ * - its {@link #getBytes(short)} returns null - with nothing on
+ * the wire and nothing in the log, so a caller which goes on to report what
+ * the peer was told must ask this before it publishes.
+ *
+ * @param protocolVersion
+ * The protocol version negotiated with the peer.
+ * @return true if this message can be encoded for that version,
+ * false if it has no encoding for it at all.
+ */
+ public boolean isEncodableFor(short protocolVersion)
+ {
+ return true;
+ }
+
/**
* Generates a ReplicationMsg from its encoded form. This un-serialization is
* done taking into account the various supported replication protocol
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java
index 05232fe9a5..961cd88d88 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java
@@ -1721,21 +1721,25 @@ public void shutdown()
* to. With no such server connected there is nobody to forward the message to, and waiting
* would only delay the shutdown by the whole grace period.
*
+ * A peer which negotiated a protocol version the message has no encoding for is no such + * server either: nothing is ever sent to it and no writer will ever report a forward for it. + *
* The recipients DSRSShutdownSync records when the message is queued are the sharper source of
* truth and cover the domains this test lets through - a peer which connected after the
- * message was queued owes nothing, and is not waited for. What this test still covers is the
- * announcement which was never queued here at all, and for which no recipient can therefore be
- * recorded: the message of a replica which picked a remote replication server, or the
- * announcement of issue #918 recorded after its message was relayed. Those wait out the whole
- * grace period on the first forward, and on a server with no peer connected nothing would ever
- * report one.
+ * message was queued owes nothing, and is not waited for, and the writer of a peer which
+ * cannot decode the message strikes it off when it drops it. What this test still covers is
+ * the announcement which was never queued here at all, and for which no recipient can
+ * therefore be recorded: the message of a replica which picked a remote replication server, or
+ * the announcement of issue #918 recorded after its message was relayed. Those wait out the
+ * whole grace period on the first forward, and on a server whose only peers cannot decode the
+ * message - or with no peer connected at all - nothing would ever report one.
*/
private void awaitReplicaOfflineMsgsForwarded()
{
final List
+ * The drop is reported rather than traced whoever the consumer is: today the only message
+ * gated on a protocol version is the ReplicaOfflineMsg, which never reaches the writer of
+ * a directory server at all - ReplicationServerDomain.put() does not queue it for one, and
+ * DataServerHandler.updateServerState() drops the copy the changelog cursor of a directory
+ * server which is catching up synthesizes from the offline CSN of the replica (issue
+ * #1029). A message which does get here is one the consumer was to be sent and will not
+ * be, which is what this record says.
+ */
+ logger.warn(WARN_IGNORING_UPDATE_UNSUPPORTED_BY_PEER,
+ handler.getReplicationServerId(), updateMsg.getCSN(), handler.getBaseDN(),
+ handler.getServerId(), session.getReadableRemoteAddress(),
+ handler.getProtocolVersion());
+ return true;
+ }
if (handler.isDataServer())
{
/**
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 594524fb59..a4eb51f839 100644
--- a/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
+++ b/opendj-server-legacy/src/messages/org/opends/messages/replication.properties
@@ -709,3 +709,6 @@ ERR_INIT_SESSION_STOPPED_DURING_IMPORT_329=Domain %s (server id: %s) : the sessi
replication server was stopped before the initialization from server %s completed. The \
entries which had arrived are imported, and the generation id of the data is computed over \
them rather than taken from the exporter
+WARN_IGNORING_UPDATE_UNSUPPORTED_BY_PEER_328=Replication server RS(%d) not sending update \
+ %s for domain "%s" to server %d at %s because the replication protocol version %d \
+ negotiated with it has no encoding for this message
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SynchronizationMsgTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SynchronizationMsgTest.java
index bba5b3ac61..2979910b51 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SynchronizationMsgTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/protocol/SynchronizationMsgTest.java
@@ -814,6 +814,22 @@ public void replicaOfflineMsgTest() throws Exception
assertEquals(decodedMsg.getCSN(), expectedMsg.getCSN());
}
+ /**
+ * The ReplicaOfflineMsg was introduced by the 8th version of the replication protocol and has
+ * no encoding below it, where getBytes() returns null. Session.publish() drops such a message
+ * without a trace, so whoever goes on to report that a peer was told must ask first.
+ */
+ @Test
+ public void replicaOfflineMsgHasNoEncodingBelowV8() throws Exception
+ {
+ final ReplicaOfflineMsg msg = new ReplicaOfflineMsg(new CSN(System.currentTimeMillis(), 0, 42));
+
+ assertTrue(msg.isEncodableFor(REPLICATION_PROTOCOL_V8));
+ assertNotNull(msg.getBytes(REPLICATION_PROTOCOL_V8));
+ assertFalse(msg.isEncodableFor(REPLICATION_PROTOCOL_V7));
+ assertNull(msg.getBytes(REPLICATION_PROTOCOL_V7));
+ }
+
/**
* Test that WindowMsg encoding and decoding works
* by checking that : msg == new WindowMsg(msg.getBytes()).
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerShutdownSyncTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerShutdownSyncTest.java
index 31d948d936..c8e503c651 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerShutdownSyncTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ReplicationServerShutdownSyncTest.java
@@ -16,6 +16,7 @@
package org.opends.server.replication.server;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.opends.messages.ReplicationMessages.WARN_IGNORING_UPDATE_UNSUPPORTED_BY_PEER;
import static org.opends.server.TestCaseUtils.TEST_ROOT_DN_STRING;
import static org.opends.server.util.CollectionUtils.newArrayList;
@@ -23,6 +24,7 @@
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.TreeSet;
@@ -46,6 +48,7 @@
import org.opends.server.replication.common.RSInfo;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.protocol.DeleteMsg;
+import org.opends.server.replication.protocol.ProtocolVersion;
import org.opends.server.replication.protocol.ReplServerStartMsg;
import org.opends.server.replication.protocol.ReplSessionSecurity;
import org.opends.server.replication.protocol.ReplicaOfflineMsg;
@@ -58,6 +61,7 @@
import org.opends.server.replication.service.ReplicationBroker;
import org.opends.server.util.StaticUtils;
import org.opends.server.util.TestTimer;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
/**
@@ -96,6 +100,15 @@ public class ReplicationServerShutdownSyncTest extends ReplicationTestCase
* published one, so the shutdown spends its whole grace period waiting for it.
*/
private static final int UNREACHABLE_DS_ID = 98;
+ /** The peer replication server which speaks a protocol version with no ReplicaOfflineMsg. */
+ private static final int OLD_PEER_RS_ID = 99;
+ /**
+ * Base of the id of the replication server serving a peer of a chosen protocol version: the
+ * version is added to it, so that the two rows of
+ * {@link #onlyAPeerWhichCanDecodeTheMessageIsToldTheReplicaWentOffline(short, boolean)} do not
+ * share a server id.
+ */
+ private static final int PEER_VERSION_RS_ID = 8240;
/** Send window a peer advertises when nothing has to hold its writer back. */
private static final int PEER_WINDOW = 100;
/**
@@ -244,6 +257,155 @@ public void thePeerReceivesTheReplicaOfflineMsgBeforeTheShutdownReturns() throws
}
}
+ @DataProvider
+ public Object[][] peerProtocolVersions()
+ {
+ return new Object[][] {
+ { ProtocolVersion.getCurrentVersion(), true },
+ { ProtocolVersion.REPLICATION_PROTOCOL_V7, false },
+ };
+ }
+
+ /**
+ * A peer replication server which negotiated a protocol version older than the one which
+ * introduced the ReplicaOfflineMsg cannot be told that a replica went offline: the message has
+ * no encoding for such a peer and {@link Session#publish(ReplicationMsg)} drops what it cannot
+ * encode. Nothing can be done about the peer itself - the announcement is not part of the
+ * protocol it speaks - but the writer must not report that drop as a forward, or the shutdown
+ * counts this peer as told by a writer which sent nothing - and, for an announcement no
+ * recipient was recorded for, stops waiting for the peers which do need the message.
+ *
+ * The drop is reported to the operator as well, once for the copy the writer took, and that
+ * record is the only account of a peer the topology leaves behind: what it names is asserted
+ * here next to what the writer told the shutdown. The peer is waited on until it is served
+ * from its queue - see {@link #waitForFollowing(ReplicationServerDomain, int)} - so that the
+ * writer is handed the announcement once and the count is not a race.
+ *
+ * The change published after the announcement is the synchronization point: the writer of a
+ * peer takes from one queue in order, so a peer which has received that change has already
+ * dealt with the announcement which precedes it - forwarded it, or dropped it, logged the drop
+ * and told the shutdown to stop waiting for this peer.
+ */
+ @Test(dataProvider = "peerProtocolVersions")
+ public void onlyAPeerWhichCanDecodeTheMessageIsToldTheReplicaWentOffline(
+ short peerVersion, boolean expectedToBeTold) throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ final RecordingShutdownSync shutdownSync = new RecordingShutdownSync();
+ ReplicationServer replicationServer = null;
+ ReplicationBroker broker = null;
+ FakePeerReplicationServer peer = null;
+ try
+ {
+ final int replicationPort = TestCaseUtils.findFreePort();
+ replicationServer = newReplicationServer(shutdownSync,
+ "shutdownSyncPeerVersion" + peerVersion + "Db", PEER_VERSION_RS_ID + peerVersion,
+ replicationPort);
+ broker = openReplicationSession(baseDN, LOCAL_DS_ID, 100, replicationPort, 5000, EMPTY_DN_GENID);
+ peer = FakePeerReplicationServer.connected(
+ replicationPort, REMOTE_RS_ID, baseDN, EMPTY_DN_GENID, PEER_WINDOW, peerVersion);
+
+ final ReplicationServerDomain domain =
+ replicationServer.getReplicationServerDomain(baseDN, true);
+ waitForConnectedReplicationServer(domain, REMOTE_RS_ID);
+ waitForFollowing(domain, REMOTE_RS_ID);
+ final Future
+ * What decides is whether any connected peer can be told, not which handler is
+ * iterated first - {@code connectedRSs} is a ConcurrentHashMap, so the order the two peers
+ * below are registered in is not the order they are read in. This case asserts the wait
+ * rather than the peer which caused it for that reason.
+ */
+ @Test
+ public void shutdownWaitsWhenOnlySomeOfTheConnectedPeersCanDecodeTheMessage() throws Exception
+ {
+ final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
+ final DSRSShutdownSync shutdownSync = new DSRSShutdownSync();
+ ReplicationServer replicationServer = null;
+ try (ServerSocket oldPeerListen = TestCaseUtils.bindFreePort();
+ ServerSocket currentPeerListen = TestCaseUtils.bindFreePort())
+ {
+ oldPeerListen.setSoTimeout(SOCKET_TIMEOUT_MS);
+ currentPeerListen.setSoTimeout(SOCKET_TIMEOUT_MS);
+ replicationServer = newReplicationServer(shutdownSync, "shutdownSyncMixedPeersDb", 8237);
+ final Session[] oldPeerPair = connectSessionPair(oldPeerListen, getReplSessionSecurity());
+ final Session[] currentPeerPair =
+ connectSessionPair(currentPeerListen, getReplSessionSecurity());
+ try (Session oldPeerRemoteEnd = oldPeerPair[0];
+ Session oldPeerSession = oldPeerPair[1];
+ Session currentPeerRemoteEnd = currentPeerPair[0];
+ Session currentPeerSession = currentPeerPair[1])
+ {
+ oldPeerSession.setProtocolVersion(ProtocolVersion.REPLICATION_PROTOCOL_V7);
+ currentPeerSession.setProtocolVersion(ProtocolVersion.getCurrentVersion());
+ registerConnectedReplicationServer(
+ replicationServer, baseDN, oldPeerSession, OLD_PEER_RS_ID);
+ registerConnectedReplicationServer(
+ replicationServer, baseDN, currentPeerSession, REMOTE_RS_ID);
+
+ final long startTime = System.nanoTime();
+ shutdownSync.replicaOfflineMsgSent(baseDN, newOfflineCSN());
+ replicationServer.shutdown();
+ final long elapsed = elapsedMillis(startTime);
+
+ assertThat(elapsed)
+ .as("the shutdown stopped waiting although a connected peer can still be told that "
+ + "the replica went offline")
+ .isGreaterThanOrEqualTo(DSRSShutdownSync.REPLICA_OFFLINE_GRACE_PERIOD);
+ }
+ }
+ finally
+ {
+ removeQuietly(replicationServer);
+ }
+ }
+
/**
* The writer serving a directory server must not hold back the shutdown either: it used to
* loop on the pending message until the grace period expired, although its handler had already
@@ -1128,6 +1383,37 @@ public void call() throws Exception
});
}
+ /**
+ * Waits for the handler of the peer to be served from its in-memory queue rather than from the
+ * changelog.
+ *
+ * A freshly connected peer is behind by definition, and a handler which is catching up reads
+ * its updates from the changelog, where {@code ReplicaCursor} synthesizes a
+ * {@link ReplicaOfflineMsg} from the offline CSN of the replica - on top of the copy
+ * {@code ReplicationServerDomain.put()} queues for it, and again on every refill of its late
+ * queue, since an offline CSN never enters the state of a handler. How many copies of one
+ * announcement its writer takes is then a race, which is a race on how many times the writer
+ * reports the drop.
+ */
+ private void waitForFollowing(final ReplicationServerDomain domain, final int serverId)
+ throws Exception
+ {
+ newConnectionTimer().repeatUntilSuccess(new TestTimer.CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ final ReplicationServerHandler rsHandler = domain.getConnectedRSs().get(serverId);
+ assertThat(rsHandler)
+ .as("the peer replication server %s is no longer connected", serverId).isNotNull();
+ assertThat(rsHandler.isFollowing())
+ .as("the peer replication server %s is still catching up from the changelog",
+ serverId)
+ .isTrue();
+ }
+ });
+ }
+
/**
* Waits for the collocated directory server to be connected and for its handshake to be over -
* {@link #waitForConnectedReplicationServer(ReplicationServerDomain, int)} says what the
@@ -1262,6 +1548,25 @@ private static long elapsedMillis(long startTime)
return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime);
}
+ /** The last message a peer read, null if it read none at all. */
+ private static ReplicationMsg lastOf(List