mirror of https://github.com/OpenIdentityPlatform/OpenDJ.git
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
