/* * The contents of this file are subject to the terms of the Common Development and * Distribution License (the License). You may not use this file except in compliance with the * License. * * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the * specific language governing permission and limitations under the License. * * When distributing Covered Software, include this CDDL Header Notice in each file and include * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL * Header, with the fields enclosed by brackets [] replaced by your own identifying * information: "Portions Copyright [year] [name of copyright owner]". * * Copyright 2006-2010 Sun Microsystems, Inc. * Portions Copyright 2011-2016 ForgeRock AS. * Portions Copyright 2023-2026 3A Systems, LLC */ package org.opends.server.replication; import static java.util.concurrent.TimeUnit.*; import static org.forgerock.opendj.ldap.ModificationType.*; import static org.forgerock.opendj.ldap.requests.Requests.*; import static org.forgerock.opendj.ldap.schema.CoreSchema.*; import static org.opends.server.TestCaseUtils.*; 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; import java.util.ArrayList; import java.util.List; 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; import org.forgerock.opendj.ldap.requests.ModifyRequest; import org.forgerock.opendj.ldap.schema.AttributeType; import org.opends.server.TestCaseUtils; import org.opends.server.core.AddOperation; 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; import org.testng.annotations.BeforeClass; import org.testng.annotations.DataProvider; import org.testng.annotations.Test; /** * Test synchronization of update operations on the directory server and through * the replication server broker interface. */ @SuppressWarnings("javadoc") public class UpdateOperationTest extends ReplicationTestCase { 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; private Entry user3Entry; private DN user3dn; private String user3UUID; private String baseUUID; private DN user1dn; private String user1entrysecondUUID; private String user1entryUUID; /** A "person" entry. */ private Entry personEntry; private int replServerPort; private String domain1uid; private String domain2uid; private String domain3uid; private DN domain1dn; private DN domain2dn; private DN domain3dn; private Entry domain1; private Entry domain2; private Entry domain3; private int domainSid = 55; private DN baseDN; /** Set up the environment for performing the tests in this Class. */ @BeforeClass @Override public void setUp() throws Exception { super.setUp(); baseDN = DN.valueOf("ou=People," + TEST_ROOT_DN_STRING); // Create necessary backend top level entry TestCaseUtils.addEntry( "dn: " + baseDN, "objectClass: top", "objectClass: organizationalUnit", "entryUUID: 11111111-1111-1111-1111-111111111111"); baseUUID = getEntryUUID(baseDN); replServerPort = TestCaseUtils.findFreePort(); // replication server String replServerLdif = "dn: cn=Replication Server, " + SYNCHRO_PLUGIN_DN + "\n" + "objectClass: top\n" + "objectClass: ds-cfg-replication-server\n" + "cn: Replication Server\n" + "ds-cfg-replication-port: " + replServerPort + "\n" + "ds-cfg-replication-db-directory: UpdateOperationTest\n" + "ds-cfg-replication-server-id: 107\n"; // suffix synchronized String testName = "updateOperationTest"; String synchroServerLdif = "dn: cn=" + testName + ", cn=domains, " + SYNCHRO_PLUGIN_DN + "\n" + "objectClass: top\n" + "objectClass: ds-cfg-replication-domain\n" + "cn: " + testName + "\n" + "ds-cfg-base-dn: " + baseDN + "\n" + "ds-cfg-replication-server: localhost:" + replServerPort + "\n" + "ds-cfg-server-id: "+ domainSid +"\n" + "ds-cfg-receive-status: true\n"; configureReplication(replServerLdif, synchroServerLdif); } private void testSetUp(String tc) throws Exception { personEntry = TestCaseUtils.makeEntry( "dn: uid=user.1." + tc + "," + baseDN, "objectClass: top", "objectClass: person", "objectClass: organizationalPerson", "objectClass: inetOrgPerson", "uid: user.1", "homePhone: 951-245-7634", "description: This is the description for Aaccf Amar.", "st: NC", "mobile: 027-085-0537", "postalAddress: Aaccf Amar$17984 Thirteenth Street $Rockford, NC 85762", "mail: user.1@example.com", "cn: Aaccf Amar", "l: Rockford", "pager: 508-763-4246", "street: 17984 Thirteenth Street", "telephoneNumber: 216-564-6748", "employeeNumber: 1", "sn: Amar", "givenName: Aaccf", "postalCode: 85762", "userPassword: password", "initials: AA"); /* * The 2 entries defined in the following code are used for the naming * conflict resolution test (called namingConflicts) * They must have the same DN but different entryUUID. */ user1entryUUID = "33333333-3333-3333-3333-333333333333"; user1entrysecondUUID = "22222222-2222-2222-2222-222222222222"; user1dn = DN.valueOf("uid=user1" + tc + "," + baseDN); personWithUUIDEntry = TestCaseUtils.makeEntry( "dn: " + user1dn, "objectClass: top", "objectClass: person", "objectClass: organizationalPerson", "objectClass: inetOrgPerson", "uid: user.1", "homePhone: 951-245-7634", "description: This is the description for Aaccf Amar.", "st: NC", "mobile: 027-085-0537", "postalAddress: Aaccf Amar$17984 Thirteenth Street $Rockford, NC 85762", "mail: user.1@example.com", "cn: Aaccf Amar", "l: Rockford", "pager: 508-763-4246", "street: 17984 Thirteenth Street", "telephoneNumber: 216-564-6748", "employeeNumber: 1", "sn: Amar", "givenName: Aaccf", "postalCode: 85762", "userPassword: password", "initials: AA", "entryUUID: " + user1entryUUID + "\n"); personWithSecondUniqueID = TestCaseUtils.makeEntry( "dn: "+ user1dn, "objectClass: top", "objectClass: person", "objectClass: organizationalPerson", "objectClass: inetOrgPerson", "uid: user.1", "homePhone: 951-245-7634", "description: This is the description for Aaccf Amar.", "st: NC", "mobile: 027-085-0537", "postalAddress: Aaccf Amar$17984 Thirteenth Street $Rockford, NC 85762", "mail: user.1@example.com", "cn: Aaccf Amar", "l: Rockford", "pager: 508-763-4246", "street: 17984 Thirteenth Street", "telephoneNumber: 216-564-6748", "employeeNumber: 1", "sn: Amar", "givenName: Aaccf", "postalCode: 85762", "userPassword: password", "initials: AA", "entryUUID: "+ user1entrysecondUUID); user3UUID = "44444444-4444-4444-4444-444444444444"; user3dn = DN.valueOf("uid=user3" + tc + "," + baseDN); user3Entry = TestCaseUtils.makeEntry("dn: "+ user3dn, "objectClass: top", "objectClass: person", "objectClass: organizationalPerson", "objectClass: inetOrgPerson", "uid: user.1", "homePhone: 951-245-7634", "description: This is the description for Aaccf Amar.", "st: NC", "mobile: 027-085-0537", "postalAddress: Aaccf Amar$17984 Thirteenth Street $Rockford, NC 85762", "mail: user.3@example.com", "cn: Aaccf Amar", "l: Rockford", "pager: 508-763-4246", "street: 17984 Thirteenth Street", "telephoneNumber: 216-564-6748", "employeeNumber: 1", "sn: Amar", "givenName: Aaccf", "postalCode: 85762", "userPassword: password", "initials: AA", "entryUUID: " + user3UUID); domain1dn = DN.valueOf("dc=domain1," + baseDN); domain2dn = DN.valueOf("dc=domain2,dc=domain1," + baseDN); domain3dn = DN.valueOf("dc=domain3,dc=domain1," + baseDN); domain1 = TestCaseUtils.makeEntry( "dn:" + domain1dn, "objectClass:domain", "dc:domain1"); domain2 = TestCaseUtils.makeEntry( "dn:" + domain2dn, "objectClass:domain", "dc:domain2"); domain3 = TestCaseUtils.makeEntry( "dn:" + domain3dn, "objectClass:domain", "dc:domain3"); } /** Add an entry in the database. */ private CSN addEntry(Entry entry) throws Exception { AddOperation addOp = connection.processAdd(entry); assertEquals(addOp.getResultCode(), ResultCode.SUCCESS); assertNotNull(getEntry(entry.getName(), 1000, true)); return OperationContext.getCSN(addOp); } /** Delete an entry in the database. */ private void delEntry(DN dn) throws Exception { connection.processDelete(dn); assertNull(getEntry(dn, 1000, false)); } /** * Tests whether the synchronization provider receive status can be disabled * then re-enabled. * FIXME Enable this test when broker suspend/resume receive are implemented. */ @Test(enabled=false) public void toggleReceiveStatus() throws Exception { testSetUp("toggleReceiveStatus"); logger.error(LocalizableMessage.raw("Starting synchronization test : toggleReceiveStatus")); /* * Open a session to the replicationServer using the broker API. * This must use a different serverId to that of the directory server. */ final int serverId = 2; ReplicationBroker broker = openReplicationSession(baseDN, serverId, 100, replServerPort, 1000); try { CSNGenerator gen = new CSNGenerator(serverId, 0); // Disable the directory server receive status. setReceiveStatus(synchroServerEntry.getName(), false); // Create and publish an update message to add an entry. broker.publish(addMsg(gen, personWithUUIDEntry, user1entryUUID, baseUUID)); assertNull(getEntry(personWithUUIDEntry.getName(), 1000, true), "The replication message was replayed while it should not have been: " + "the server receive status was disabled"); // Enable the directory server receive status. setReceiveStatus(synchroServerEntry.getName(), true); broker.publish(addMsg(gen, personWithUUIDEntry, user1entryUUID, baseUUID)); assertNotNull(getEntry(personWithUUIDEntry.getName(), 10000, true), "The replication message was not replayed while it should have been: " + "the server receive status was reenabled"); // Delete the entries to clean the database. broker.publish( new DeleteMsg(personWithUUIDEntry.getName(), gen.newCSN(), user1entryUUID)); assertNull(getEntry(personWithUUIDEntry.getName(), 10000, false), "The DELETE replication message was not replayed"); } finally { broker.stop(); } } private AddMsg addMsg(CSNGenerator gen, Entry entry, String uniqueId, String parentId) { return new AddMsg(gen.newCSN(), entry.getName(), uniqueId, parentId, entry.getObjectClassAttribute(), entry.getAllAttributes(), null); } /** * Tests whether the synchronization provider fails over when it loses * the heartbeat from the replication server. */ @Test public void lostHeartbeatFailover() throws Exception { testSetUp("lostHeartbeatFailover"); logger.error(LocalizableMessage.raw("Starting replication test : lostHeartbeatFailover")); /* * Open a session to the replicationServer using the broker API. * This must use a different serverId to that of the directory server. */ int serverId = 2; ReplicationBroker broker = openReplicationSession(baseDN, serverId, 100, replServerPort, 1000); try { CSNGenerator gen = new CSNGenerator(serverId, 0); // Create and publish an update message to add an entry. broker.publish(addMsg(gen, personWithUUIDEntry, user1entryUUID, baseUUID)); assertNotNull(getEntry(personWithUUIDEntry.getName(), 30000, true), "The ADD replication message was not replayed"); // Send a first modify operation message. List mods = generatemods("telephonenumber", "01 02 45"); ModifyMsg modMsg = new ModifyMsg(gen.newCSN(), personWithUUIDEntry.getName(), mods, user1entryUUID); broker.publish(modMsg); // Check that the modify has been replayed. checkEntryHasAttributeValue(personWithUUIDEntry.getName(), "telephonenumber", "01 02 45", 10, "The first modification was not replayed."); // Simulate loss of heartbeats. HeartbeatThread.setHeartbeatsDisabled(true); Thread.sleep(3000); HeartbeatThread.setHeartbeatsDisabled(false); // Send a second modify operation message. mods = generatemods("description", "Description was changed"); modMsg = new ModifyMsg(gen.newCSN(), personWithUUIDEntry.getName(), mods, user1entryUUID); broker.publish(modMsg); // Check that the modify has been replayed. checkEntryHasAttributeValue(personWithUUIDEntry.getName(), "description", "Description was changed", 10, "The second modification was not replayed."); // Delete the entries to clean the database. broker.publish( new DeleteMsg(personWithUUIDEntry.getName(), gen.newCSN(), user1entryUUID)); assertNull(getEntry(personWithUUIDEntry.getName(), 10000, false), "The DELETE replication message was not replayed"); } finally { broker.stop(); } } /** * Tests the modify conflict resolution code. * In this test, the local server acts both as an LDAP server and * a replicationServer that are inter-connected. * * The test creates an other session to the replicationServer using * directly the ReplicationBroker API. * It then uses this session to simulate conflicts and therefore * test the modify conflict resolution code. */ @Test(enabled=true) public void modifyConflicts() throws Exception { testSetUp("modifyConflicts"); final DN dn1 = DN.valueOf("cn=test1," + baseDN); final AttributeType attrType = getServerContext().getSchema().getAttributeType("displayname"); final AttributeType entryuuidType = getEntryUUIDAttributeType(); String monitorAttr = "resolved-modify-conflicts"; /* * Open a session to the replicationServer using the broker API. * This must use a different serverId to that of the directory server. */ ReplicationBroker broker = openReplicationSession(baseDN, 2, 100, replServerPort, 1000); try { // Add the first test entry. TestCaseUtils.addEntry( "dn: cn=test1," + baseDN, "displayname: Test1", "objectClass: top", "objectClass: person", "objectClass: organizationalPerson", "objectClass: inetOrgPerson", "cn: test1", "sn: test"); // Read the entry back to get its UUID. Entry entry = DirectoryServer.getEntry(dn1); List attrs = entry.getAllAttributes(entryuuidType); String entryuuid = attrs.get(0).iterator().next().toString(); // A change on a first server. long changeTime = TimeThread.getTime(); CSN t1 = new CSN(changeTime, 0, 3); // A change on a second server. changeTime++; CSN t2 = new CSN(changeTime, 0, 4); // Simulate the ordering t2:replace:B followed by t1:add:A that updateMonitorCount(baseDN, monitorAttr); // Replay a replace of a value B at time t2 on a second server. Attribute attr = Attributes.create(attrType, "B"); List mods = newArrayList(new Modification(ModificationType.REPLACE, attr)); ModifyMsg modMsg = new ModifyMsg(t2, dn1, mods, entryuuid); broker.publish(modMsg); Thread.sleep(2000); // Replay an add of a value A at time t1 on a first server. attr = Attributes.create(attrType, "A"); mods = newArrayList(new Modification(ModificationType.ADD, attr)); modMsg = new ModifyMsg(t1, dn1, mods, entryuuid); broker.publish(modMsg); Thread.sleep(2000); // Read the entry to see how the conflict was resolved. entry = DirectoryServer.getEntry(dn1); attrs = entry.getAllAttributes(attrType); String attrValue1 = attrs.get(0).iterator().next().toString(); // the value should be the last (time t2) value added assertEquals(attrValue1, "B"); assertEquals(getMonitorDelta(), 1); // Simulate the ordering t2:delete:displayname followed by // t1:replace:displayname // A change on a first server. changeTime++; t1 = new CSN(changeTime, 0, 3); // A change on a second server. changeTime++; t2 = new CSN(changeTime, 0, 4); // Simulate the ordering t2:delete:displayname followed by t1:replace:A updateMonitorCount(baseDN, monitorAttr); // Replay an delete of attribute displayname at time t2 on a second server. attr = Attributes.empty(attrType); mods = newArrayList(new Modification(ModificationType.DELETE, attr)); modMsg = new ModifyMsg(t2, dn1, mods, entryuuid); broker.publish(modMsg); Thread.sleep(2000); // Replay a replace of a value A at time t1 on a first server. attr = Attributes.create(attrType, "A"); mods = newArrayList(new Modification(ModificationType.REPLACE, attr)); modMsg = new ModifyMsg(t1, dn1, mods, entryuuid); broker.publish(modMsg); Thread.sleep(2000); // Read the entry to see how the conflict was resolved. entry = DirectoryServer.getEntry(dn1); attrs = entry.getAllAttributes(attrType); // there should not be a value (delete at time t2) Assertions.assertThat(attrs).isEmpty(); assertEquals(getMonitorDelta(), 1); } finally { broker.stop(); } } /** * Waits for the replay thread to update the monitored counter: reading the * monitor immediately after publishing a message races with the replay. */ private void waitForMonitorDelta(final long expectedDelta) throws Exception { final long[] total = { 0 }; TestTimer timer = new TestTimer.Builder() .maxSleep(30, SECONDS) .sleepTimes(100, MILLISECONDS) .toTimer(); timer.repeatUntilSuccess(new CallableVoid() { @Override public void call() throws Exception { total[0] += getMonitorDelta(); assertEquals(total[0], expectedDelta); } }); } /** * Tests the naming conflict resolution code. * In this test, the local server act both as an LDAP server and * a replicationServer that are inter-connected. * * The test creates an other session to the replicationServer using * directly the ReplicationBroker API. * It then uses this session to simulate conflicts and therefore * test the naming conflict resolution code. */ @Test(enabled=true) public void namingConflicts() throws Exception { testSetUp("namingConflicts"); logger.error(LocalizableMessage.raw("Starting replication test : namingConflicts")); String resolvedMonitorAttr = "resolved-naming-conflicts"; String unresolvedMonitorAttr = "unresolved-naming-conflicts"; /* * Open a session to the replicationServer using the ReplicationServer broker API. * This must use a serverId different from the LDAP server ID */ final int serverId = 2; ReplicationBroker broker = openReplicationSession(baseDN, serverId, 100, replServerPort, 1000); try { CSNGenerator gen = new CSNGenerator(serverId, 0); /* * Test that the conflict resolution code is able to find entries * that have been renamed by an other master. * To simulate this, create an entry with a given UUID and a given DN * then send a modify operation using another DN but the same UUID. * Finally check that the modify operation has been applied. */ // create the entry with a given DN broker.publish(addMsg(gen, personWithUUIDEntry, user1entryUUID, baseUUID)); // Check that the entry has been created in the local DS. assertNotNull(getEntry(personWithUUIDEntry.getName(), 10000, true), "The send ADD replication message was not applied"); // send a modify operation with the correct unique ID but another DN List mods = generatemods("telephonenumber", "01 02 45"); ModifyMsg modMsg = new ModifyMsg(gen.newCSN(), DN.valueOf("cn=something," + baseDN), mods, user1entryUUID); updateMonitorCount(baseDN, resolvedMonitorAttr); int alertCount = DummyAlertHandler.getAlertCount(); broker.publish(modMsg); // check that the modify has been applied as if the entry had been renamed. checkEntryHasAttributeValue(personWithUUIDEntry.getName(), "telephonenumber", "01 02 45", 10, "The modification has not been correctly replayed."); assertEquals(getMonitorDelta(), 1); assertConflictAutomaticallyResolved(alertCount); /* * Test that modify conflict resolution is able to detect that * because there is a conflict between a MODIFYDN and a MODIFY, * when a MODIFY is replayed the attribute that is being modified is * now the RDN of the entry and therefore should not be deleted. */ // send a modify operation attempting to replace the RDN entry // with a new value mods = generatemods("uid", "AnotherUid"); modMsg = new ModifyMsg(gen.newCSN(), personWithUUIDEntry.getName(), mods, user1entryUUID); updateMonitorCount(baseDN, resolvedMonitorAttr); alertCount = DummyAlertHandler.getAlertCount(); broker.publish(modMsg); // check that the modify has been applied. checkEntryHasAttributeValue(personWithUUIDEntry.getName(), "uid", "AnotherUid", 10, "The modification has not been correctly replayed."); assertEquals(getMonitorDelta(), 1); /* * Test that the conflict resolution code is able to detect * that an entry has been renamed and that a new entry has * been created with the same DN but another entry UUID * To simulate this, create and entry with a given UUID and a given DN * then send a modify operation using the same DN but another UUID. * Finally check that the modify operation has not been applied to the * entry with the given DN. */ // create the entry with a given DN and unique ID broker.publish(addMsg(gen, personWithUUIDEntry, user1entryUUID, baseUUID)); // Check that the entry has been created in the local DS. assertNotNull(getEntry(personWithUUIDEntry.getName(), 10000, true), "The ADD replication message was not applied"); // send a modify operation with a wrong unique ID but the same DN mods = generatemods("telephonenumber", "02 01 03 05"); modMsg = new ModifyMsg(gen.newCSN(), user1dn, mods, "10000000-9abc-def0-1234-1234567890ab"); updateMonitorCount(baseDN, resolvedMonitorAttr); alertCount = DummyAlertHandler.getAlertCount(); broker.publish(modMsg); // check that the modify has not been applied Thread.sleep(2000); checkEntryHasNoSuchAttributeValue(personWithUUIDEntry.getName(), "telephonenumber", "02 01 03 05", 10, "The modification has been replayed while it should not."); assertEquals(getMonitorDelta(), 1); assertConflictAutomaticallyResolved(alertCount); /* * Test that the conflict resolution code is able to find entries * that have been renamed by an other master. * To simulate this, send a delete operation using another DN but * the same UUID has the entry that has been used in the tests above. * Finally check that the delete operation has been applied. */ // send a delete operation with a wrong dn but the unique ID of the entry // used above updateMonitorCount(baseDN, resolvedMonitorAttr); alertCount = DummyAlertHandler.getAlertCount(); DN delDN = DN.valueOf("cn=anotherdn," + baseDN); broker.publish(new DeleteMsg(delDN, gen.newCSN(), user1entryUUID)); // check that the delete operation has been applied assertNull(getEntry(personWithUUIDEntry.getName(), 10000, false), "The DELETE replication message was not replayed"); assertEquals(getMonitorDelta(), 1); assertConflictAutomaticallyResolved(alertCount); /* * Test that two adds with the same DN but a different unique ID result * cause a conflict and result in the second entry to be renamed. */ // create an entry with a given DN and unique ID broker.publish(addMsg(gen, personWithUUIDEntry, user1entryUUID, baseUUID)); // Check that the entry has been created in the local DS. assertNotNull(getEntry(personWithUUIDEntry.getName(), 10000, true), "The ADD replication message was not applied"); // create an entry with the same DN and another unique ID updateMonitorCount(baseDN, unresolvedMonitorAttr); alertCount = DummyAlertHandler.getAlertCount(); broker.publish(addMsg(gen, personWithSecondUniqueID, user1entrysecondUUID, baseUUID)); // Check that the entry has been renamed and created in the local DS. DN dn2 = DN.valueOf("entryuuid=" + user1entrysecondUUID + " + " + user1dn); final Entry entryAfterAdd = getEntry(dn2, 10000, true); assertNotNull(entryAfterAdd, "The ADD replication message was not applied"); assertEquals(getMonitorDelta(), 1); assertConflictAttributeExists(entryAfterAdd); assertNewAlertsGenerated(alertCount, 1); // delete the entries to clean the database. broker.publish( new DeleteMsg(personWithUUIDEntry.getName(), gen.newCSN(), user1entryUUID)); broker.publish( new DeleteMsg(personWithSecondUniqueID.getName(), gen.newCSN(), user1entrysecondUUID)); assertNull(getEntry(personWithUUIDEntry.getName(), 10000, false), "The DELETE replication message was not replayed"); // The second entry was created with the same DN as the first one, so waiting // on that DN again returns as soon as the first delete is replayed. Wait on // the DN the naming conflict renamed it to instead: the second delete then // has resolved its conflict, and counted it, before the monitor is read below. assertNull(getEntry(dn2, 10000, false), "The DELETE replication message was not replayed"); /* * Check that and added entry is correctly added below it's * parent entry when this parent entry has been renamed. * * Simulate this by trying to add an entry below a DN that does not * exist but with a parent ID that exist. */ String addDN = "uid=new person,o=nothere,o=below," + baseDN; AddMsg addMsg = new AddMsg(gen.newCSN(), DN.valueOf(addDN), user1entryUUID, baseUUID, personWithUUIDEntry.getObjectClassAttribute(), personWithUUIDEntry.getAllAttributes(), null); updateMonitorCount(baseDN, resolvedMonitorAttr); alertCount = DummyAlertHandler.getAlertCount(); broker.publish(addMsg); // Check that the entry has been created in the local DS. DN newPersonDN = DN.valueOf("uid=new person," + baseDN); assertNotNull(getEntry(newPersonDN, 10000, true), "The ADD replication message was not applied"); assertEquals(getMonitorDelta(), 1); assertConflictAutomaticallyResolved(alertCount); /* * Check that when replaying delete the naming conflict code * verify that the unique ID op the replayed operation is * the same as the unique ID of the entry with the given DN * * To achieve this send a delete operation with a correct DN * but a wrong unique ID. */ updateMonitorCount(baseDN, resolvedMonitorAttr); alertCount = DummyAlertHandler.getAlertCount(); broker.publish( new DeleteMsg(newPersonDN, gen.newCSN(), "11111111-9abc-def0-1234-1234567890ab")); // check that the delete operation has not been applied assertNotNull(getEntry(newPersonDN, 10000, true), "The DELETE replication message was replayed when it should not"); // The entry already exists, so the getEntry() call above does not wait // for the replay: poll the monitor until the replay thread has resolved // the conflict. waitForMonitorDelta(1); assertConflictAutomaticallyResolved(alertCount); /* * Check that when replaying modify dn operations, the conflict * resolution code is able to find the new DN of the parent entry * if it has been renamed on another master. * * To simulate this try to rename an entry below an entry that does * not exist but giving the unique ID of an existing entry. */ ModifyDNMsg modDnMsg = new ModifyDNMsg( newPersonDN, gen.newCSN(), user1entryUUID, baseUUID, false, "uid=wrong, " + baseDN, "uid=newrdn"); updateMonitorCount(baseDN, resolvedMonitorAttr); alertCount = DummyAlertHandler.getAlertCount(); broker.publish(modDnMsg); // check that the operation has been correctly relayed assertNotNull(getEntry(DN.valueOf("uid=newrdn," + baseDN), 10000, true), "The modify dn was not or badly replayed"); assertEquals(getMonitorDelta(), 1); assertConflictAutomaticallyResolved(alertCount); /* same test but by giving a bad entry DN */ DN modDN = DN.valueOf("uid=wrong," + baseDN); modDnMsg = new ModifyDNMsg(modDN, gen.newCSN(), user1entryUUID, null, false, null, "uid=reallynewrdn"); updateMonitorCount(baseDN, resolvedMonitorAttr); alertCount = DummyAlertHandler.getAlertCount(); broker.publish(modDnMsg); DN reallyNewDN = DN.valueOf("uid=reallynewrdn," + baseDN); // check that the operation has been correctly relayed assertNotNull(getEntry(reallyNewDN, 10000, true), "The modify dn was not or badly replayed"); assertEquals(getMonitorDelta(), 1); assertConflictAutomaticallyResolved(alertCount); /* * Check that conflicting entries are renamed when a * modifyDN is done with the same DN as an entry added on another server. */ // add a second entry broker.publish(addMsg(gen, personWithSecondUniqueID, user1entrysecondUUID, baseUUID)); // check that the second entry has been added assertNotNull(getEntry(user1dn, 10000, true), "The add operation was not replayed"); // try to rename the first entry modDnMsg = new ModifyDNMsg(user1dn, gen.newCSN(), user1entrysecondUUID, baseUUID, false, baseDN.toString(), "uid=reallynewrdn"); updateMonitorCount(baseDN, unresolvedMonitorAttr); alertCount = DummyAlertHandler.getAlertCount(); broker.publish(modDnMsg); // check that the second entry has been renamed DN dn = DN.valueOf("entryUUID = " + user1entrysecondUUID + "+uid=reallynewrdn," + baseDN); final Entry entryAfterModDN = getEntry(dn, 10000, true); assertNotNull(entryAfterModDN, "The modifyDN was not or incorrectly replayed"); assertEquals(getMonitorDelta(), 1); assertConflictAttributeExists(entryAfterModDN); assertNewAlertsGenerated(alertCount, 1); // delete the entries to clean the database DN delDN2 = DN.valueOf( "entryUUID = " + user1entrysecondUUID + "+" + user1dn.rdn() + "," + baseDN); broker.publish(new DeleteMsg(delDN2, gen.newCSN(), user1entrysecondUUID)); assertNull(getEntry(delDN2, 10000, false), "The DELETE replication message was not replayed"); broker.publish(new DeleteMsg(reallyNewDN, gen.newCSN(), user1entryUUID)); assertNull(getEntry(reallyNewDN, 10000, false), "The DELETE replication message was not replayed"); /* * When replaying add operations it is possible that the parent entry has * been renamed before and that another entry have taken the former dn of * the parent entry. In such case the replication replay code should * detect that the parent has been renamed and should add the entry below * the new dn of the parent (thus changing the original dn with which the * entry had been created) * * Steps * - create parent entry 1 with baseDn1 * - create Add Msg for user1 with parent entry 1 UUID * - MODDN parent entry 1 to baseDn2 in the LDAP server * - add new parent entry 2 with baseDn1 * - publish msg * - check that the Dn has been changed to baseDn2 in the msg received */ DN baseDN1 = DN.valueOf("ou=baseDn1," + baseDN); DN baseDN2 = DN.valueOf("ou=baseDn2," + baseDN); // - create parent entry 1 with baseDn1 connection.processAdd(TestCaseUtils.makeEntry( "dn: " + baseDN1, "objectClass: top", "objectClass: organizationalUnit", "entryUUID: 55555555-5555-5555-5555-555555555555")); assertNotNull(getEntry(baseDN1, 10000, true), "Entry not added: " + baseDN1); // - create Add Msg for user1 with parent entry 1 UUID DN newPersonDN2 = DN.valueOf("uid=new person," + baseDN1); addMsg = new AddMsg(gen.newCSN(), newPersonDN2, user1entryUUID, getEntryUUID(baseDN1), personWithUUIDEntry.getObjectClassAttribute(), personWithUUIDEntry.getAllAttributes(), null); // - MODDN parent entry 1 to baseDn2 in the LDAP server ModifyDNRequest modifyDNRequest = newModifyDNRequest(baseDN1.toString(), "ou=baseDn2") .setDeleteOldRDN(true) .setNewSuperior(baseDN.toString()); connection.processModifyDN(modifyDNRequest); assertNotNull(getEntry(baseDN2, 10000, true), "Entry not moved from " + baseDN1 + " to " + baseDN2); // - add new parent entry 2 with baseDn1 connection.processAdd(TestCaseUtils.makeEntry( "dn: " + baseDN1, "objectClass: top", "objectClass: organizationalUnit", "entryUUID: 66666666-6666-6666-6666-666666666666")); // - publish msg updateMonitorCount(baseDN, resolvedMonitorAttr); alertCount = DummyAlertHandler.getAlertCount(); broker.publish(addMsg); // - check that the DN has been changed to baseDn2 assertNull(getEntry(newPersonDN2, 10000, false), "The ADD replication message was applied under " + baseDN1); assertNotNull(getEntry(DN.valueOf("uid=new person," + baseDN2), 10000, true), "The ADD replication message was NOT applied under " + baseDN2); assertEquals(getMonitorDelta(), 1); assertConflictAutomaticallyResolved(alertCount); // Check that when a delete is conflicting with Add of some entries // below the deleted entries, the child entry that have been added // before the deleted is replayed gets renamed correctly. // add domain1 entry with 2 children : domain2 and domain3 addEntry(domain1); CSN olderCSN = gen.newCSN(); Thread.sleep(1000); domain1uid = getEntryUUID(domain1dn); addEntry(domain2); domain2uid = getEntryUUID(domain2dn); addEntry(domain3); domain3uid = getEntryUUID(domain3dn); DN conflictDomain2dn = DN.valueOf( "entryUUID = " + domain2uid + "+dc=domain2," + baseDN); DN conflictDomain3dn = DN.valueOf( "entryUUID = " + domain3uid + "+dc=domain3," + baseDN); updateMonitorCount(baseDN, unresolvedMonitorAttr); alertCount = DummyAlertHandler.getAlertCount(); // delete domain1 broker.publish(new DeleteMsg(domain1dn, olderCSN, domain1uid)); // check that the domain1 has correctly been deleted assertNull(getEntry(domain1dn, 10000, false), "The DELETE replication message was not replayed"); // check that domain2 and domain3 have been renamed assertNotNull(getEntry(conflictDomain2dn, 1000, true), "The conflicting entries were not created"); assertNotNull(getEntry(conflictDomain3dn, 1000, true), "The conflicting entries were not created"); // check that the 2 conflicting entries have been correctly marked checkEntryHasAttributeValue(conflictDomain2dn, DS_SYNC_CONFLICT, domain2dn.toString(), 1, null); checkEntryHasAttributeValue(conflictDomain3dn, DS_SYNC_CONFLICT, domain3dn.toString(), 1, null); // check that unresolved conflict count has been incremented assertEquals(getMonitorDelta(), 1); assertNewAlertsGenerated(alertCount, 2); // delete the resulting entries for the next test delEntry(conflictDomain2dn); delEntry(conflictDomain3dn); // Check that when a delete is replayed over an entry which has child // those child are also deleted // add domain1 entry with 2 children : domain2 and domain3 addEntry(domain1); domain1uid = getEntryUUID(domain1dn); addEntry(domain2); domain2uid = getEntryUUID(domain2dn); CSN addCSN = addEntry(domain3); gen.adjust(addCSN); domain3uid = getEntryUUID(domain3dn); updateMonitorCount(baseDN, unresolvedMonitorAttr); alertCount = DummyAlertHandler.getAlertCount(); // delete domain1 broker.publish(new DeleteMsg(domain1dn, gen.newCSN(), domain1uid)); // check that the domain1 has correctly been deleted assertNull(getEntry(domain1dn, 10000, false), "The DELETE replication message was not replayed"); // check that domain2 and domain3 have been renamed as conflicting assertTrue(DirectoryServer.entryExists(conflictDomain2dn), "The conflicting entry exist for domain2" + conflictDomain2dn); assertTrue(DirectoryServer.entryExists(conflictDomain3dn), "The conflicting entry exist for domain3" + conflictDomain3dn); // check that unresolved conflict count has been incremented assertEquals(getMonitorDelta(), 1); delEntry(conflictDomain2dn); delEntry(conflictDomain3dn); // Check that when an entry is added on one master below an entry // that is currently deleted on another master, the replay of the // add on the second master cause the added entry to be renamed broker.publish(addMsg(gen, domain2, domain2uid, domain1uid)); // check that conflict entry was created assertNotNull(getEntry(conflictDomain2dn, 1000, true), "The conflicting entries were not created"); // check that the entry have been correctly marked as conflicting. checkEntryHasAttributeValue(conflictDomain2dn, DS_SYNC_CONFLICT, domain2dn.toString(), 1, null); // check that unresolved conflict count has been incremented assertEquals(getMonitorDelta(), 1); // Check that when an entry is deleted on a first master and // renamed on a second master and the rename is replayed last // this is correctly detected as a resolved conflict. // To simulate this simply try a modifyDN on a non existent uid. modDnMsg = new ModifyDNMsg( newPersonDN, gen.newCSN(), "33343333-3533-3633-3373-333333833333", baseUUID, false, "uid=wrong, " + baseDN, "uid=newrdn"); updateMonitorCount(baseDN, resolvedMonitorAttr); alertCount = DummyAlertHandler.getAlertCount(); broker.publish(modDnMsg); // unfortunately it is difficult to check that the operation // did not do anything. // The only thing we can check is that resolved naming conflict counter // has correctly been incremented. waitForNonZeroMonitorDelta(); assertConflictAutomaticallyResolved(alertCount); /* * Check that a conflict is detected when an entry is moved below an entry that does not exist. */ updateMonitorCount(baseDN, unresolvedMonitorAttr); alertCount = DummyAlertHandler.getAlertCount(); modDnMsg = new ModifyDNMsg( newPersonDN, gen.newCSN(), "33333333-3333-3333-3333-333333333333", "12343333-3533-3633-3333-333333833333" , false, "uid=wrong, " + baseDN, "uid=newrdn"); broker.publish(modDnMsg); waitForNonZeroMonitorDelta(); // check that the entry have been correctly marked as conflicting. checkEntryHasAttributeValue( DN.valueOf("uid=new person," + baseDN2), DS_SYNC_CONFLICT, "uid=newrdn," + baseDN2, 1, null); } finally { broker.stop(); } } private void waitForNonZeroMonitorDelta() throws Exception, InterruptedException { // if the monitor counter did not get incremented after 200sec // then something got wrong. TestTimer timer = new TestTimer.Builder() .maxSleep(200, SECONDS) .sleepTimes(100, MILLISECONDS) .toTimer(); timer.repeatUntilSuccess(new CallableVoid() { @Override public void call() throws Exception { assertNotEquals(getMonitorDelta() , 0); } }); } /** * Check that there was an administrative alert generated because the conflict * has not been automatically resolved. */ private void assertNewAlertsGenerated(int oldAlertCount, int expectedNbNewAlerts) { assertEquals(DummyAlertHandler.getAlertCount(), oldAlertCount + expectedNbNewAlerts, "An alert was not generated when resolving conflicts"); } /** * Check that there was no administrative alert generated because the conflict * has been automatically resolved. */ private void assertConflictAutomaticallyResolved(int expectedAlertCount) { assertEquals(DummyAlertHandler.getAlertCount(), expectedAlertCount, "Expected no new alert to be generated when automatically resolving conflicts"); } /** * Check that the given entry does contain the attribute that mark the * entry as conflicting. * * @param entry The entry that needs to be asserted. * @return A boolean indicating if the entry is correctly marked. */ private boolean assertConflictAttributeExists(Entry entry) { return !isEmpty(entry.getAllAttributes("ds-sync-confict")); } @DataProvider(name="assured") public Object[][] getAssuredFlag() { return new Object[][] { { false }, {true} }; } private void cleanupTest() throws Exception { classCleanUp(); setUp(); } /** Tests done using directly the ReplicationBroker interface. */ @Test(enabled=true, dataProvider="assured") public void updateOperations(boolean assured) throws Exception { testSetUp("updateOperations"); logger.error(LocalizableMessage.raw("Starting replication test : updateOperations " + assured)); // Cleanup from previous run cleanupTest(); final int serverId = 27; ReplicationBroker broker = openReplicationSession(baseDN, serverId, 100, replServerPort, 2000); try { CSNGenerator gen = new CSNGenerator(serverId, 0); /* * Test that operations done on this server are sent to the * replicationServer and forwarded to our replicationServer broker session. */ // Create an Entry (add operation) Entry tmp = personEntry.duplicate(false); AddOperation addOp = connection.processAdd(tmp); assertTrue(DirectoryServer.entryExists(personEntry.getName()), "The Add Entry operation failed"); assertEquals(addOp.getResultCode(), ResultCode.SUCCESS); assertClientReceivesExpectedMsg(broker, AddMsg.class, personEntry.getName()); // Modify the entry connection.processModify(modifyRequest(personEntry.getName(), REPLACE, "telephonenumber", "01 02 45")); assertClientReceivesExpectedMsg(broker, ModifyMsg.class, personEntry.getName()); // Modify the entry DN ModifyDNRequest modifyDNRequest = newModifyDNRequest(personEntry.getName().toString(), "uid=new person") .setDeleteOldRDN(true) .setNewSuperior(baseDN.toString()); connection.processModifyDN(modifyDNRequest); DN newDN = DN.valueOf("uid= new person," + baseDN); assertTrue(DirectoryServer.entryExists(newDN), "The MOD_DN operation didn't create the new person entry"); assertFalse(DirectoryServer.entryExists(personEntry.getName()), "The MOD_DN operation didn't delete the old person entry"); assertClientReceivesExpectedMsg(broker, ModifyDNMsg.class, personEntry.getName()); // Delete the entry connection.processDelete(newDN); assertFalse(DirectoryServer.entryExists(newDN), "Unable to delete the new person Entry"); assertClientReceivesExpectedMsg(broker, DeleteMsg.class, newDN); /* * Now check that when we send message to the ReplicationServer * and that they are received and correctly replayed by the server. * * Start by testing the Add message reception */ AddMsg addMsg = addMsg(gen, personWithUUIDEntry, user1entryUUID, baseUUID); addMsg.setAssured(assured); broker.publish(addMsg); /* * Check that the entry has been created in the local DS. */ Entry resultEntry = getEntry(personWithUUIDEntry.getName(), 10000, true); assertNotNull(resultEntry, "The send ADD replication message was not applied for "+personWithUUIDEntry.getName()); /* * Test the reception of Modify Msg */ ModifyMsg modMsg = new ModifyMsg(gen.newCSN(), personWithUUIDEntry.getName(), generatemods("telephonenumber", "01 02 45"), user1entryUUID); modMsg.setAssured(assured); broker.publish(modMsg); checkEntryHasAttributeValue(personWithUUIDEntry.getName(), "telephonenumber", "01 02 45", 10, "The modification has not been correctly replayed."); // Test that replication is able to add attribute that do // not exist in the schema. List invalidMods = generatemods("badattribute", "value"); modMsg = new ModifyMsg(gen.newCSN(), personWithUUIDEntry.getName(), invalidMods, user1entryUUID); modMsg.setAssured(assured); broker.publish(modMsg); checkEntryHasAttributeValue(personWithUUIDEntry.getName(), "badattribute", "value", 10, "The modification has not been correctly replayed."); /* * Test the Reception of Modify Dn Msg */ ModifyDNMsg moddnMsg = new ModifyDNMsg(personWithUUIDEntry.getName(), gen.newCSN(), user1entryUUID, null, true, null, "uid= new person"); moddnMsg.setAssured(assured); broker.publish(moddnMsg); assertNotNull(getEntry(newDN, 10000, true), "The modify DN replication message was not applied"); /* * Test the Reception of Delete Msg */ DeleteMsg delMsg = new DeleteMsg(newDN, gen.newCSN(), user1entryUUID); delMsg.setAssured(assured); broker.publish(delMsg); assertNull(getEntry(newDN, 10000, false), "The DELETE replication message was not replayed"); } finally { broker.stop(); } } private void assertClientReceivesExpectedMsg(ReplicationBroker broker, Class type, DN expectedDN) throws Exception { final ReplicationMsg msg = broker.receive(); Assertions.assertThat(msg).isInstanceOf(type); final LDAPUpdateMsg opMsg = (LDAPUpdateMsg) msg; final OperationType opType = getOperationType(opMsg); final Operation receivedOp = opMsg.createOperation(connection); assertEquals(receivedOp.getOperationType(), opType, "The received replication message is not of corrct type. msg : " + opMsg); assertEquals(opMsg.getDN(), expectedDN, "The received " + opType + " replication message is not for the expected DN : " + opMsg); } private OperationType getOperationType(LDAPUpdateMsg msg) { if (msg instanceof AddMsg) { return OperationType.ADD; } else if (msg instanceof DeleteMsg) { return OperationType.DELETE; } else if (msg instanceof ModifyMsg) { return OperationType.MODIFY; } else if (msg instanceof ModifyDNMsg) { return OperationType.MODIFY_DN; } throw new RuntimeException("Unhandled type: " + msg.getClass()); } /** Test case for [Issue 635] NullPointerException when trying to access non existing entry. */ @Test(enabled=true) public void deleteNoSuchObject() throws Exception { testSetUp("deleteNoSuchObject"); logger.error(LocalizableMessage.raw("Starting replication test : deleteNoSuchObject")); DeleteOperation op = connection.processDelete("cn=No Such Object," + baseDN); assertEquals(op.getResultCode(), ResultCode.NO_SUCH_OBJECT); } /** Test case for [Issue 798] break infinite loop when problems with naming resolution conflict. */ @Test(enabled=true) public void infiniteReplayLoop() throws Exception { testSetUp("infiniteReplayLoop"); logger.error(LocalizableMessage.raw("Starting replication test : infiniteReplayLoop")); int serverId = 11; ReplicationBroker broker = openReplicationSession(baseDN, serverId, 100, replServerPort, 1000); try { CSNGenerator gen = new CSNGenerator(serverId, 0); // Create a test entry. Entry tmp = TestCaseUtils.addEntry( "dn: uid=user.2," + baseDN, "objectClass: top", "objectClass: person", "objectClass: organizationalPerson", "objectClass: inetOrgPerson", "uid: user.2", "homePhone: 951-245-7634", "description: This is the description for Aaccf Amar.", "st: NC", "mobile: 027-085-0537", "postalAddress: Aaccf Amar$17984 Thirteenth Street $Rockford, NC 85762", "mail: user.1@example.com", "cn: Aaccf Amar", "l: Rockford", "pager: 508-763-4246", "street: 17984 Thirteenth Street", "telephoneNumber: 216-564-6748", "employeeNumber: 1", "sn: Amar", "givenName: Aaccf", "postalCode: 85762", "userPassword: password", "initials: AA"); final long initialCount = getMonitorAttrValue(baseDN, "replayed-updates"); // Get the UUID of the test entry. Entry resultEntry = getEntry(tmp.getName(), 1, true); String uuid = resultEntry.parseAttribute("entryuuid").asString(); // Register a short circuit that will fake a no-such-object result code // on a delete. This will cause a replication replay loop. ShortCircuitPlugin.registerShortCircuit(OperationType.DELETE, "PreParse", 32); try { // Publish a delete message for this test entry. broker.publish(new DeleteMsg(tmp.getName(), gen.newCSN(), uuid)); // Wait for the operation to be replayed. TestTimer timer = new TestTimer.Builder() .maxSleep(5, SECONDS) .sleepTimes(100, MILLISECONDS) .toTimer(); timer.repeatUntilSuccess(new CallableVoid() { @Override public void call() throws Exception { assertNotEquals(getMonitorAttrValue(baseDN, "replayed-updates"), initialCount); } }); } finally { ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse"); } } finally { broker.stop(); } } /** * 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. *

* 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. *

* 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. *

* 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 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. *

* 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 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. *

* 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 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()); 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 * entry. * @param enable Specifies whether the receive status should be enabled * or disabled. */ private static void setReceiveStatus(DN syncConfigDN, boolean enable) { String attrValue = enable ? "TRUE" : "FALSE"; ModifyRequest request = modifyRequest(syncConfigDN, REPLACE, "ds-cfg-receive-status", attrValue); ModifyOperation modOp = getRootConnection().processModify(request); assertEquals(modOp.getResultCode(), ResultCode.SUCCESS, "Cannot set receive status"); } /** * Test that the ReplicationDomain (plugin inside LDAP server) adjust * its internal CSN generator to the last CSN received. Steps: * - create a domain with the current date in the CSN generator * - make it receive an update with a CSN in the future * - do a local operation replicated on that domain * - check that the update generated for that operation has a CSN in the future. */ @Test(enabled=true) public void csnGeneratorAdjust() throws Exception { testSetUp("csnGeneratorAdjust"); logger.error(LocalizableMessage.raw("Starting synchronization test : CSNGeneratorAdjust")); /* * Open a session to the replicationServer using the broker API. * This must use a different serverId to that of the directory server. */ final int serverId = 88; ReplicationBroker broker = openReplicationSession(baseDN, serverId, 100, replServerPort, 1000); consumeAllMessages(broker); // clean leftover messages from lostHeartbeatFailover() try { final long inTheFuture = System.currentTimeMillis() + (3600 * 1000); CSNGenerator gen = new CSNGenerator(serverId, inTheFuture); // Create and publish an update message to add an entry. AddMsg addMsg = addMsg(gen, user3Entry, user3UUID, baseUUID); broker.publish(addMsg); // Check that the entry has not been created in the directory server. assertNotNull(getEntry(user3Entry.getName(), 1000, true), "The entry has not been created"); // Modify the entry connection.processModify(modifyRequest(user3Entry.getName(), REPLACE, "telephonenumber", "01 02 45")); // See if the client has received the msg ReplicationMsg msg = broker.receive(); Assertions.assertThat(msg).isInstanceOf(ModifyMsg.class); ModifyMsg modMsg = (ModifyMsg) msg; assertTrue(modMsg.getCSN().getTimeSec()-addMsg.getCSN().getTimeSec()<=1, "The MOD timestamp should have been adjusted to the ADD one"); // Delete the entries to clean the database. broker.publish( new DeleteMsg(user3Entry.getName(), gen.newCSN(), user3UUID)); // Check that the delete operation has been applied. assertNull(getEntry(user3Entry.getName(), 10000, false), "The DELETE replication message was not replayed"); } finally { broker.stop(); } } /** * Consumes all the messages sent to this broker. This is useful at the start * of a test to avoid leftover messages from previous test runs. */ private void consumeAllMessages(ReplicationBroker broker) { final List msgs = new ArrayList<>(); try { while (true) { msgs.add(broker.receive()); } } catch (SocketTimeoutException expectedAtSomeStage) { // this is expected to happen when there will not be any more messages to // consume from the socket } if (!msgs.isEmpty()) { logger.error(LocalizableMessage.raw("Leftover messages from previous test runs " + msgs)); } } }