mirror of https://github.com/OpenIdentityPlatform/OpenDJ.git
opends/src/server/org/opends/server/synchronization/plugin/AttrInfo.java
@@ -37,7 +37,7 @@ * This classes is used to store historical information. * One object of this type is created for each attribute that was changed in * the entry. * It allows to record the last time a givene value was added, the last * It allows to record the last time a given value was added, the last * time a given value was deleted and the last time the whole attribute was * deleted. */ @@ -130,8 +130,14 @@ { if (this.valuesInfo != null) this.valuesInfo.clear(); deleteTime = CN; lastUpdateTime = CN; if (CN.newer(deleteTime)) { deleteTime = CN; } if (CN.newer(lastUpdateTime)) { lastUpdateTime = CN; } } /** @@ -144,7 +150,10 @@ ValueInfo info = new ValueInfo(val, null, CN); this.valuesInfo.remove(info); this.valuesInfo.add(info); lastUpdateTime = CN; if (CN.newer(lastUpdateTime)) { lastUpdateTime = CN; } } /** @@ -160,7 +169,10 @@ ValueInfo info = new ValueInfo(val, null, CN); this.valuesInfo.remove(info); this.valuesInfo.add(info); lastUpdateTime = CN; if (CN.newer(lastUpdateTime)) { lastUpdateTime = CN; } } } @@ -175,7 +187,10 @@ ValueInfo info = new ValueInfo(val, CN, null); this.valuesInfo.remove(info); valuesInfo.add(info); lastUpdateTime = CN; if (CN.newer(lastUpdateTime)) { lastUpdateTime = CN; } } /** @@ -192,7 +207,10 @@ ValueInfo info = new ValueInfo(val, CN, null); this.valuesInfo.remove(info); valuesInfo.add(info); lastUpdateTime = CN; if (CN.newer(lastUpdateTime)) { lastUpdateTime = CN; } } }
@@ -26,7 +26,6 @@ */ package org.opends.server.synchronization.plugin; import java.util.HashSet; import java.util.LinkedHashSet; import java.util.Set; @@ -40,7 +39,7 @@ /** * This Class is used to encode/decode hsitorical information * This Class is used to encode/decode historical information * from the String form to the internal usable form. * * @author Gilles Bellaton @@ -51,7 +50,7 @@ private String attrString; private AttributeValue attributeValue; private ChangeNumber cn; private Set<String> options = null; private LinkedHashSet<String> options; private HistKey histKey; private String stringValue; @@ -84,10 +83,10 @@ */ String[] token = strVal.split(":", 4); options = new LinkedHashSet<String>(); if (token[0].contains(";")) { String[] optionsToken = token[0].split(";"); options = new HashSet<String>(); int index = 1; while (index < optionsToken.length) { @@ -153,7 +152,7 @@ } /** * Get the options. * Get the options or an empty set if there are no options. * @return Returns the options. */ public Set<String> getOptions() @@ -186,7 +185,7 @@ */ public Modification generateMod() { Attribute attr = new Attribute(attrType); Attribute attr = new Attribute(attrType, attrString, options, null); Modification mod; if (histKey != HistKey.DELATTR) { opends/src/server/org/opends/server/synchronization/plugin/Historical.java
@@ -34,6 +34,7 @@ import java.util.Map; import java.util.Set; import java.util.TreeMap; import java.util.HashSet; import org.opends.server.core.AddOperation; import org.opends.server.core.DirectoryServer; @@ -141,10 +142,6 @@ Modification m = (Modification) modsIterator.next(); Attribute modAttr = m.getAttribute(); Set<String> options = modAttr.getOptions(); if (options.isEmpty()) { options = null; } AttributeType type = modAttr.getAttributeType(); AttrInfoWithOptions attrInfoWithOptions = attributesInfo.get(type); AttrInfo attrInfo = null; @@ -263,10 +260,6 @@ return; } Set<String> options = modAttr.getOptions(); if (options.isEmpty()) { options = null; } AttributeType type = modAttr.getAttributeType(); AttrInfoWithOptions attrInfoWithOptions = attributesInfo.get(type); AttrInfo attrInfo; @@ -477,7 +470,7 @@ */ private boolean hasConflict(AttrInfo info, ChangeNumber newChange) { // if I've already seen a change that is more recetn than the one // if I've already seen a change that is more recent than the one // that is currently being processed, then there is // a potential conflict if (ChangeNumber.compare(newChange, moreRecentChangenumber) <= 0) @@ -758,7 +751,7 @@ List<Attribute> hist = entry.getAttribute(historicalAttrType); Historical histObj = new Historical(); AttributeType lastAttrType = null; Set<String> lastOptions = null; Set<String> lastOptions = new HashSet<String>(); AttrInfo attrInfo = null; AttrInfoWithOptions attrInfoWithOptions = null;
@@ -1423,6 +1423,8 @@ else return true; } // TODO log a message for the repair tool. return true; } @@ -1506,7 +1508,7 @@ /* * This entry is the base dn of the backend. * It is quite weird that the operation result be NO_SUCH_OBJECT. * There is notthing more we can do except TODO log a * There is nothing more we can do except TODO log a * message for the repair tool to look at this problem. */ return true; opends/tests/unit-tests-testng/src/server/org/opends/server/synchronization/SynchronizationTestCase.java
@@ -28,25 +28,38 @@ import static org.testng.Assert.assertNotNull; import static org.testng.Assert.assertTrue; import static org.testng.Assert.fail; import java.net.SocketException; import java.util.ArrayList; import java.util.LinkedList; import java.util.NoSuchElementException; import java.util.List; import java.util.concurrent.locks.Lock; import org.opends.server.DirectoryServerTestCase; import org.opends.server.TestCaseUtils; import org.opends.server.schema.IntegerSyntax; import org.opends.server.config.ConfigEntry; import org.opends.server.config.ConfigException; import org.opends.server.core.DeleteOperation; import org.opends.server.core.DirectoryServer; import org.opends.server.protocols.internal.InternalClientConnection; import org.opends.server.protocols.internal.InternalSearchOperation; import org.opends.server.protocols.ldap.LDAPFilter; import org.opends.server.synchronization.common.ServerState; import org.opends.server.synchronization.plugin.ChangelogBroker; import org.opends.server.synchronization.plugin.MultimasterSynchronization; import org.opends.server.synchronization.plugin.PersistentServerState; import org.opends.server.types.DN; import org.opends.server.types.Entry; import org.opends.server.types.ByteStringFactory; import org.opends.server.types.SearchScope; import org.opends.server.types.SearchResultEntry; import org.opends.server.types.AttributeType; import org.opends.server.types.LockManager; import org.opends.server.types.Attribute; import org.opends.server.types.AttributeValue; import org.testng.annotations.AfterClass; import org.testng.annotations.Test; @@ -295,4 +308,84 @@ entryList.add(synchroServerEntry.getDN()); } /** * Retrieve the number of replayed updates for a given synchronization * domain from the monitor entry. * @return The number of replayed updates. * @throws Exception If an error occurs. */ protected long getReplayedUpdatesCount(DN syncDN) throws Exception { String monitorFilter = "(&(cn=synchronization*)(base-dn=" + syncDN + "))"; InternalSearchOperation op; op = connection.processSearch( ByteStringFactory.create("cn=monitor"), SearchScope.SINGLE_LEVEL, LDAPFilter.decode(monitorFilter)); SearchResultEntry entry = op.getSearchEntries().getFirst(); AttributeType attrType = DirectoryServer.getDefaultAttributeType("replayed-updates"); return entry.getAttributeValue(attrType, IntegerSyntax.DECODER).longValue(); } /** * Check that the entry with the given dn has the given valueString value * for the given attrTypeStr attribute type. */ protected boolean checkEntryHasAttribute(DN dn, String attrTypeStr, String valueString, int timeout, boolean hasAttribute) throws Exception { boolean found; int count = timeout/100; if (count<1) count=1; do { Entry newEntry; Lock lock = null; for (int j=0; j < 3; j++) { lock = LockManager.lockRead(dn); if (lock != null) { break; } } if (lock == null) { throw new Exception("could not lock entry " + dn); } try { newEntry = DirectoryServer.getEntry(dn); if (newEntry == null) fail("The entry " + dn + " has incorrectly been deleted from the database."); List<Attribute> tmpAttrList = newEntry.getAttribute(attrTypeStr); Attribute tmpAttr = tmpAttrList.get(0); AttributeType attrType = DirectoryServer.getAttributeType(attrTypeStr, true); found = tmpAttr.hasValue(new AttributeValue(attrType, valueString)); } finally { LockManager.unlock(dn, lock); } if (found != hasAttribute) Thread.sleep(100); } while ((--count > 0) && (found != hasAttribute)); return found; } } opends/tests/unit-tests-testng/src/server/org/opends/server/synchronization/UpdateOperationTest.java
@@ -38,7 +38,6 @@ import org.opends.server.TestCaseUtils; import org.opends.server.plugins.ShortCircuitPlugin; import org.opends.server.schema.DirectoryStringSyntax; import org.opends.server.schema.IntegerSyntax; import org.opends.server.synchronization.common.ChangeNumberGenerator; import org.opends.server.synchronization.plugin.ChangelogBroker; import org.opends.server.synchronization.protocol.AddMsg; @@ -55,8 +54,6 @@ import org.opends.server.core.Operation; import org.opends.server.protocols.asn1.ASN1OctetString; import org.opends.server.protocols.internal.InternalClientConnection; import org.opends.server.protocols.internal.InternalSearchOperation; import org.opends.server.protocols.ldap.LDAPFilter; import org.opends.server.protocols.ldap.LDAPAttribute; import org.opends.server.protocols.ldap.LDAPModification; import org.opends.server.types.*; @@ -117,9 +114,9 @@ + "objectClass: organizationalUnit\n" + "entryUUID: 11111111-1111-1111-1111-111111111111\n"; Entry entry; for (int i = 0; i < topEntries.length; i++) for (String entryStr : topEntries) { entry = TestCaseUtils.entryFromLdifString(topEntries[i]); entry = TestCaseUtils.entryFromLdifString(entryStr); AddOperation addOp = new AddOperation(connection, InternalClientConnection.nextOperationID(), InternalClientConnection .nextMessageID(), null, entry.getDN(), entry.getObjectClasses(), @@ -321,7 +318,7 @@ logError(ErrorLogCategory.SYNCHRONIZATION, ErrorLogSeverity.NOTICE, "Starting synchronization test : lostHeartbeatFailover" , 1); cleanEntries(); final DN baseDn = DN.decode("ou=People,dc=example,dc=com"); @@ -745,9 +742,9 @@ + "objectClass: organizationalUnit\n" + "entryUUID: 55555555-5555-5555-5555-555555555555\n"; Entry entry; for (int i = 0; i < topEntries.length; i++) for (String entryStr : topEntries) { entry = TestCaseUtils.entryFromLdifString(topEntries[i]); entry = TestCaseUtils.entryFromLdifString(entryStr); AddOperation addOp = new AddOperation(connection, InternalClientConnection.nextOperationID(), InternalClientConnection .nextMessageID(), null, entry.getDN(), entry.getObjectClasses(), @@ -785,9 +782,9 @@ "Entry not moved from ou=baseDn1,"+baseDn+" to ou=baseDn2,"+baseDn); // - add new parent entry 2 with baseDn1 String p2 = new String("dn: ou=baseDn1,"+baseDn+"\n" + "objectClass: top\n" + "objectClass: organizationalUnit\n" + "entryUUID: 66666666-6666-6666-6666-666666666666\n"); String p2 = "dn: ou=baseDn1,"+baseDn+"\n" + "objectClass: top\n" + "objectClass: organizationalUnit\n" + "entryUUID: 66666666-6666-6666-6666-666666666666\n"; entry = TestCaseUtils.entryFromLdifString(p2); AddOperation addOp = new AddOperation(connection, InternalClientConnection.nextOperationID(), InternalClientConnection @@ -1020,9 +1017,6 @@ } } /** * @return */ private List<Modification> generatemods(String attrName, String attrValue) { AttributeType attrType = @@ -1037,63 +1031,6 @@ } /** * Check that the entry with the given dn has the given valueString value * for the given attrTypeStr attribute type. */ private boolean checkEntryHasAttribute(DN dn, String attrTypeStr, String valueString, int timeout, boolean hasAttribute) throws Exception { boolean found; int count = timeout/100; if (count<1) count=1; do { Entry newEntry; Lock lock = null; for (int j=0; j < 3; j++) { lock = LockManager.lockRead(dn); if (lock != null) { break; } } if (lock == null) { throw new Exception("could not lock entry " + dn); } try { newEntry = DirectoryServer.getEntry(personWithUUIDEntry.getDN()); if (newEntry == null) fail("The entry " + personWithUUIDEntry.getDN() + " has incorrectly been deleted from the database."); List<Attribute> tmpAttrList = newEntry.getAttribute(attrTypeStr); Attribute tmpAttr = tmpAttrList.get(0); AttributeType attrType = DirectoryServer.getAttributeType(attrTypeStr, true); found = tmpAttr.hasValue(new AttributeValue(attrType, valueString)); } finally { LockManager.unlock(dn, lock); } if (found != hasAttribute) Thread.sleep(100); } while ((--count > 0) && (found != hasAttribute)); return found; } /** * Get the entryUUID for a given DN. * * @throws Exception if the entry does not exist or does not have @@ -1272,7 +1209,7 @@ assertEquals(addOp.getResultCode(), ResultCode.SUCCESS); entryList.add(tmp.getDN()); long initialCount = getReplayedUpdatesCount(); long initialCount = getReplayedUpdatesCount(baseDn); // Get the UUID of the test entry. Entry resultEntry = getEntry(tmp.getDN(), 1, true); @@ -1295,7 +1232,7 @@ // Wait for the operation to be replayed. long endTime = System.currentTimeMillis() + 5000; while (getReplayedUpdatesCount() == initialCount && while (getReplayedUpdatesCount(baseDn) == initialCount && System.currentTimeMillis() < endTime) { Thread.sleep(100); @@ -1309,7 +1246,7 @@ // If the synchronization replay loop was detected and broken then the // counter will still be updated even though the replay was unsuccessful. if (getReplayedUpdatesCount() == initialCount) if (getReplayedUpdatesCount(baseDn) == initialCount) { fail("Synchronization operation was not replayed"); } @@ -1321,28 +1258,6 @@ } /** * Retrieve the number of replayed updates from the monitor entry. * @return The number of replayed updates. * @throws Exception If an error occurs. */ private long getReplayedUpdatesCount() throws Exception { String monitorFilter = "(&(cn=synchronization*)(base-dn=ou=People,dc=example,dc=com))"; InternalSearchOperation op; op = connection.processSearch( ByteStringFactory.create("cn=monitor"), SearchScope.SINGLE_LEVEL, LDAPFilter.decode(monitorFilter)); SearchResultEntry entry = op.getSearchEntries().getFirst(); AttributeType attrType = DirectoryServer.getDefaultAttributeType("replayed-updates"); return entry.getAttributeValue(attrType, IntegerSyntax.DECODER).longValue(); } /** * Enable or disable the receive status of a synchronization provider. * * @param syncConfigDN The DN of the synchronization provider configuration opends/tests/unit-tests-testng/src/server/org/opends/server/synchronization/plugin/HistoricalTest.java
@@ -28,18 +28,24 @@ package org.opends.server.synchronization.plugin; import org.opends.server.synchronization.SynchronizationTestCase; import org.opends.server.synchronization.protocol.ModifyMsg; import org.opends.server.synchronization.common.ChangeNumber; import org.opends.server.TestCaseUtils; import org.opends.server.protocols.internal.InternalClientConnection; import org.opends.server.tools.LDAPModify; import org.opends.server.types.DN; import org.opends.server.types.Entry; import org.opends.server.types.Attribute; import org.opends.server.types.Modification; import org.opends.server.types.ModificationType; import org.opends.server.types.AttributeType; import org.opends.server.core.DirectoryServer; import org.testng.annotations.Test; import org.testng.annotations.BeforeClass; import static org.testng.Assert.assertEquals; import java.util.List; import java.util.ArrayList; /** * Tests the Historical class. @@ -122,10 +128,12 @@ "sn: Amar", "givenName: Aaccf", "userPassword: password", "description: Initial description" "description: Initial description", "displayName: 1" ); // Modify the test entry to give it some history. // Test both single and multi-valued attributes. String path = TestCaseUtils.createTempFile( "dn: uid=user.1,o=test", @@ -136,6 +144,12 @@ "-", "replace: description", "description: replaced description", "-", "add: displayName", "displayName: 2", "-", "delete: displayName", "displayName: 1", "-" ); @@ -150,17 +164,172 @@ assertEquals(LDAPModify.mainModify(args, false, null, System.err), 0); args[9] = TestCaseUtils.createTempFile( "dn: uid=user.1,o=test", "changetype: modify", "replace: displayName", "displayName: 2", "-" ); assertEquals(LDAPModify.mainModify(args, false, null, System.err), 0); // Read the entry back to get its history operational attribute. DN dn = DN.decode("uid=user.1,o=test"); Entry entry = DirectoryServer.getEntry(dn); List<Attribute> attrs = entry.getAttribute(Historical.historicalAttrType); Attribute before = attrs.get(0); // Check that encoding and decoding preserves the history information. Historical hist = Historical.load(entry); Attribute after = hist.encode(); List<Attribute> attrs = entry.getAttribute(Historical.historicalAttrType); Attribute before = attrs.get(0); assertEquals(after, before); } /** * The scenario for this test case is that two modify operations occur at * two different servers at nearly the same time, each operation adding a * different value for a single-valued attribute. Synchronization then * replays the operations and we expect the conflict to be resolved on both * servers by keeping whichever value was actually added first. * For the unit test, we employ a single directory server. We use the * broker API to simulate the ordering that would happen on the first server * on one entry, and the reverse ordering that would happen on the * second server on a different entry. Confused yet? * @throws Exception If the test fails. */ @Test(enabled=false, groups="slow") public void conflictSingleValue() throws Exception { final DN dn1 = DN.decode("cn=test1,o=test"); final DN dn2 = DN.decode("cn=test2,o=test"); final DN baseDn = DN.decode("o=test"); final AttributeType attrType = DirectoryServer.getAttributeType("displayname"); final AttributeType entryuuidType = DirectoryServer.getAttributeType("entryuuid"); /* * Open a session to the changelog server using the broker API. * This must use a different serverId to that of the directory server. */ ChangelogBroker broker = openChangelogSession(baseDn, (short) 2, 100, 8989, 1000, true); // Clear the backend. TestCaseUtils.initializeTestBackend(true); // Add the first test entry. TestCaseUtils.addEntry( "dn: cn=test1,o=test", "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<Attribute> attrs = entry.getAttribute(entryuuidType); String entryuuid = attrs.get(0).getValues().iterator().next().getStringValue(); // Add the second test entry. TestCaseUtils.addEntry( "dn: cn=test2,o=test", "objectClass: top", "objectClass: person", "objectClass: organizationalPerson", "objectClass: inetOrgPerson", "cn: test2", "sn: test", "description: Description" ); // Read the entry back to get its UUID. entry = DirectoryServer.getEntry(dn2); attrs = entry.getAttribute(entryuuidType); String entryuuid2 = attrs.get(0).getValues().iterator().next().getStringValue(); // A change on a first server. ChangeNumber t1 = new ChangeNumber(1, (short) 0, (short) 3); // A change on a second server. ChangeNumber t2 = new ChangeNumber(2, (short) 0, (short) 4); // Simulate the ordering t1:add:A followed by t2:add:B that would // happen on one server. // Replay an add of a value A at time t1 on a first server. Attribute attr = new Attribute(attrType.getNormalizedPrimaryName(), "A"); Modification mod = new Modification(ModificationType.ADD, attr); publishModify(broker, t1, dn1, entryuuid, mod); // It would be nice to avoid these sleeps. // We need to preserve the replay order but the order could be changed // due to the multi-threaded nature of the synchronization replay. // Putting a sentinel value in the modification is not foolproof since // the operation might not get replayed at all. Thread.sleep(2000); // Replay an add of a value B at time t2 on a second server. attr = new Attribute(attrType.getNormalizedPrimaryName(), "B"); mod = new Modification(ModificationType.ADD, attr); publishModify(broker, t2, dn1, entryuuid, mod); Thread.sleep(2000); // Simulate the reverse ordering t2:add:B followed by t1:add:A that // would happen on the other server. t1 = new ChangeNumber(3, (short) 0, (short) 3); t2 = new ChangeNumber(4, (short) 0, (short) 4); // Replay an add of a value B at time t2 on a second server. attr = new Attribute(attrType.getNormalizedPrimaryName(), "B"); mod = new Modification(ModificationType.ADD, attr); publishModify(broker, t2, dn2, entryuuid2, mod); Thread.sleep(2000); // Replay an add of a value A at time t1 on a first server. attr = new Attribute(attrType.getNormalizedPrimaryName(), "A"); mod = new Modification(ModificationType.ADD, attr); publishModify(broker, t1, dn2, entryuuid2, mod); Thread.sleep(2000); // Read the first entry to see how the conflict was resolved. entry = DirectoryServer.getEntry(dn1); attrs = entry.getAttribute(attrType); String attrValue1 = attrs.get(0).getValues().iterator().next().getStringValue(); // Read the second entry to see how the conflict was resolved. entry = DirectoryServer.getEntry(dn2); attrs = entry.getAttribute(attrType); String attrValue2 = attrs.get(0).getValues().iterator().next().getStringValue(); // The two values should be the first value added. assertEquals(attrValue1, "A"); assertEquals(attrValue2, "A"); } private static void publishModify(ChangelogBroker broker, ChangeNumber changeNum, DN dn, String entryuuid, Modification mod) { List<Modification> mods = new ArrayList<Modification>(1); mods.add(mod); ModifyMsg modMsg = new ModifyMsg(changeNum, dn, mods, entryuuid); broker.publish(modMsg); } } opends/tests/unit-tests-testng/src/server/org/opends/server/synchronization/plugin/ModifyConflictTest.java
@@ -74,8 +74,7 @@ /** * Test that conflict between a modify-replace and modify-add for * multi-valued attributes@DataProvider(name = "ackMsg") are handled * correctly. * multi-valued attributes are handled correctly. */ @Test() public void replaceAndAdd() throws Exception
