Fix for issue 3402 : Replication conflict: fail to resolve double mod_rdn o...

gbellato

16.06.2009 7bb5b9a55a8d68f9622ca3ae6bb22b889b0a6a3f

 opends/src/server/org/opends/server/replication/plugin/LDAPReplicationDomain.java

@@ -59,7 +59,7 @@
import java.util.NoSuchElementException;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.zip.CheckedOutputStream;
@@ -178,7 +178,7 @@

  // The update to replay message queue where the listener thread is going to
  // push incoming update messages.
  private final LinkedBlockingQueue<UpdateToReplay> updateToReplayQueue;
  private final BlockingQueue<UpdateToReplay> updateToReplayQueue;
  private final AtomicInteger numResolvedNamingConflicts = new AtomicInteger();
  private final AtomicInteger numResolvedModifyConflicts = new AtomicInteger();
  private final AtomicInteger numUnresolvedNamingConflicts =
@@ -301,7 +301,7 @@
   * @throws ConfigException In case of invalid configuration.
   */
  public LDAPReplicationDomain(ReplicationDomainCfg configuration,
    LinkedBlockingQueue<UpdateToReplay> updateToReplayQueue)
    BlockingQueue<UpdateToReplay> updateToReplayQueue)
    throws ConfigException
  {
    super(configuration.getBaseDN().toNormalizedString(),
@@ -703,6 +703,16 @@
            ResultCode.NO_SUCH_OBJECT, null);
        }
      }
      /*
       * If the object has been renamed more recently than this
       * operation, cancel the operation.
       */
      Historical hist = Historical.load(modifyDNOperation.getOriginalEntry());
      if (hist.AddedOrRenamedAfter(ctx.getChangeNumber()))
      {
        return new SynchronizationProviderResult.StopProcessing(
            ResultCode.SUCCESS, null);
      }
    }
    else
    {

 opends/src/server/org/opends/server/replication/plugin/MultimasterReplication.java

@@ -22,7 +22,7 @@
 * CDDL HEADER END
 *
 *
 *      Copyright 2006-2008 Sun Microsystems, Inc.
 *      Copyright 2006-2009 Sun Microsystems, Inc.
 */
package org.opends.server.replication.plugin;

@@ -33,6 +33,7 @@
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.opends.messages.Message;
@@ -207,6 +208,30 @@
  }

  /**
   * Creates a new domain from its configEntry, do the
   * necessary initialization and starts it so that it is
   * fully operational when this method returns.
   *
   * @param configuration The entry with the configuration of this domain.
   * @param queue         The BlockingQueue that this domain will use.
   *
   * @return              The domain created.
   *
   * @throws ConfigException When the configuration is not valid.
   */
  public static LDAPReplicationDomain createNewDomain(
      ReplicationDomainCfg configuration,
      BlockingQueue<UpdateToReplay> queue)
      throws ConfigException
  {
    LDAPReplicationDomain domain;
    domain = new LDAPReplicationDomain(configuration, queue);

    domains.put(domain.getBaseDN(), domain);
    return domain;
  }

  /**
   * Deletes a domain.
   * @param dn : the base DN of the domain to delete.
   */

 opends/tests/unit-tests-testng/src/server/org/opends/server/replication/ReplicationTestCase.java

@@ -1095,4 +1095,63 @@
      fail("addEntries Exception:"+ e.getMessage() + " " + stackTraceToSingleLineString(e));
    }
  }

  /**
   *  Get the entryUUID for a given DN.
   *
   * @throws Exception if the entry does not exist or does not have
   *                   an entryUUID.
   */
  protected String getEntryUUID(DN dn) throws Exception
  {
    Entry newEntry;
    int count = 10;
    if (count<1)
      count=1;
    String found = null;
    while ((count> 0) && (found == null))
    {
      Thread.sleep(100);

      Lock lock = null;
      for (int i=0; i < 3; i++)
      {
        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)
        {
          List<Attribute> tmpAttrList = newEntry.getAttribute("entryuuid");
          Attribute tmpAttr = tmpAttrList.get(0);

          for (AttributeValue val : tmpAttr)
          {
            found = val.getValue().toString();
            break;
          }
        }
      }
      finally
      {
        LockManager.unlock(dn, lock);
      }
      count --;
    }
    if (found == null)
      throw new Exception("Entry: " + dn + " Could not be found.");
    return found;
  }
}

 opends/tests/unit-tests-testng/src/server/org/opends/server/replication/UpdateOperationTest.java

