From 92d88ca699cd8090a26b92cbe46789d2b848195f Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Mon, 07 Sep 2026 09:30:36 +0000
Subject: [PATCH] [#889] Keep a change the replay could not apply out of the ServerState (#892)
---
opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java | 728 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 728 insertions(+), 0 deletions(-)
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java
index 4c9dde1..b9ce72b 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java
@@ -26,6 +26,7 @@
import static org.opends.server.protocols.internal.InternalClientConnection.*;
import static org.opends.server.replication.plugin.LDAPReplicationDomain.*;
import static org.opends.server.util.CollectionUtils.*;
+import static org.opends.server.util.ServerConstants.*;
import static org.testng.Assert.*;
import java.net.SocketTimeoutException;
@@ -35,7 +36,9 @@
import org.assertj.core.api.Assertions;
import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.i18n.slf4j.LocalizedLogger;
+import org.forgerock.opendj.ldap.ByteString;
import org.forgerock.opendj.ldap.DN;
+import org.forgerock.opendj.ldap.DecodeException;
import org.forgerock.opendj.ldap.ModificationType;
import org.forgerock.opendj.ldap.ResultCode;
import org.forgerock.opendj.ldap.requests.ModifyDNRequest;
@@ -46,25 +49,33 @@
import org.opends.server.core.DeleteOperation;
import org.opends.server.core.DirectoryServer;
import org.opends.server.core.ModifyOperation;
+import org.opends.server.core.ModifyOperationBasis;
import org.opends.server.extensions.DummyAlertHandler;
import org.opends.server.plugins.ShortCircuitPlugin;
+import org.opends.server.protocols.internal.InternalClientConnection;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.CSNGenerator;
+import org.opends.server.replication.plugin.LDAPReplicationDomain;
+import org.opends.server.replication.plugin.MultimasterReplication;
import org.opends.server.replication.protocol.AddMsg;
import org.opends.server.replication.protocol.DeleteMsg;
import org.opends.server.replication.protocol.HeartbeatThread;
import org.opends.server.replication.protocol.LDAPUpdateMsg;
+import org.opends.server.replication.protocol.ModifyContext;
import org.opends.server.replication.protocol.ModifyDNMsg;
import org.opends.server.replication.protocol.ModifyMsg;
import org.opends.server.replication.protocol.OperationContext;
+import org.opends.server.replication.protocol.ProtocolVersion;
import org.opends.server.replication.protocol.ReplicationMsg;
import org.opends.server.replication.service.ReplicationBroker;
import org.opends.server.types.Attribute;
import org.opends.server.types.Attributes;
import org.opends.server.types.Entry;
+import org.opends.server.types.LDAPException;
import org.opends.server.types.Modification;
import org.opends.server.types.Operation;
import org.opends.server.types.OperationType;
+import org.opends.server.types.RawModification;
import org.opends.server.util.TestTimer;
import org.opends.server.util.TestTimer.CallableVoid;
import org.opends.server.util.TimeThread;
@@ -81,6 +92,13 @@
{
private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();
+ /**
+ * How long a change is retried in the tests which check that this replica gives up on
+ * a change it can never apply: long enough for the change to be delivered again a
+ * couple of times, short enough not to make the test wait out a real backend outage.
+ */
+ private static final long TEST_GIVE_UP_DELAY_IN_MS = 2000;
+
/** An entry with a entryUUID. */
private Entry personWithUUIDEntry;
private Entry personWithSecondUniqueID;
@@ -1376,6 +1394,716 @@
}
/**
+ * Test case for [Issue 889]: a change whose replay failed on the server itself must
+ * not be recorded as replayed. Recording it would advance the ServerState past the
+ * change, so the replication server would never send it again while this replica
+ * reports itself up to date.
+ */
+ @Test
+ public void failedReplayIsNotRecordedAsReplayed() throws Exception
+ {
+ testSetUp("failedReplayIsNotRecordedAsReplayed");
+ logger.error(LocalizableMessage.raw("Starting replication test : failedReplayIsNotRecordedAsReplayed"));
+
+ final int serverId = 12;
+ ReplicationBroker broker =
+ openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
+ try
+ {
+ CSNGenerator gen = new CSNGenerator(serverId, 0);
+
+ Entry tmp = TestCaseUtils.addEntry(
+ "dn: uid=user.889," + baseDN,
+ "objectClass: top",
+ "objectClass: person",
+ "objectClass: organizationalPerson",
+ "objectClass: inetOrgPerson",
+ "uid: user.889",
+ "cn: Aaccf Amar",
+ "sn: Amar");
+ String uuid = getEntry(tmp.getName(), 1, true).parseAttribute("entryuuid").asString();
+
+ final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
+ final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
+ domain.resetUnreplayedChangeAlertThrottle();
+ final int initialAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE);
+ final long giveUpDelay = domain.getReplayGiveUpDelay();
+ try
+ {
+ // A backend which is down for maintenance is waited out for minutes: this test
+ // can not, so the change is given up on after a couple of deliveries instead.
+ // Set inside the try which puts it back, like the short circuit below: both are
+ // the domain's and the server's for as long as they are left behind.
+ domain.setReplayGiveUpDelay(TEST_GIVE_UP_DELAY_IN_MS);
+ /*
+ * Fail the replay the way a storage failure does: the backend reports it with the
+ * server-error-result-code, 80 by default. The short circuit has to be set at the
+ * pre-parse plugin point, the pre-operation ones are not invoked for
+ * synchronization operations.
+ */
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue());
+
+ final CSN csn = gen.newCSN();
+ broker.publish(new DeleteMsg(tmp.getName(), csn, uuid));
+
+ /*
+ * The replication server resumes from the ServerState of this replica, so it only
+ * sends the change again as long as the state does not cover it: seeing the same
+ * change delivered more than once is what tells that it was not recorded as
+ * replayed.
+ *
+ * One delivery is retried in place IN_PLACE_REPLAY_ATTEMPTS times before the
+ * session is restarted, so it takes more than that many short circuits to prove
+ * that the change was delivered a second time.
+ */
+ TestTimer timer = new TestTimer.Builder()
+ .maxSleep(60, SECONDS)
+ .sleepTimes(100, MILLISECONDS)
+ .toTimer();
+ timer.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ assertTrue(ShortCircuitPlugin.getShortCircuitCount(OperationType.DELETE, "PreParse")
+ > IN_PLACE_REPLAY_ATTEMPTS,
+ "the change was not sent again after its replay failed");
+ }
+ });
+ assertNotNull(getEntry(tmp.getName(), 1, true), "the entry must not have been deleted");
+
+ /*
+ * The change can never be applied here, so the replica eventually gives up on it
+ * rather than stopping for good: it then warns that it has diverged.
+ */
+ TestTimer giveUpTimer = new TestTimer.Builder()
+ .maxSleep(120, SECONDS)
+ .sleepTimes(200, MILLISECONDS)
+ .toTimer();
+ giveUpTimer.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ assertTrue(domain.getServerState().cover(csn),
+ "the replica did not give up on a change it can never replay");
+ }
+ });
+ assertMonitorAttrValueEventually(baseDN, "replayed-updates-failed", initialFailures + 1,
+ "a change which could not be replayed must be counted once, not once per attempt");
+ /*
+ * A counter bumped once per attempt rather than once per change goes through the
+ * expected value on its way, so the value has to be seen to stay put rather than
+ * to be reached once.
+ */
+ assertMonitorAttrValueStays(baseDN, "replayed-updates-failed", initialFailures + 1,
+ "a change which could not be replayed must be counted once, not once per attempt");
+ Assertions.assertThat(DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE))
+ .as("the administrator must be told that this replica now diverges")
+ .isGreaterThan(initialAlerts);
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse");
+ domain.setReplayGiveUpDelay(giveUpDelay);
+ }
+ }
+ finally
+ {
+ broker.stop();
+ }
+ }
+
+ /**
+ * Test case for [Issue 889]: every change which can not be replayed must be given up
+ * on, not only the one which fails on its own.
+ * <p>
+ * A backend which is failing fails every change in flight, which is what this test
+ * reproduces with two changes. A count kept for the last failed change only is reset
+ * by each of them in turn, so the give up would never be reached and this replica
+ * would restart its session to the replication server without end.
+ */
+ @Test
+ public void everyChangeWhichCanNotBeReplayedIsGivenUpOn() throws Exception
+ {
+ testSetUp("everyChangeWhichCanNotBeReplayedIsGivenUpOn");
+ logger.error(LocalizableMessage.raw("Starting replication test : everyChangeWhichCanNotBeReplayedIsGivenUpOn"));
+
+ final int serverId = 13;
+ ReplicationBroker broker =
+ openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
+ try
+ {
+ CSNGenerator gen = new CSNGenerator(serverId, 0);
+
+ Entry first = TestCaseUtils.addEntry(
+ "dn: uid=user.889.1," + baseDN,
+ "objectClass: top",
+ "objectClass: person",
+ "objectClass: organizationalPerson",
+ "objectClass: inetOrgPerson",
+ "uid: user.889.1",
+ "cn: Aaccf Amar",
+ "sn: Amar");
+ Entry second = TestCaseUtils.addEntry(
+ "dn: uid=user.889.2," + baseDN,
+ "objectClass: top",
+ "objectClass: person",
+ "objectClass: organizationalPerson",
+ "objectClass: inetOrgPerson",
+ "uid: user.889.2",
+ "cn: Aaccf Amar",
+ "sn: Amar");
+ String firstUuid = getEntry(first.getName(), 1, true).parseAttribute("entryuuid").asString();
+ String secondUuid = getEntry(second.getName(), 1, true).parseAttribute("entryuuid").asString();
+
+ final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
+ final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
+ final long giveUpDelay = domain.getReplayGiveUpDelay();
+ try
+ {
+ // Both are put back by the finally below, so both are set inside the try.
+ domain.setReplayGiveUpDelay(TEST_GIVE_UP_DELAY_IN_MS);
+ ShortCircuitPlugin.registerShortCircuit(
+ OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue());
+
+ final CSN firstCSN = gen.newCSN();
+ final CSN secondCSN = gen.newCSN();
+ broker.publish(new DeleteMsg(first.getName(), firstCSN, firstUuid));
+ broker.publish(new DeleteMsg(second.getName(), secondCSN, secondUuid));
+
+ TestTimer giveUpTimer = new TestTimer.Builder()
+ .maxSleep(120, SECONDS)
+ .sleepTimes(200, MILLISECONDS)
+ .toTimer();
+ giveUpTimer.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ assertTrue(domain.getServerState().cover(firstCSN),
+ "the replica did not give up on the first change it can never replay");
+ assertTrue(domain.getServerState().cover(secondCSN),
+ "the replica did not give up on the second change it can never replay");
+ }
+ });
+ assertMonitorAttrValueEventually(baseDN, "replayed-updates-failed", initialFailures + 2,
+ "both changes must be counted as failed, once each");
+ /*
+ * Two changes counted more than once each climb past +2, and the poll which lands
+ * on it would pass: the value has to be seen to stay put.
+ */
+ assertMonitorAttrValueStays(baseDN, "replayed-updates-failed", initialFailures + 2,
+ "both changes must be counted as failed, once each");
+ assertNotNull(getEntry(first.getName(), 1, true), "the first entry must not have been deleted");
+ assertNotNull(getEntry(second.getName(), 1, true), "the second entry must not have been deleted");
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse");
+ domain.setReplayGiveUpDelay(giveUpDelay);
+ }
+ }
+ finally
+ {
+ broker.stop();
+ }
+ }
+
+ /**
+ * The result codes a replay is retried on rather than skipped: the storage failing to
+ * serve the operation, and a lock which could not be taken (OPENDJ-885) - the ten
+ * in-place attempts only yield to the thread holding it, so a lock held for a while
+ * burns every one of them and the change is as absent from the data as after a storage
+ * failure.
+ */
+ @DataProvider(name = "transientReplayFailures")
+ public Object[][] transientReplayFailures()
+ {
+ return new Object[][] {
+ { ResultCode.UNAVAILABLE, 14, "user.889.3" },
+ { ResultCode.BUSY, 15, "user.889.4" },
+ };
+ }
+
+ /**
+ * Test case for [Issue 889]: a replay which fails on the server itself has the session
+ * restarted and the change delivered again, and a failure which clears in the meantime
+ * has the change applied exactly once, without the change being given up on and without
+ * it being reported as failed.
+ */
+ @Test(dataProvider = "transientReplayFailures")
+ public void transientReplayFailureIsRetriedAndTheChangeApplied(
+ final ResultCode transientFailure, final int serverId, final String uid) throws Exception
+ {
+ testSetUp("transientReplayFailureIsRetriedAndTheChangeApplied." + uid);
+ logger.error(LocalizableMessage.raw(
+ "Starting replication test : transientReplayFailureIsRetriedAndTheChangeApplied "
+ + transientFailure));
+
+ ReplicationBroker broker =
+ openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
+ try
+ {
+ CSNGenerator gen = new CSNGenerator(serverId, 0);
+
+ Entry tmp = TestCaseUtils.addEntry(
+ "dn: uid=" + uid + "," + baseDN,
+ "objectClass: top",
+ "objectClass: person",
+ "objectClass: organizationalPerson",
+ "objectClass: inetOrgPerson",
+ "uid: " + uid,
+ "cn: Aaccf Amar",
+ "sn: Amar");
+ String uuid = getEntry(tmp.getName(), 1, true).parseAttribute("entryuuid").asString();
+
+ final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
+ final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
+ final long initialReplayed = getMonitorAttrValue(baseDN, "replayed-updates-ok");
+ final int initialAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE);
+
+ /*
+ * The backend is unavailable the way it is while a rebuild is performed or while it
+ * is offline (OPENDJ-49), and it stays unavailable for longer than the replay is
+ * retried in place: the change is only applied if the session is restarted and the
+ * replication server delivers it a second time.
+ */
+ try
+ {
+ // Registered inside the try which deregisters it: the plugin is consulted for
+ // every delete in this server, so one left behind fails the tests which follow.
+ ShortCircuitPlugin.registerShortCircuit(OperationType.DELETE, "PreParse",
+ transientFailure.intValue(), IN_PLACE_REPLAY_ATTEMPTS + 2);
+
+ final CSN csn = gen.newCSN();
+ broker.publish(new DeleteMsg(tmp.getName(), csn, uuid));
+
+ assertNull(getEntry(tmp.getName(), 30000, false),
+ "the change was not replayed once the backend served the operation again");
+ Assertions.assertThat(ShortCircuitPlugin.getShortCircuitCount(OperationType.DELETE, "PreParse"))
+ .as("the change must have been delivered again after the session was restarted")
+ .isGreaterThan(IN_PLACE_REPLAY_ATTEMPTS);
+
+ TestTimer timer = new TestTimer.Builder()
+ .maxSleep(30, SECONDS)
+ .sleepTimes(100, MILLISECONDS)
+ .toTimer();
+ timer.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ assertTrue(domain.getServerState().cover(csn),
+ "a change which was replayed must be recorded as replayed");
+ }
+ });
+ assertMonitorAttrValueEventually(baseDN, "replayed-updates-ok", initialReplayed + 1,
+ "the change must be recorded as replayed");
+ /*
+ * A change applied twice - the delivery which failed and the one which took over
+ * from it, the OPENDJ-1115 regression the takeover is there to prevent - takes the
+ * counter through +1 on its way to +2, so the value has to be seen to stay put
+ * rather than to be reached once. It has to be watched for longer than the
+ * session restart which brings that second delivery, too, or the assertion stops
+ * looking before the delivery it is looking for could arrive.
+ */
+ assertMonitorAttrValueStays(baseDN, "replayed-updates-ok", initialReplayed + 1,
+ MONITOR_ATTR_SAMPLES_ACROSS_A_REDELIVERY,
+ "a change which was delivered again must be applied exactly once");
+ assertMonitorAttrValueStays(baseDN, "replayed-updates-failed", initialFailures,
+ MONITOR_ATTR_SAMPLES_ACROSS_A_REDELIVERY,
+ "a change which was replayed after a transient failure must not count as failed");
+ assertEquals(DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE), initialAlerts,
+ "a transient failure must not tell the administrator that this replica diverged");
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse");
+ }
+ }
+ finally
+ {
+ broker.stop();
+ }
+ }
+
+ /**
+ * Test case for [Issue 889]: the result code the server puts on an internal error is
+ * configurable and is not validated as a result code, so it can be set to one conflict
+ * resolution knows how to solve. Such a change is left to conflict resolution, and when
+ * that can not solve it either the change is retried as the storage failure it is -
+ * recording it as replayed after one attempt would be issue #889 again.
+ */
+ @Test
+ public void changeConflictResolutionCanNotSolveOnTheServerErrorCodeIsRetried() throws Exception
+ {
+ testSetUp("changeConflictResolutionCanNotSolveOnTheServerErrorCodeIsRetried");
+ logger.error(LocalizableMessage.raw(
+ "Starting replication test : changeConflictResolutionCanNotSolveOnTheServerErrorCodeIsRetried"));
+
+ final int serverId = 16;
+ ReplicationBroker broker =
+ openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
+ try
+ {
+ CSNGenerator gen = new CSNGenerator(serverId, 0);
+
+ Entry tmp = TestCaseUtils.addEntry(
+ "dn: uid=user.889.5," + baseDN,
+ "objectClass: top",
+ "objectClass: person",
+ "objectClass: organizationalPerson",
+ "objectClass: inetOrgPerson",
+ "uid: user.889.5",
+ "cn: Aaccf Amar",
+ "sn: Amar");
+ String uuid = getEntry(tmp.getName(), 1, true).parseAttribute("entryuuid").asString();
+
+ final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
+ final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
+ domain.resetUnreplayedChangeAlertThrottle();
+ final int initialAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE);
+
+ /*
+ * UNWILLING_TO_PERFORM is one of the codes solveNamingConflict(ModifyDNOperation)
+ * solves, so it must not be treated as a failure of the server before conflict
+ * resolution had its chance - and it is what the storage reports here.
+ */
+ // Put back whatever was configured, not the default: a suite which runs with
+ // another server-error-result-code must not be rewritten by this test.
+ final int previousServerErrorResultCode =
+ getServerContext().getCoreConfigManager().getServerErrorResultCode().intValue();
+ try
+ {
+ /*
+ * Changed inside the try which puts it back: the result code this server reports
+ * an internal error with is server-wide, so one left behind would change which
+ * road every later replay of this suite takes.
+ */
+ setServerErrorResultCode(ResultCode.UNWILLING_TO_PERFORM.intValue());
+ /*
+ * The failure lasts longer than the attempts made in place, so the change is only
+ * applied if it was left out of the ServerState and delivered again rather than
+ * recorded as replayed once conflict resolution reported it could not be solved.
+ */
+ ShortCircuitPlugin.registerShortCircuit(OperationType.DELETE, "PreParse",
+ ResultCode.UNWILLING_TO_PERFORM.intValue(), IN_PLACE_REPLAY_ATTEMPTS + 2);
+
+ final CSN csn = gen.newCSN();
+ broker.publish(new DeleteMsg(tmp.getName(), csn, uuid));
+
+ assertNull(getEntry(tmp.getName(), 120000, false),
+ "the change was skipped rather than retried once the storage served the operation");
+ Assertions.assertThat(ShortCircuitPlugin.getShortCircuitCount(OperationType.DELETE, "PreParse"))
+ .as("the change must have been delivered again rather than recorded as replayed")
+ .isGreaterThan(IN_PLACE_REPLAY_ATTEMPTS);
+ assertMonitorAttrValueStays(baseDN, "replayed-updates-failed", initialFailures,
+ MONITOR_ATTR_SAMPLES_ACROSS_A_REDELIVERY,
+ "a change which was replayed in the end must not be counted as given up on");
+ assertEquals(DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE), initialAlerts,
+ "a change which was replayed in the end must not tell the administrator that this replica diverged");
+ }
+ finally
+ {
+ ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse");
+ setServerErrorResultCode(previousServerErrorResultCode);
+ }
+ }
+ finally
+ {
+ broker.stop();
+ }
+ }
+
+ /**
+ * Test case for [Issue 889]: a change whose message can not be turned into an operation
+ * must not hold this replica's ServerState back for good.
+ * <p>
+ * There is no operation to retry and no delivery which would decode any better, so the
+ * change has to be skipped rather than left listed as the barrier: a change which stays
+ * uncommitted holds back the ServerState - and every change which follows it, from
+ * every master - and the delivery which would replace it is turned down while a replay
+ * thread still owns it, so nothing would ever move it again.
+ */
+ @Test
+ public void aChangeWhichCanNotBeDecodedIsNotLeftHoldingTheServerStateBack() throws Exception
+ {
+ testSetUp("aChangeWhichCanNotBeDecodedIsNotLeftHoldingTheServerStateBack");
+ logger.error(LocalizableMessage.raw(
+ "Starting replication test : aChangeWhichCanNotBeDecodedIsNotLeftHoldingTheServerStateBack"));
+
+ final int serverId = 17;
+ ReplicationBroker broker =
+ openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
+ try
+ {
+ CSNGenerator gen = new CSNGenerator(serverId, 0);
+
+ Entry tmp = TestCaseUtils.addEntry(
+ "dn: uid=user.889.6," + baseDN,
+ "objectClass: top",
+ "objectClass: person",
+ "objectClass: organizationalPerson",
+ "objectClass: inetOrgPerson",
+ "uid: user.889.6",
+ "cn: Aaccf Amar",
+ "sn: Amar");
+ String uuid = getEntry(tmp.getName(), 1, true).parseAttribute("entryuuid").asString();
+
+ final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
+ final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
+ domain.resetUnreplayedChangeAlertThrottle();
+ final int initialAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE);
+
+ final CSN csn = gen.newCSN();
+ broker.publish(undecodableModifyMsg(csn, tmp.getName(), uuid));
+
+ TestTimer timer = new TestTimer.Builder()
+ .maxSleep(60, SECONDS)
+ .sleepTimes(200, MILLISECONDS)
+ .toTimer();
+ timer.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ assertTrue(domain.getServerState().cover(csn),
+ "a change which can never be decoded must not hold the ServerState back");
+ }
+ });
+ assertMonitorAttrValueEventually(baseDN, "replayed-updates-failed", initialFailures + 1,
+ "a change which could not be decoded must be counted as failed");
+ assertMonitorAttrValueStays(baseDN, "replayed-updates-failed", initialFailures + 1,
+ "a change which could not be decoded must be counted once");
+ Assertions.assertThat(DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE))
+ .as("the administrator must be told that this replica now diverges")
+ .isGreaterThan(initialAlerts);
+ }
+ finally
+ {
+ broker.stop();
+ }
+ }
+
+ /**
+ * Builds a ModifyMsg which travels the protocol intact and can not be turned into an
+ * operation.
+ * <p>
+ * The encoded modifications are carried as an opaque byte array and are only read by
+ * {@code createOperation()}, so a message whose modifications are corrupt is decoded,
+ * listed as pending and handed to a replay thread before it fails - which is the point
+ * of this test.
+ *
+ * @param csn the CSN to give the change
+ * @param dn the entry the change is on
+ * @param entryUUID the UUID of that entry
+ * @return a message whose replay can not build an operation
+ * @throws Exception if the message could not be built
+ */
+ private ModifyMsg undecodableModifyMsg(CSN csn, DN dn, String entryUUID) throws Exception
+ {
+ final List<Modification> mods = generatemods("description", "the decoding must fail here");
+ final byte[] bytes =
+ new ModifyMsg(csn, dn, mods, entryUUID).getBytes(ProtocolVersion.getCurrentVersion());
+
+ /*
+ * Break the length of the attribute description inside the encoded modifications, so
+ * that the ASN.1 reader runs past the end of them. The attribute name only appears
+ * there, and the byte before it is the length it is read with.
+ */
+ final int attributeName = indexOf(bytes, "description".getBytes("UTF-8"));
+ assertTrue(attributeName > 0, "the encoded modifications must carry the attribute name");
+ bytes[attributeName - 1] = (byte) 0x7F;
+
+ final ModifyMsg corrupted =
+ (ModifyMsg) ReplicationMsg.generateMsg(bytes, ProtocolVersion.getCurrentVersion());
+ try
+ {
+ corrupted.createOperation(getRootConnection());
+ fail("this test needs a message which can not be turned into an operation");
+ }
+ catch (LDAPException | DecodeException expected)
+ {
+ /*
+ * Which is what the replay of this message hits: the ASN.1 reader reports a
+ * DecodeException, which RawModification.decode() reports as an LDAPException and
+ * ModifyCommonMsg.decodeRawMods() lets through as it is when the over-read lands
+ * between two modifications rather than inside one. The two are named rather than
+ * caught as an Exception so that this test says what the message does, but neither
+ * is what decides its fate: this change is given up on because no operation could
+ * be built from it, and a failure of an operation which was built takes the other
+ * road whatever it was thrown as, which
+ * aChangeWhoseOperationWasBuiltIsNotGivenUpOnWhereItFailed pins.
+ */
+ }
+ return corrupted;
+ }
+
+ /**
+ * Test case for [Issue 889]: a change whose operation was built is delivered again
+ * rather than recorded as replayed when the replay fails before that operation could
+ * tell which change it carries.
+ * <p>
+ * Which of the two roads a failure takes is decided by the operation rather than by
+ * its CSN: a message no operation could be built from will not build one on the next
+ * delivery either, so it is given up on where it is reported, while an operation which
+ * was built may well have reached the backend - so its change is kept out of the
+ * ServerState and asked for again, wherever in the replay the failure happened. The
+ * entry DN of a ModifyMsg which does not parse is that case: it leaves
+ * {@code getEntryDN()} null and the replay throws before the CSN of the operation is
+ * read, so a give-up keyed off that CSN would record a change which never reached the
+ * backend as replayed, which is this issue by another route.
+ */
+ @Test
+ public void aChangeWhoseOperationWasBuiltIsNotGivenUpOnWhereItFailed() throws Exception
+ {
+ testSetUp("aChangeWhoseOperationWasBuiltIsNotGivenUpOnWhereItFailed");
+ logger.error(LocalizableMessage.raw(
+ "Starting replication test : aChangeWhoseOperationWasBuiltIsNotGivenUpOnWhereItFailed"));
+
+ Entry tmp = TestCaseUtils.addEntry(
+ "dn: uid=user.889.7," + baseDN,
+ "objectClass: top",
+ "objectClass: person",
+ "objectClass: organizationalPerson",
+ "objectClass: inetOrgPerson",
+ "uid: user.889.7",
+ "cn: Aaccf Amar",
+ "sn: Amar");
+ final DN dn = tmp.getName();
+ final String uuid = getEntry(dn, 1, true).parseAttribute("entryuuid").asString();
+
+ final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
+ final long initialFailures = getMonitorAttrValue(baseDN, "replayed-updates-failed");
+ domain.resetUnreplayedChangeAlertThrottle();
+ final int initialAlerts = DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE);
+
+ final CSNGenerator gen = new CSNGenerator(18, TimeThread.getTime());
+ final CSN csn = gen.newCSN();
+ final String description = "the replay must fail once the operation is built";
+ final List<Modification> mods = generatemods("description", description);
+
+ domain.processUpdate(new ModifyMsgWithAnUnparseableOperationDN(csn, dn, mods, uuid));
+
+ /*
+ * Long enough to outlast the session restart the failure asks for: a change which is
+ * being asked for again is not in the data at any point of it.
+ */
+ for (int i = 0; i < MONITOR_ATTR_SAMPLES_ACROSS_A_REDELIVERY; i++)
+ {
+ assertFalse(domain.getServerState().cover(csn),
+ "a change whose operation was built must be asked for again, not recorded as replayed");
+ Thread.sleep(200);
+ }
+ assertMonitorAttrValueStays(baseDN, "replayed-updates-failed", initialFailures,
+ "a change which is still to be delivered again must not be counted as given up on");
+ assertEquals(DummyAlertHandler.getAlertCount(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE), initialAlerts,
+ "a change which is still to be delivered again must not be alerted on as a divergence");
+
+ /*
+ * The failed change is the barrier which holds this domain's ServerState back until
+ * it is replayed, and the replication server sending it again is what replays it.
+ * Nothing sends this one - it never travelled a session - so the delivery which takes
+ * over from the one which failed is made here, and it is made until it is taken: a
+ * delivery is dropped rather than queued while the listener thread is down, which it
+ * is for as long as the recovery is restarting the session, and the monitor entry
+ * read above comes back with the broker rather than with the listener. A delivery of
+ * a change a replay thread owns is refused as the duplicate it is, and the ServerState
+ * keeps this from delivering a change which was replayed a second time.
+ */
+ TestTimer timer = new TestTimer.Builder()
+ .maxSleep(60, SECONDS)
+ .sleepTimes(200, MILLISECONDS)
+ .toTimer();
+ timer.repeatUntilSuccess(new CallableVoid()
+ {
+ @Override
+ public void call() throws Exception
+ {
+ if (!domain.getServerState().cover(csn))
+ {
+ domain.processUpdate(new ModifyMsg(csn, dn, mods, uuid));
+ }
+ assertTrue(domain.getServerState().cover(csn),
+ "the change must be recorded as replayed once it has been delivered again");
+ }
+ });
+ checkEntryHasAttributeValue(dn, "description", description, 30,
+ "the change must be applied by the delivery which took over from the failed one");
+ }
+
+ /**
+ * A ModifyMsg whose operation can not tell which change it carries.
+ * <p>
+ * The operation is built - so the replay is past the point where a message is given up
+ * on - and its entry DN does not parse, which is what has
+ * {@code ModifyOperationBasis.getEntryDN()} return null and the replay throw before
+ * {@code OperationContext.getCSN(op)} is reached. Such a message can not travel the
+ * protocol: the DN of a ModifyMsg is decoded on the way in and the operation is built
+ * from its {@code toString()}, so this one is handed to the domain rather than
+ * published.
+ */
+ private static final class ModifyMsgWithAnUnparseableOperationDN extends ModifyMsg
+ {
+ private ModifyMsgWithAnUnparseableOperationDN(
+ CSN csn, DN dn, List<Modification> mods, String entryUUID)
+ {
+ super(csn, dn, mods, entryUUID);
+ }
+
+ @Override
+ public ModifyOperation createOperation(InternalClientConnection connection, DN newDN)
+ {
+ final ModifyOperation op = new ModifyOperationBasis(connection, nextOperationID(),
+ nextMessageID(), null, ByteString.valueOfUtf8("this is not a DN"),
+ new ArrayList<RawModification>());
+ op.setAttachment(OperationContext.SYNCHROCONTEXT,
+ new ModifyContext(getCSN(), getEntryUUID()));
+ return op;
+ }
+ }
+
+ /**
+ * Returns the offset of the first occurrence of {@code needle} in {@code haystack}, or
+ * -1 when it does not occur.
+ */
+ private static int indexOf(byte[] haystack, byte[] needle)
+ {
+ for (int i = 0; i <= haystack.length - needle.length; i++)
+ {
+ int j = 0;
+ while (j < needle.length && haystack[i + j] == needle[j])
+ {
+ j++;
+ }
+ if (j == needle.length)
+ {
+ return i;
+ }
+ }
+ return -1;
+ }
+
+ /**
+ * Sets the result code this server puts on an internal error, the way an administrator
+ * would.
+ *
+ * @param resultCode the numeric result code
+ * @throws Exception if the configuration could not be changed
+ */
+ private void setServerErrorResultCode(int resultCode) throws Exception
+ {
+ assertEquals(TestCaseUtils.applyModifications(true,
+ "dn: cn=config",
+ "changetype: modify",
+ "replace: ds-cfg-server-error-result-code",
+ "ds-cfg-server-error-result-code: " + resultCode), 0,
+ "the server error result code could not be changed");
+ }
+
+ /**
* Enable or disable the receive status of a synchronization provider.
*
* @param syncConfigDN The DN of the synchronization provider configuration
--
Gitblit v1.10.0