@@ -590,6 +590,7 @@
    broker.stop();
  }


  /**
   * Tests the naming conflict resolution code.
   * In this test, the local server act both as an LDAP server and
@@ -1500,64 +1501,6 @@
    }
  }

  /**
   *  Get the entryUUID for a given DN.
   *
   * @throws Exception if the entry does not exist or does not have
   *                   an entryUUID.
   */
  private String getEntryUUID(DN dn) throws Exception
  {
    Entry newEntry;
    int count = 10;
    if (count<1)
      count=1;
    String found = null;
    while ((count> 0) && (found == null))
    {
      Thread.sleep(100);

      Lock lock = null;
      for (int i=0; i < 3; i++)
      {
        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)
        {
          List<Attribute> tmpAttrList = newEntry.getAttribute("entryuuid");
          Attribute tmpAttr = tmpAttrList.get(0);

          for (AttributeValue val : tmpAttr)
          {
            found = val.getValue().toString();
            break;
          }
        }
      }
      finally
      {
        LockManager.unlock(dn, lock);
      }
      count --;
    }
    if (found == null)
      throw new Exception("Entry: " + dn + " Could not be found.");
    return found;
  }

  /**
   * Test case for

 opends/tests/unit-tests-testng/src/server/org/opends/server/replication/plugin/AssuredReplicationPluginTest.java

@@ -1300,65 +1300,6 @@
  }

  /**
   *  Get the entryUUID for a given DN.
   *
   * @throws Exception if the entry does not exist or does not have
   *                   an entryUUID.
   */
  private String getEntryUUID(DN dn) throws Exception
  {
    Entry newEntry;
    int count = 10;
    if (count<1)
      count=1;
    String found = null;
    while ((count> 0) && (found == null))
    {
      Thread.sleep(100);

      Lock lock = null;
      for (int i=0; i < 3; i++)
      {
        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)
        {
          List<Attribute> tmpAttrList = newEntry.getAttribute("entryuuid");
          Attribute tmpAttr = tmpAttrList.get(0);

          for (AttributeValue val : tmpAttr)
          {
            found = val.toString();
            break;
          }
        }
      }
      finally
      {
        LockManager.unlock(dn, lock);
      }
      count --;
    }
    if (found == null)
      throw new Exception("Entry: " + dn + " Could not be found.");
    return found;
  }

  /**
   * Tests that a DS receiving an update from a RS in safe read mode effectively
   * sends an ack back (with or without error)
   */

 opends/tests/unit-tests-testng/src/server/org/opends/server/replication/plugin/NamingConflictTest.java

New file
@@ -0,0 +1,134 @@
/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License, Version 1.0 only
 * (the "License").  You may not use this file except in compliance
 * with the License.
 *
 * You can obtain a copy of the license at
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE
 * or https://OpenDS.dev.java.net/OpenDS.LICENSE.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE.  If applicable,
 * add the following below this CDDL HEADER, with the fields enclosed
 * by brackets "[]" replaced with your own identifying information:
 *      Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 *
 *
 *      Copyright 2009 Sun Microsystems, Inc.
 */
package org.opends.server.replication.plugin;

import static org.opends.server.TestCaseUtils.TEST_ROOT_DN_STRING;

import java.util.TreeSet;

import org.opends.server.TestCaseUtils;
import org.opends.server.admin.std.meta.ReplicationDomainCfgDefn.IsolationPolicy;
import org.opends.server.core.DirectoryServer;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.common.ChangeNumber;
import org.opends.server.replication.common.ChangeNumberGenerator;
import org.opends.server.replication.protocol.ModifyDNMsg;
import org.opends.server.types.DN;
import org.opends.server.types.Entry;
import org.testng.annotations.Test;

import static org.testng.Assert.*;


/**
 * Test the naming conflict resolution code.
 */
public class NamingConflictTest extends ReplicationTestCase
{
  /**
   * Test for issue 3402 : test, that a modrdn that is older than an other
   * modrdn but that is applied later is ignored.
   *
   * 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 simultaneousModrdnConflict() throws Exception
  {
    TestCaseUtils.initializeTestBackend(true);

    final DN baseDn = DN.decode(TEST_ROOT_DN_STRING);

    TestSynchronousReplayQueue queue = new TestSynchronousReplayQueue();
    DomainFakeCfg conf = new DomainFakeCfg(baseDn, 1, new TreeSet<String>());
    conf.setIsolationPolicy(IsolationPolicy.ACCEPT_ALL_UPDATES);

    LDAPReplicationDomain domain =
      MultimasterReplication.createNewDomain(conf, queue);

    /*
     * Create a Change number generator to generate new ChangeNumbers
     * when we need to send operations messages to the replicationServer.
     */
    ChangeNumberGenerator gen = new ChangeNumberGenerator((short) 201, 0);

    String parentUUID = getEntryUUID(DN.decode(TEST_ROOT_DN_STRING));

    Entry entry = TestCaseUtils.entryFromLdifString(
        "dn: cn=simultaneousModrdnConflict, "+ TEST_ROOT_DN_STRING + "\n"
        + "objectClass: top\n" + "objectClass: person\n"
        + "objectClass: organizationalPerson\n"
        + "objectClass: inetOrgPerson\n" + "uid: user.1\n"
        + "homePhone: 951-245-7634\n"
        + "description: This is the description for Aaccf Amar.\n" + "st: NC\n"
        + "mobile: 027-085-0537\n"
        + "postalAddress: Aaccf Amar$17984 Thirteenth Street"
        + "$Rockford, NC  85762\n" + "mail: user.1@example.com\n"
        + "cn: Aaccf Amar\n" + "l: Rockford\n" + "pager: 508-763-4246\n"
        + "street: 17984 Thirteenth Street\n"
        + "telephoneNumber: 216-564-6748\n" + "employeeNumber: 1\n"
        + "sn: Amar\n" + "givenName: Aaccf\n" + "postalCode: 85762\n"
        + "userPassword: password\n" + "initials: AA\n");

    TestCaseUtils.addEntry(entry);
    String entryUUID = getEntryUUID(entry.getDN());

    // generate two consecutive ChangeNumber that will be used in backward order
    ChangeNumber cn1 = gen.newChangeNumber();
    ChangeNumber cn2 = gen.newChangeNumber();

    ModifyDNMsg  modDnMsg = new ModifyDNMsg(
        entry.getDN().toNormalizedString(), cn2,
        entryUUID, parentUUID, false,
        TEST_ROOT_DN_STRING,
        "uid=simultaneous2");

    domain.processUpdate(modDnMsg);
    domain.replay(queue.take().getUpdateMessage());

    // This MODIFY DN uses an older DN and should therefore be cancelled
    // at replay time.
    modDnMsg = new ModifyDNMsg(
        entry.getDN().toNormalizedString(), cn1,
        entryUUID, parentUUID, false,
        TEST_ROOT_DN_STRING,
        "uid=simulatneouswrong");

    domain.processUpdate(modDnMsg);
    domain.replay(queue.take().getUpdateMessage());

    assertFalse(DirectoryServer.entryExists(entry.getDN()),
        "The modDN conflict was not resolved as expected.");

    MultimasterReplication.deleteDomain(baseDn);
  }
}

 opends/tests/unit-tests-testng/src/server/org/opends/server/replication/plugin/TestSynchronousReplayQueue.java

New file
@@ -0,0 +1,223 @@
/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License, Version 1.0 only
 * (the "License").  You may not use this file except in compliance
 * with the License.
 *
 * You can obtain a copy of the license at
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE
 * or https://OpenDS.dev.java.net/OpenDS.LICENSE.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE.  If applicable,
 * add the following below this CDDL HEADER, with the fields enclosed
 * by brackets "[]" replaced with your own identifying information:
 *      Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 *
 *
 *      Copyright 2009 Sun Microsystems, Inc.
 */
package org.opends.server.replication.plugin;

import java.util.Collection;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;


/**
 * A very partial implementation of a Blocking queue that should
 * be used for test purpose only.
 *
 * Only the offer and take methods are implemented.
 *
 */
public class TestSynchronousReplayQueue implements BlockingQueue<UpdateToReplay>
{
  LinkedList<UpdateToReplay> list = new LinkedList<UpdateToReplay>();

  @Override
  public boolean add(UpdateToReplay e)
  {
    // TODO Auto-generated method stub
    return false;
  }

  @Override
  public boolean contains(Object o)
  {
    // TODO Auto-generated method stub
    return false;
  }

  @Override
  public int drainTo(Collection<? super UpdateToReplay> c)
  {
    // TODO Auto-generated method stub
    return 0;
  }

  @Override
  public int drainTo(Collection<? super UpdateToReplay> c, int maxElements)
  {
    // TODO Auto-generated method stub
    return 0;
  }

  @Override
  public boolean offer(UpdateToReplay e)
  {
    list.add(e);
    return true;
  }

  @Override
  public boolean offer(UpdateToReplay e, long timeout, TimeUnit unit)
      throws InterruptedException
  {
    // TODO Auto-generated method stub
    return false;
  }

  @Override
  public UpdateToReplay poll(long timeout, TimeUnit unit)
      throws InterruptedException
  {
    // TODO Auto-generated method stub
    return null;
  }

  @Override
  public void put(UpdateToReplay e) throws InterruptedException
  {
    // TODO Auto-generated method stub

  }

  @Override
  public int remainingCapacity()
  {
    // TODO Auto-generated method stub
    return 0;
  }

  @Override
  public boolean remove(Object o)
  {
    // TODO Auto-generated method stub
    return false;
  }

  @Override
  public UpdateToReplay take() throws InterruptedException
  {
    return list.pop();
  }

  @Override
  public UpdateToReplay element()
  {
    // TODO Auto-generated method stub
    return null;
  }

  @Override
  public UpdateToReplay peek()
  {
    // TODO Auto-generated method stub
    return null;
  }

  @Override
  public UpdateToReplay poll()
  {
    // TODO Auto-generated method stub
    return null;
  }

  @Override
  public UpdateToReplay remove()
  {
    // TODO Auto-generated method stub
    return null;
  }

  @Override
  public boolean addAll(Collection<? extends UpdateToReplay> c)
  {
    // TODO Auto-generated method stub
    return false;
  }

  @Override
  public void clear()
  {
    // TODO Auto-generated method stub

  }

  @Override
  public boolean containsAll(Collection<?> c)
  {
    // TODO Auto-generated method stub
    return false;
  }

  @Override
  public boolean isEmpty()
  {
    // TODO Auto-generated method stub
    return false;
  }

  @Override
  public Iterator<UpdateToReplay> iterator()
  {
    // TODO Auto-generated method stub
    return null;
  }

  @Override
  public boolean removeAll(Collection<?> c)
  {
    // TODO Auto-generated method stub
    return false;
  }

  @Override
  public boolean retainAll(Collection<?> c)
  {
    // TODO Auto-generated method stub
    return false;
  }

  @Override
  public int size()
  {
    // TODO Auto-generated method stub
    return 0;
  }

  @Override
  public Object[] toArray()
  {
    // TODO Auto-generated method stub
    return null;
  }

  @Override
  public <T> T[] toArray(T[] a)
  {
    // TODO Auto-generated method stub
    return null;
  }

}